《Kafka0.11》中sendOffsetsToTransaction的含义

Question

新的Kafka版本（0.11）只支持一次语义。

https://cwiki.apache.org/confluence/display/KAFKA/KIP-98-精确一次交付和事务消息传递

我有一个用java编写的kafka事务代码制作程序，如下所示。

producer.initTransactions();
    try {
        producer.beginTransaction();
        for (ProducerRecord<String, String> record : payload) {
            producer.send(record);
        }

        Map<TopicPartition, OffsetAndMetadata> groupCommit = new HashMap<TopicPartition, OffsetAndMetadata>() {
            {
                put(new TopicPartition(TOPIC, 0), new OffsetAndMetadata(42L, null));
            }
        };
        producer.sendOffsetsToTransaction(groupCommit, "groupId");
        producer.commitTransaction();
    } catch (ProducerFencedException e) {
        producer.close();
    } catch (KafkaException e) {
        producer.abortTransaction();
    }

我不太确定如何使用sendOffsetsToTransaction及其预期用例。AFAIK，消费者组是消费者端的多线程读取功能。

javadoc说

“将已消耗的偏移量列表发送给使用者组协调员，并将这些偏移量标记为当前事务的一部分。只有事务成功提交时，这些偏移量才会被视为已消耗。当您需要将已消耗和已生成的消息批处理在一起时，应该使用此方法，通常是在消耗-转换-生成模式中。”

生产部门如何维护消耗补偿的列表？这有什么意义？

Answer 1

这只与您正在使用的工作流相关，然后根据您使用的内容生成消息。此函数仅允许在下游生产成功时提交消耗的偏移量。如果您消费数据，以某种方式对其进行处理，然后生成结果，这将在整个消费/生产过程中实现事务性保证。

在没有事务的情况下，通常使用Consumer#commitSync（）或Consumer#commitSync（）来提交Consumer偏移量。但是，如果您在与制作人制作之前使用这些方法，那么在知道制作人是否成功发送之前，您就已经提交了偏移量。

因此，您可以在生产者上使用Producer#sendOffsetsToTransaction（）来提交偏移量，而不是向消费者提交偏移量。这会将偏移量发送给处理事务的事务管理器。只有当整个事务-消费和生产-成功时，它才会提交偏移量。

（注意：当您发送要提交的偏移量时，应在上次读取的偏移量中添加1，以便以后的读取从尚未读取的偏移量继续。无论您是与消费者还是生产者提交，这都是正确的。请参阅：KafkaProducer sendOffsetsToTransaction需要偏移量1才能成功提交当前偏移量）。