Spring Kafka 事务导致生产者每条消息的偏移量增加了 2

Question

我在使用 Spring(boot) Kafka 的微服务中有一个消费-转换-生产工作流程。我需要实现 Kafka 事务提供的一次性语义。下面是代码片段：

配置

@Bean
public ProducerFactory<String, String> producerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
    props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 1024 * 1024);
    DefaultKafkaProducerFactory<String, String> defaultKafkaProducerFactory = new DefaultKafkaProducerFactory<>(props);
    defaultKafkaProducerFactory.setTransactionIdPrefix("kafka-trx-");
    return defaultKafkaProducerFactory;
}

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

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

@Bean
public KafkaTransactionManager<String, String> kafkaTransactionManager() {
    return new KafkaTransactionManager<>(producerFactory());
}

@Bean
@Qualifier("chainedKafkaTransactionManager")
public ChainedKafkaTransactionManager<String, Object> chainedKafkaTransactionManager(KafkaTransactionManager<String, String> kafkaTransactionManager) {
    return new ChainedKafkaTransactionManager<>(kafkaTransactionManager);
}

@Bean
public ConcurrentKafkaListenerContainerFactory<?, ?> concurrentKafkaListenerContainerFactory(ChainedKafkaTransactionManager<String, Object> chainedKafkaTransactionManager) {
    ConcurrentKafkaListenerContainerFactory<String, String> concurrentKafkaListenerContainerFactory = new ConcurrentKafkaListenerContainerFactory<>();
    concurrentKafkaListenerContainerFactory.setConsumerFactory(consumerFactory());
    concurrentKafkaListenerContainerFactory.setBatchListener(true);
    concurrentKafkaListenerContainerFactory.setConcurrency(nexusConsumerConcurrency);
    //concurrentKafkaListenerContainerFactory.setReplyTemplate(kafkaTemplate());
    concurrentKafkaListenerContainerFactory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.BATCH);
    concurrentKafkaListenerContainerFactory.getContainerProperties().setTransactionManager(chainedKafkaTransactionManager);
    return concurrentKafkaListenerContainerFactory;
}

聆听者

@KafkaListener(topics = "${kafka.xxx.consumerTopic}", groupId = "${kafka.xxx.consumerGroup}", containerFactory = "concurrentKafkaListenerContainerFactory")
public void listen(@Payload List<String> msgs, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions, @Header(KafkaHeaders.OFFSET) List<Integer> offsets) {

    int i = -1;
    for (String msg : msgs) {
        ++i;
        LOGGER.debug("partition={}; offset={}; msg={}", partitions.get(i), offsets.get(i), msg);
        String json = transform(msg);
        kafkaTemplate.executeInTransaction(kt -> kt.send(producerTopic, json));
    }
}

然而在产品环境中，我遇到了一个奇怪的问题。生产者发送的每条消息的偏移量都会增加 2，并且消费者不会提交消费偏移量。

消费者对主题 1 的抵消

Topic1 消费者细节

生成主题2

但是，生产者发送的消息数与消费者发送的消息数相同。生产者下游可以持续接收来自topic2的消息。日志中没有发现任何错误或异常。

我想知道为什么消费-转换-生产工作流程看起来没问题（也保证了恰好一次语义），但消费的偏移量未提交，并且生成的消息偏移量增量是每条消息的 2 而不是 1。

如何修复它？谢谢！

Answer 1

这就是它的设计方式。Kafka 日志是不可变的，因此在事务结束时使用额外的“槽”来指示事务是已提交还是已回滚。这允许具有隔离级别的消费者read_committed跳过回滚的事务。

如果您在一笔事务中发布 10 条记录，您将看到偏移量增加 11。如果您只发布 1 条记录，则会增加 2。

如果您希望发布参与消费者启动的事务（仅一次），则不应使用executeInTransaction; 这将开始一个新的交易。

/**
 * Execute some arbitrary operation(s) on the operations and return the result.
 * The operations are invoked within a local transaction and do not participate
 * in a global transaction (if present).
 * @param callback the callback.
 * @param <T> the result type.
 * @return the result.
 * @since 1.1
 */
<T> T executeInTransaction(OperationsCallback<K, V, T> callback);

我不明白为什么消费者偏移量不会仍然发送到消费者启动的交易。您应该打开调试日志记录以查看发生了什么（如果修复模板代码后仍然发生）。

编辑

当监听器退出时，消耗的偏移量（+1）由监听器容器发送到事务；打开提交日志记录，你会看到它......

/**
 * Execute some arbitrary operation(s) on the operations and return the result.
 * The operations are invoked within a local transaction and do not participate
 * in a global transaction (if present).
 * @param callback the callback.
 * @param <T> the result type.
 * @return the result.
 * @since 1.1
 */
<T> T executeInTransaction(OperationsCallback<K, V, T> callback);

和

spring.kafka.producer.transaction-id-prefix=tx-
spring.kafka.consumer.properties.isolation.level=read_committed
spring.kafka.consumer.auto-offset-reset=earliest

和

foo0@56
2019-12-04 10:07:18.551  INFO 55430 --- [      foo-0-C-1] essageListenerContainer$ListenerConsumer : Sending offsets to transaction: {so59152915-1-0=OffsetAndMetadata{offset=57, leaderEpoch=null, metadata=''}}
foo1@57
FOO0
2019-12-04 10:07:18.558  INFO 55430 --- [      bar-0-C-1] essageListenerContainer$ListenerConsumer : Sending offsets to transaction: {so59152915-2-0=OffsetAndMetadata{offset=63, leaderEpoch=null, metadata=''}}
2019-12-04 10:07:20.562  INFO 55430 --- [      foo-0-C-1] essageListenerContainer$ListenerConsumer : Sending offsets to transaction: {so59152915-1-0=OffsetAndMetadata{offset=58, leaderEpoch=null, metadata=''}}
foo2@58