KafkaTemplate是事务发送同步还是异步？

Question

我试图找出如何在事务上下文中正确处理原子级的对Kafka的多次写入。在此场景中，事务不是由kafka消息侦听器启动的，而是通过@Transactional注释以编程方式启动的，请参见下面的代码段。

我使用的是spring boot 2.4.2和spring kafka 2.6.5。

KafkaProducer文档指出，在事务上下文中，不需要调用。get（），因为它最终会在尝试提交事务时引发异常。此外，KafkaTemplate呼吁。KafkaProducer返回的Future上的get（），因此它看起来是同步的。

  @PostMapping("/ingest/{topic}")
  public ResponseEntity ingest(@PathVariable(value = "topic") String topic, @RequestBody String numbersString) {

    List<String> numbers = Arrays.stream(numbersString.split(",")).collect(Collectors.toList());
    kafkaWriterService.writeMany(numbers,topic); 
    return ResponseEntity.ok().build();
  }

  @Transactional
  @Service
  class KafkaWriterService {

    @Autowired
    KafkaTemplate<String, String> kafkaTemplate;

    public void writeMany(List<String> messages, String topic) {

      for (String message : messages) {
        kafkaTemplate.send(message, topic, topic);
      }
    }
  }

所以据我所知下面的KafkaTemplate方法

protected ListenableFuture<SendResult<K, V>> doSend(ProducerRecord<K, V> producerRecord)

等待生产者完成发送，然后返回另一个ListenableFutre，此接口是异步的。

那么这是同步的，因为我们处于事务上下文中，还是我应该等待kafkaTemplate返回的所有ListenableFutures结束？我的意思是考虑到我需要以同步方式回复调用者。

谢谢，谢谢

Answer 1

不；它只调用get（）如果未来立即由发送完成...

        Future<RecordMetadata> sendFuture =
                producer.send(producerRecord, buildCallback(producerRecord, producer, future, sample));
        // May be an immediate failure
        if (sendFuture.isDone()) {
            try {
                sendFuture.get();
            }
...

在实际执行发送之前，客户端中会出现某些错误。

看见https://github.com/spring-projects/spring-kafka/issues/1437