CAMEL-11215: Add breakOnError to camel-kafka so the consumer stops on first exception and allow the same message to be polled again for retry.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/fe15e617 Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/fe15e617 Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/fe15e617 Branch: refs/heads/master Commit: fe15e617aba4acfebfdc85f3735e3140ea35b93d Parents: a914e6b Author: Claus Ibsen <davscl...@apache.org> Authored: Mon May 15 13:37:41 2017 +0200 Committer: Claus Ibsen <davscl...@apache.org> Committed: Thu May 25 14:17:12 2017 +0200 ---------------------------------------------------------------------- .../src/main/docs/kafka-component.adoc | 3 +- .../component/kafka/KafkaConfiguration.java | 18 ++++++ .../camel/component/kafka/KafkaConsumer.java | 61 ++++++++++++++++---- .../springboot/KafkaComponentConfiguration.java | 20 +++++++ 4 files changed, 90 insertions(+), 12 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/camel/blob/fe15e617/components/camel-kafka/src/main/docs/kafka-component.adoc ---------------------------------------------------------------------- diff --git a/components/camel-kafka/src/main/docs/kafka-component.adoc b/components/camel-kafka/src/main/docs/kafka-component.adoc index 5637673..bdbebce 100644 --- a/components/camel-kafka/src/main/docs/kafka-component.adoc +++ b/components/camel-kafka/src/main/docs/kafka-component.adoc @@ -66,7 +66,7 @@ with the following path and query parameters: | **topic** | *Required* Name of the topic to use. On the consumer you can use comma to separate multiple topics. A producer can only send a message to a single topic. | | String |======================================================================= -#### Query Parameters (83 parameters): +#### Query Parameters (84 parameters): [width="100%",cols="2,5,^1,2",options="header"] |======================================================================= @@ -77,6 +77,7 @@ with the following path and query parameters: | **autoCommitIntervalMs** (consumer) | The frequency in ms that the consumer offsets are committed to zookeeper. | 5000 | Integer | **autoCommitOnStop** (consumer) | Whether to perform an explicit auto commit when the consumer stops to ensure the broker has a commit from the last consumed message. This requires the option autoCommitEnable is turned on. The possible values are: sync async or none. And sync is the default value. | sync | String | **autoOffsetReset** (consumer) | What to do when there is no initial offset in ZooKeeper or if an offset is out of range: smallest : automatically reset the offset to the smallest offset largest : automatically reset the offset to the largest offset fail: throw exception to the consumer | latest | String +| **breakOnFirstError** (consumer) | This options controls what happens when a consumer is processing an exchange and it fails. If the option is false then the consumer continues to the next message and processes it. If the option is true then the consumer breaks out and will seek back to offset of the message that caused a failure and then re-attempt to process this message. However this can lead to endless processing of the same message if its bound to fail every time eg a poison message. Therefore its recommended to deal with that for example by using Camel's error handler. | true | boolean | **bridgeErrorHandler** (consumer) | Allows for bridging the consumer to the Camel routing Error Handler which mean any exceptions occurred while the consumer is trying to pickup incoming messages or the likes will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions that will be logged at WARN or ERROR level and ignored. | false | boolean | **checkCrcs** (consumer) | Automatically check the CRC32 of the records consumed. This ensures no on-the-wire or on-disk corruption to the messages occurred. This check adds some overhead so it may be disabled in cases seeking extreme performance. | true | Boolean | **consumerRequestTimeoutMs** (consumer) | The configuration controls the maximum amount of time the client will wait for the response of a request. If the response is not received before the timeout elapses the client will resend the request if necessary or fail the request if retries are exhausted. | 40000 | Integer http://git-wip-us.apache.org/repos/asf/camel/blob/fe15e617/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java ---------------------------------------------------------------------- diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java index bb4acbc..9c288c7 100644 --- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java +++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java @@ -115,6 +115,8 @@ public class KafkaConfiguration implements Cloneable { private Boolean autoCommitEnable = true; @UriParam(label = "consumer", defaultValue = "sync", enums = "sync,async,none") private String autoCommitOnStop = "sync"; + @UriParam(label = "consumer", defaultValue = "true") + private boolean breakOnFirstError = true; @UriParam(label = "consumer") private StateRepository<String, String> offsetRepository; @@ -684,6 +686,22 @@ public class KafkaConfiguration implements Cloneable { this.autoCommitOnStop = autoCommitOnStop; } + public boolean isBreakOnFirstError() { + return breakOnFirstError; + } + + /** + * This options controls what happens when a consumer is processing an exchange and it fails. + * If the option is <tt>false</tt> then the consumer continues to the next message and processes it. + * If the option is <tt>true</tt> then the consumer breaks out, and will seek back to offset of the + * message that caused a failure, and then re-attempt to process this message. However this can lead + * to endless processing of the same message if its bound to fail every time, eg a poison message. + * Therefore its recommended to deal with that for example by using Camel's error handler. + */ + public void setBreakOnFirstError(boolean breakOnFirstError) { + this.breakOnFirstError = breakOnFirstError; + } + public String getBrokers() { return brokers; } http://git-wip-us.apache.org/repos/asf/camel/blob/fe15e617/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java ---------------------------------------------------------------------- diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java index 321aebb..7aa6234 100644 --- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java +++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java @@ -186,15 +186,23 @@ public class KafkaConsumer extends DefaultConsumer { } } while (isRunAllowed() && !isStoppingOrStopped() && !isSuspendingOrSuspended()) { + + // flag to break out processing on the first exception + boolean breakOnErrorHit = false; + log.trace("Polling {} from topic: {} with timeout: {}", threadId, topicName, pollTimeoutMs); ConsumerRecords<Object, Object> allRecords = consumer.poll(pollTimeoutMs); + for (TopicPartition partition : allRecords.partitions()) { Iterator<ConsumerRecord<Object, Object>> recordIterator = allRecords.records(partition).iterator(); - if (recordIterator.hasNext()) { - ConsumerRecord<Object, Object> record = null; - while (recordIterator.hasNext()) { + if (!breakOnErrorHit && recordIterator.hasNext()) { + ConsumerRecord<Object, Object> record; + + long partitionLastOffset = -1; + + while (!breakOnErrorHit && recordIterator.hasNext()) { record = recordIterator.next(); if (LOG.isTraceEnabled()) { - LOG.trace("partition = {}, offset = {}, key = {}, value = {}", record.partition(), record.offset(), record.key(), + LOG.trace("Partition = {}, offset = {}, key = {}, value = {}", record.partition(), record.offset(), record.key(), record.value()); } Exchange exchange = endpoint.createKafkaExchange(record); @@ -204,15 +212,34 @@ public class KafkaConsumer extends DefaultConsumer { try { processor.process(exchange); } catch (Exception e) { - getExceptionHandler().handleException("Error during processing", exchange, e); + exchange.setException(e); + } + + if (exchange.getException() != null) { + // processing failed due to an unhandled exception, what should we do + if (endpoint.getConfiguration().isBreakOnFirstError()) { + // commit last good offset before we try again + commitOffset(offsetRepository, partition, partitionLastOffset); + // we are failing but store last good offset + log.warn("Error during processing {} from topic: {}. Will seek consumer to last good offset: {} and poll again.", exchange, topicName, partitionLastOffset); + if (partitionLastOffset != -1) { + consumer.seek(partition, partitionLastOffset); + } + // continue to next partition + breakOnErrorHit = true; + } else { + // will handle/log the exception and then continue to next + getExceptionHandler().handleException("Error during processing", exchange, exchange.getException()); + } + } else { + // record was success so remember its offset + partitionLastOffset = record.offset(); } } - long partitionLastOffset = record.offset(); - if (offsetRepository != null) { - offsetRepository.setState(serializeOffsetKey(partition), serializeOffsetValue(partitionLastOffset)); - // if autocommit is false - } else if (endpoint.getConfiguration().isAutoCommitEnable() != null && !endpoint.getConfiguration().isAutoCommitEnable()) { - consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(partitionLastOffset + 1))); + + if (!breakOnErrorHit) { + // all records processed from partition so commit them + commitOffset(offsetRepository, partition, partitionLastOffset); } } } @@ -243,6 +270,18 @@ public class KafkaConsumer extends DefaultConsumer { } } + private void commitOffset(StateRepository<String, String> offsetRepository, TopicPartition partition, long partitionLastOffset) { + if (partitionLastOffset != -1) { + if (offsetRepository != null) { + offsetRepository.setState(serializeOffsetKey(partition), serializeOffsetValue(partitionLastOffset)); + // if autocommit is false + } else if (endpoint.getConfiguration().isAutoCommitEnable() != null && !endpoint.getConfiguration().isAutoCommitEnable()) { + LOG.debug("Auto commitSync {} from topic {} with offset: {}", threadId, topicName, partitionLastOffset); + consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(partitionLastOffset + 1))); + } + } + } + private void shutdown() { // As advised in the KAFKA-1894 ticket, calling this wakeup method breaks the infinite loop consumer.wakeup(); http://git-wip-us.apache.org/repos/asf/camel/blob/fe15e617/platforms/spring-boot/components-starter/camel-kafka-starter/src/main/java/org/apache/camel/component/kafka/springboot/KafkaComponentConfiguration.java ---------------------------------------------------------------------- diff --git a/platforms/spring-boot/components-starter/camel-kafka-starter/src/main/java/org/apache/camel/component/kafka/springboot/KafkaComponentConfiguration.java b/platforms/spring-boot/components-starter/camel-kafka-starter/src/main/java/org/apache/camel/component/kafka/springboot/KafkaComponentConfiguration.java index 9d3b644..0559abb 100644 --- a/platforms/spring-boot/components-starter/camel-kafka-starter/src/main/java/org/apache/camel/component/kafka/springboot/KafkaComponentConfiguration.java +++ b/platforms/spring-boot/components-starter/camel-kafka-starter/src/main/java/org/apache/camel/component/kafka/springboot/KafkaComponentConfiguration.java @@ -206,6 +206,18 @@ public class KafkaComponentConfiguration */ private String autoCommitOnStop = "sync"; /** + * This options controls what happens when a consumer is processing an + * exchange and it fails. If the option is <tt>false</tt> then the + * consumer continues to the next message and processes it. If the + * option is <tt>true</tt> then the consumer breaks out, and will seek + * back to offset of the message that caused a failure, and then + * re-attempt to process this message. However this can lead to endless + * processing of the same message if its bound to fail every time, eg a + * poison message. Therefore its recommended to deal with that for + * example by using Camel's error handler. + */ + private Boolean breakOnFirstError = true; + /** * URL of the Kafka brokers to use. The format is * host1:port1,host2:port2, and the list can be a subset of brokers or a * VIP pointing to a subset of brokers. @@ -770,6 +782,14 @@ public class KafkaComponentConfiguration this.autoCommitOnStop = autoCommitOnStop; } + public Boolean getBreakOnFirstError() { + return breakOnFirstError; + } + + public void setBreakOnFirstError(Boolean breakOnFirstError) { + this.breakOnFirstError = breakOnFirstError; + } + public String getBrokers() { return brokers; }
