By default, the Kafka consumer will use auto commit, where the offset will be committed automatically in the background using a given interval.
Enable Manual Commit 🔗
camel.component.kafka.auto-commit-enable=false
camel.component.kafka.allow-manual-commit=true
The Kafka consumer periodically fetch a set of records and feeds them to the route.
If an exception is thrown while processing the current message at offset N.
camel.component.kafka.break-on-first-error=true is required to stop processing further messages.
Otherwise the next message would be processed and offset N + 1 would be commited.
The following 3 properties are required to never skip a failed message:
camel.component.kafka.auto-commit-enable=false
camel.component.kafka.allow-manual-commit=true
camel.component.kafka.break-on-first-error=true
When to commit the Kafka offset 🔗
The following route is processing Kafka messages from input to output topics. After each message is processed successfully the current offset is commited and the next message goes in.
@Component
public class KafkaRouteBuilder extends RouteBuilder {
@Override
public void configure() {
from("kafka:input")
.routePolicy(new SuspendRoutePolicy())
.process(e -> {
// do work
})
.to("kafka:output")
.process(e -> {
KafkaManualCommit commit = exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
commit.commit();
});
}
}
Route policy SuspendRoutePolicy is required to prevent the Kafka consumer to poll records at offset N and feed endlessly the same failing message into the route.
Add a Dead Letter Queue 🔗
If an exception is thrown within the route, onException pipeline will process the failed message.
@Component
public class KafkaRouteBuilder extends RouteBuilder {
@Override
public void configure() {
onException(Exception.class)
.handled(true)
.useOriginalMessage()
.maximumRedeliveries(0)
.to("kafka:dlq")
.process(e -> {
KafkaManualCommit commit = exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
commit.commit();
});
from("kafka:input")
.routePolicy(new SuspendRoutePolicy())
.process(e -> {
// do work
})
.to("kafka:output")
.process(e -> {
KafkaManualCommit commit = exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
commit.commit();
});
}
}
If an exception is thrown within the onException pipeline, since there is no other exception handler onException won’t be called again.
Instead the message will be marked as failed and the route policy SuspendRoutePolicy will pause the Kafka consumer.
Code snippets 🔗
public class SuspendRoutePolicy extends RoutePolicySupport {
@Override
public void onExchangeDone(Route route, Exchange exchange) {
if (exchange.isFailed()) {
try {
suspendRoute(route); // synchronously stop consuming messages
} catch (Exception e) {
handleException(e);
}
}
}
}
@Component
public class KafkaCommitProcessor implements Processor {
@Override
public void process(Exchange exchange) {
KafkaManualCommit commit =
exchange.getIn().getHeader(
KafkaConstants.MANUAL_COMMIT,
KafkaManualCommit.class);
if (commit != null) {
commit.commit();
}
}
}