osdir.com

[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

Camel Kafka Consumer KafkaException ignored


I have an application that uses Camel 2.21.1 and Kafka 1.1

I had a bug where the deserializer for the consumer threw an exception when deserializing the message. I could see an exception in the log, but the consumer never handled the exception and the message was throwing an exception each time, forever. Why didn't camel see the exception so that the application code could commit that offset?

This is my consumer endpoint
                @EndpointInject(uri = "kafka:{{deep.send.order.email}}?brokers={{deep.notify.kafka.server}}&groupId=orders-enrichment-grp&"+
                        "autoCommitEnable=false&allowManualCommit=true&" +
            "maxPollRecords={{deep.kafka.poll.count}}&" +
                                                "keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer&" +
                                                "valueDeserializer=com.sial.notifications.kafka.utilities.MessageJSONFormatter")

The problem was in the MessageJSONFormatter and I have fixed the problem. However, I expected that the exception be caught in Camel's kafka consumer and propagated to the global handler where the offset could be committed

How do I address an issue like this in the future?