如何在事务上从Camel中选出Kafka？

Question

我目前正在开发基于Kafka的消息总线，由camel和Spring管理。我有一个XML路由定义，用于轮询事件，并从外部API检索相应的完整业务对象，该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事件主题被轮询时，如果api。my PollRich中的url不可用，未检索到任何业务对象，事件丢失。因此，我需要实现一个事务逻辑，以便能够回滚路由中的初始kafka轮询，以便可以多次轮询同一事件，直到api启动。url将等待的业务对象发送给我。

我尝试了几种方法，从更新我的org.apache.camel版本开始：骆驼Kafka到2.22.0，以便能够玩手动提交。然后，我尝试实现一个基本的错误处理程序（配置为最大值重新交付=-1以进行无限重试），这样当污染丰富触发onExc0019时，我可以设置一个标头来避免执行最后的手动提交。显然，它是有效的，但是我的事件再也不会被重新轮询了。

我还尝试将Transactid标记用于组织。springframework。Kafka。交易来自spring kafka的KafkaTransactionManager实例，但这不是好方法，因为只有生产者是事务性的。

我错过了什么，正确的方法是什么？

我使用Java8、Camel 2.22.0和Spring 4.3.18。发布（不推荐，但应该有效）。

Answer 1

如果你从这个交换中得到空值。getIn（）。getHeader（KafkaConstants.MANUAL_COMMIT，KafkaManualCommit.class）

您需要在kafka uri中设置allowManualCommit=true

Answer 2

它看起来像是骆驼中支持Kafka手动提交的一个相对较新的功能。留档不是特别清楚。我使用骆驼2.22.1。

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

让它工作的Kafkauri如下所示：Kafka:TestLog？brokers=localhost:9092

把它分解一下：

• kafka： TestLog：指定要从使用的Kafka主题
• brokers=localhost:9092：指定Kafka集群的引导服务器
• group pId=kafkaGroup：指定Kafka消费者组
• 消费者Count=1：指定该骆驼路线的Kafka消费者数量

在使用带有多个分区的Kafka主题时，最后两个配置设置很重要。需要对它们进行优化/配置，以便它们考虑到您计划运行的Camel实例的数量。

更有趣的配置得到"至少一次"语义学：

• 自动提交启用=false：关闭偏移量的自动提交，以便我们可以使用手动提交。
• 允许ManualCommit=true：打开手动提交，让我们可以访问KafkaManualCommit功能（请参阅下面的代码）。
• broOnFirstError=true：当此值为true时，路由将停止处理上次轮询主题时接收到的批处理中的其余消息。
• maxPollRecords=3：指定在对Kafka主题进行一次轮询期间所消耗的消息数。将此设置保持在较低的数量可能是一个好主意，因为在批处理中发出消息会导致批处理中的所有消息都被重新处理。
• 自动偏移重置=最早的：当当前偏移量和标记分区结束的偏移量之间存在差异时，将导致消费者从最早的偏移量读取（稍后会有更多信息）。

骆驼路线看起来像这样：

      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路由，应用程序将从最早的偏移量（92）而不是最新的中获取消耗，后者基于autoOffsetReset为95。使用最新将导致丢失消息，因为Camel的重新启动将使用最新偏移开始处理。

这里提供了一个示例应用程序