4

I'm currently working on message bus based on kafka and managed by camel and Spring. I have a XML route definition to poll events and retrieve the corresponding complete business objects from an external API that looks like that :

`

<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>

` My problem is the following : when the kafka event topic is polled, and if the api.url in my pollEnrich is not available, no business object is retrieved and the event is lost. So I need to implement a transactional logic to be able to rollback on the initial kafka polling in my route so that the same event could be polled several times till the api.url send me the awaited business object.

I tried several approaches, starting from updating my version of org.apache.camel:camel-kafka to 2.22.0 to be able to play with the manual commit. Then, I tried to implement a basic error handler (configured with maximumRedeliveries=-1 to have infinite retries) so that when the pollEnrich triggers a onException, I can set a header to avoid doing the final manual commit. It works, apparently, but my event is never repolled again.

I also tried to use the transacted tag with a org.springframework.kafka.transaction.KafkaTransactionManager instance from spring-kafka, but it's not the good approach as only the producers are transactional.

What I am missing, and what is the correct approach ?

I use Java 8, Camel 2.22.0 and Spring 4.3.18.RELEASE (not recommended but it should work).

2 Answers 2

14

It looks like a relatively new feature in Camel to support Kafka manual commits. And the documentation wasn't particularly clear. I'm using Camel 2.22.1.

From the description of your problem, you are looking for "at least once" semantics. That is you want to be able to re-process a message when there was an issue. Of course the result of this approach is that no other messages in the partition w/ a failing message can be processed (or seen) until the application can successfully process it. In the case of a failing service, this would likely result in all partitions for a given topic being blocked until the service is back up.

The Kafka uri to get this to work would look like this: kafka:TestLog?brokers=localhost:9092&groupId=kafkaGroup&maxPollRecords=3&consumersCount=1&autoOffsetReset=earliest&autoCommitEnable=false&allowManualCommit=true&breakOnFirstError=true

Breaking that down a bit:

  • kafka:TestLog : specifies the Kafka topic to consume from
  • brokers=localhost:9092 : specifies the bootstrap servers for Kafka cluster
  • groupId=kafkaGroup : specifies the Kafka consumer group
  • consumersCount=1 : specifies the number of Kafka consumers for that Camel route

The last two configuration settings are important when consuming from a Kafka topic with a number of partitions. They need to be tuned/configured so that they are taking into account the number of Camel instances you are planning to run.

The more interesting configuration to get to "at least once" semantics:

  • autoCommitEnable=false : turn off auto committing of offsets so we can use manual commits.
  • allowManualCommit=true : turn on manual commits, giving us access to the KafkaManualCommit capability (see code below).
  • breakOnFirstError=true : when this is true, the route will stop processing the rest of the messages in the batch received on last poll of the topic.
  • maxPollRecords=3 : specifies the number of messages consumed during a single poll of the Kafka topic.It is probably is a good idea to keep this set to a low number, since issues w/ a message in the batch would cause all of the messages in the batch to be re-processed.
  • autoOffsetReset=earliest : will cause the consumer to read from the earliest offset when there is a difference between the current offset and the offset marking the end of the partition (more on that in a bit).

The Camel route would look something like this:

      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");
            }
        });

After running this route and getting an error you can see the state of the consumer group with this command:

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

that might yield this result:

TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
TestLog 0 92 95 3

This is where the autoOffsetReset setting comes into play. The current offset is where the consumer group wants to consume from. If that offset (92) is the error message, then the group will fall behind as more messages (in this case two more) are added. The route (using the given settings) will cause Camel to continually process the message at offset 92 until it succeeds. If the Camel route is stopped and started, the application would pick up consuming from the earliest offset (the 92) and not the latest which would be 95 based on autoOffsetReset. Using latest would result in "lost" messages, because a restart of Camel would start processing using latest offset.

A sample application is available here

5
  • I'm getting null value from this exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class); my route is this brokers=localhost:9092&groupId=kafkaGroup&maxPollRecords=3&consumersCount=1&autoOffsetReset=earliest&autoCommitEnable=false&allowManualCommit=true&breakOnFirstError=true Dec 21, 2019 at 18:52
  • There's a flaw in your answer. The camel route keeps processing messages even though the offsets are not committed. If you send 1 message that can't be processed followed by one that can, then the offset will be updated after the last message and the first one is lost. Jan 8, 2021 at 11:20
  • If anyone is using this as a template to perform batch reads from camel kafka component, please note that this is still pulling messages off the topic one at a time, and you need to save them in memory for batch processing (ex. bulk database insert). Spring kafka looks like it polls a list of messages off a topic, but it's a facade according to this discussion: stackoverflow.com/questions/71175506/batch-consumer-camel-kafka/… Feb 22, 2022 at 15:41
  • Not sure totally follow the issue. You are right that the Kafka consumer will continue to poll and consume messages, however if the Camel config is correct the route would be pulling messages from the last good offset. That means that if a batch of 3 messages is consumed (A, B, C) during polling and the processing of B throws unhandled exception then the offset is not committed and A, B, and C will be consumed and seen again. The semantics here was "at least once" so an error causes the batch to be seen again. A more robust route would need to account for seeing a message again. Apr 25, 2022 at 22:15
  • 2
    The method commitSync has been refactor to commit()
    – A.Casanova
    Jun 1, 2023 at 9:59
0

If you are getting null value from this exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class)

You need to set allowManualCommit=true in kafka uri

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.