使用 Exactly Once 处理的 Kafka 的 Spring Boot

Question

我正在使用 kafka 设置一个 Spring Boot 应用程序。如何实现“一次交货保证”。

我已阅读并尝试按照“https://www.baeldung.com/kafka-exactly-once”实施但卡在 producer.initTransactions();

// NotificationKafkaConsumerConfig.java

@Bean
public ConsumerFactory < String, String > consumerFactory() {
    Map < String, Object > props = new HashMap < > ();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "notification_group");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
    return new DefaultKafkaConsumerFactory < > (props);
}

@Bean
public ConcurrentKafkaListenerContainerFactory < String, String > kafkaListenerContainerFactory() {

    ConcurrentKafkaListenerContainerFactory < String, String > factory = new ConcurrentKafkaListenerContainerFactory < > ();
    factory.setConsumerFactory(consumerFactory());
    return factory;
}
// NotificationKafkaProducerConfig
@Bean
public ProducerFactory < String, String > producerFactory() {

    Map < String, Object > configProps = new HashMap < > ();
    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    return new DefaultKafkaProducerFactory < > (configProps);
}

@Bean
public KafkaTemplate < String, String > kafkaTemplate() {
    return new KafkaTemplate < > (producerFactory());
}

// NotificationKafkaListner
@KafkaListener(topics = "notification", groupId = "notification_group")
public void listen(String message) {

    try {} catch (Exception e) {}
}

// NotificationController
@SuppressWarnings({
    "rawtypes",
    "unchecked"
})
@PostMapping(path = "/send")
public ResponseEntity << ? > send(@Valid @RequestBody NotificationRequest notificationReq) {
    log.debug("NotificationController: invoked to send notification");
    kafkaTemplate.send(notificationTopic, message);
}

如何进行交易

Answer 1

考虑使用KafkaTransactionManager:

 * {@link org.springframework.transaction.PlatformTransactionManager} implementation for a
 * single Kafka {@link ProducerFactory}. Binds a Kafka producer from the specified
 * ProducerFactory to the thread, potentially allowing for one thread-bound producer per
 * ProducerFactory.

在参考手册中查看更多信息：https://docs.spring.io/spring-kafka/docs/current/reference/html/#transactional-listener-container-and-exactly-once-processing