如何从 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 时，我可以设置一个标头以避免执行最终的手动提交。显然，它有效，但我的事件再也不会被重新轮询。

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

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

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

Answer 1

它看起来像是 Camel 中一个相对较新的功能来支持 Kafka 手动提交。而且文档也不是特别清楚。我正在使用骆驼 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 : 当当前偏移量与标记分区结束的偏移量之间存在差异时，将导致消费者从最早的偏移量读取（稍后会详细介绍）。

骆驼路线看起来像这样：

      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基于 95的偏移量中获取消耗autoOffsetReset。使用latest会导致“丢失”消息，因为重新启动 Camel 将使用最新的偏移量开始处理。

此处提供示例应用程序