java - 如何从 Camel 以事务方式轮询 Kafka？

Question

我目前正在研究基于kafka并由camel和Spring管理的消息总线。我有一个 XML 路由定义来轮询事件并从外部 API 检索相应的完整业务对象，如下所示:

`

<route id="station-event-enrich-route" autoStartup="true" >
        <from
            uri="kafka:{{kafka.cluster.url}}?brokers={{kafka.cluster.url}}&amp;topic={{events.topic.name}}&amp;autoCommitEnable=false&amp;allowManualCommit=true&amp;maxPollRecords={{station.brocker.bulk.limit}}&amp;groupId={{kafka.groupId}}" />

        <!-- SNIP logic to aggregate several events -->

        <pollEnrich strategyRef="keepHeadersAggregationStrategy">
            <simple>{{api.url}}?view=full&amp;id=$simple{in.headers.BUSINESS_ID}</simple>
        </pollEnrich>

        <!-- SNIP logic to split the retrieved events according to their ids -->

        <to uri="velocity:velocity/resource-object.vm"/>    

        <removeHeaders pattern="*" excludePattern="MANUAL_COMMIT"/>

        <to uri="kafka:{{kafka.cluster.url}}?brokers={{kafka.cluster.url}}&amp;topic={{objects.topic.name}}&amp;groupId={{kafka.groupId}}&amp;requestRequiredAcks=all" />

        <transform>
            <simple>${headers.MANUAL_COMMIT.commitSync()}</simple>
        </transform>
</route>

`
我的问题如下:轮询 kafka 事件主题时，如果我的 pollEnrich 中的 api.url 不可用，则不会检索到业务对象并且事件丢失。所以我需要实现一个事务逻辑，以便能够在我的路由中回滚初始 kafka 轮询，以便可以多次轮询同一个事件，直到 api.url 向我发送等待的业务对象。

我尝试了几种方法，从将我的 org.apache.camel:camel-kafka 版本更新到 2.22.0 开始，以便能够进行手动提交。然后，我尝试实现一个基本的错误处理程序(配置为 maximumRedeliveries=-1 以进行无限重试)，以便当 pollEnrich 触发 onException 时，我可以设置一个 header 以避免执行最终的手动提交。显然，它有效，但我的事件再也不会被重新轮询。

我还尝试将事务标记与 spring-kafka 的 org.springframework.kafka.transaction.KafkaTransactionManager 实例一起使用，但这不是好方法，因为只有生产者是事务性的。

我缺少什么，正确的方法是什么？

我使用 Java 8、Camel 2.22.0 和 Spring 4.3.18.RELEASE(不推荐，但它应该可以工作)。

Answer 1

它看起来像是 Camel 中一个相对较新的功能来支持 Kafka 手动提交。而且文档也不是特别清楚。我正在使用 Camel 2.22.1。

根据您的问题描述，您正在寻找“至少一次”语义。也就是说，您希望能够在出现问题时重新处理消息。当然，这种方法的结果是在应用程序可以成功处理它之前，无法处理(或看到)分区中带有失败消息的其他消息。在服务失败的情况下，这可能会导致给定主题的所有分区被阻塞，直到服务备份。

使这个工作的 Kafka uri 看起来像这样:kafka:TestLog?brokers=localhost:9092&groupId=kafkaGroup&maxPollRecords=3&consumersCount=1&autoOffsetReset=earliest&autoCommitEnable=false&allowManualCommit=true&breakOnFirstError=true
稍微分解一下:

kafka:TestLog : 指定要从

消费的 Kafka 主题

brokers=localhost:9092 : 指定 Kafka 集群的引导服务器

groupId=kafkaGroup : 指定Kafka消费者组

consumersCount=1 :指定该 Camel 路由的 Kafka 消费者数量

当使用具有多个分区的 Kafka 主题时，最后两个配置设置很重要。它们需要进行调整/配置，以便将您计划运行的 Camel 实例数量考虑在内。

获得“至少一次”语义的更有趣的配置:

autoCommitEnable=false :关闭偏移量的自动提交，以便我们可以使用手动提交。

allowManualCommit=true : 打开手动提交，让我们可以访问 KafkaManualCommit能力(见下面的代码)。

breakOnFirstError=true :当这是真的时，路由将停止处理在最后一次轮询主题时收到的批处理中的其余消息。

maxPollRecords=3 : 指定在 Kafka 主题的单次轮询期间消耗的消息数。将此设置为较低的数字可能是个好主意，因为批处理中的消息问题会导致批处理中的所有消息要重新处理。

autoOffsetReset=earliest : 当当前偏移量和标记分区结束的偏移量之间存在差异时，将导致消费者从最早的偏移量读取(稍后会详细介绍)。

Camel 路线看起来像这样:

      from(kafkaUrl)
        .routeId("consumeFromKafka")
        .process(exchange -> {
            LOGGER.info(this.dumpKafkaDetails(exchange));
        })
        .process(exchange -> {
            // do something
        })
        .process(exchange -> {
            // do something else
        })
        .process(exchange -> {
            exchange.setProperty(Exchange.FILE_NAME, UUID.randomUUID().toString() + ".txt");
        })
        .to("file://files")
        // at the end of the route 
        // manage the manual commit
        .process(exchange -> {
            // manually commit offset if it is last message in batch
            Boolean lastOne = exchange.getIn().getHeader(KafkaConstants.LAST_RECORD_BEFORE_COMMIT, Boolean.class);

            if (lastOne) {
                KafkaManualCommit manual =
                        exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
                if (manual != null) {
                    LOGGER.info("manually committing the offset for batch");
                    manual.commitSync();
                }
            } else {
                LOGGER.info("NOT time to commit the offset yet");
            }
        });

运行此路由并收到错误后，您可以使用以下命令查看消费者组的状态:

./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group kafkaGroup --describe

这可能会产生这个结果:

TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG

TestLog 0 92 95 3

这是 autoOffsetReset的地方设置开始发挥作用。当前偏移量是消费者组想要消费的地方。如果该偏移量 (92) 是错误消息，则该组将随着更多消息(在本例中为另外两条)的添加而落后。路由(使用给定的设置)将导致 Camel 继续处理偏移量 92 处的消息，直到成功为止。如果 Camel 路由停止并启动，应用程序将从 earliest 开始消费。偏移量(92)而不是 latest这将是基于 autoOffsetReset 的 95 .使用 latest将导致“丢失”消息，因为重新启动 Camel 将使用最新的偏移量开始处理。

提供示例应用程序 here