This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/master by this push:
new f667f6c CAMEL-12031: KafkaConsumer stops consuming messages when
exception occurs during offset commit
f667f6c is described below
commit f667f6c891cb111461bfadeb7423840e86300444
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Dec 18 15:04:08 2017 +0100
CAMEL-12031: KafkaConsumer stops consuming messages when exception occurs
during offset commit
---
.../java/org/apache/camel/component/kafka/KafkaConsumer.java | 12 ++++++++++++
1 file changed, 12 insertions(+)
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 b8b2d0c..769a5b8 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
@@ -39,6 +39,7 @@ import
org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.InterruptException;
@@ -190,6 +191,7 @@ public class KafkaConsumer extends DefaultConsumer {
protected boolean doRun() {
// allow to re-connect thread in case we use that to retry failed
messages
boolean reConnect = false;
+ boolean unsubscribing = false;
try {
if (topicPattern != null) {
@@ -323,12 +325,22 @@ public class KafkaConsumer extends DefaultConsumer {
}
log.info("Unsubscribing {} from topic {}", threadId,
topicName);
+ // we are unsubscribing so do not re connect
+ unsubscribing = true;
consumer.unsubscribe();
} catch (InterruptException e) {
getExceptionHandler().handleException("Interrupted while
consuming " + threadId + " from kafka topic", e);
log.info("Unsubscribing {} from topic {}", threadId,
topicName);
consumer.unsubscribe();
Thread.currentThread().interrupt();
+ } catch (KafkaException e) {
+ // some kind of error in kafka, it may happen during
unsubscribing or during normal processing
+ if (unsubscribing) {
+ getExceptionHandler().handleException("Error unsubscribing
" + threadId + " from kafka topic " + topicName, e);
+ } else {
+ log.warn("KafkaException consuming {} from topic {}. Will
attempt to re-connect on next run", threadId, topicName);
+ reConnect = true;
+ }
} catch (Exception e) {
getExceptionHandler().handleException("Error consuming " +
threadId + " from kafka topic", e);
} finally {
--
To stop receiving notification emails like this one, please contact
['"[email protected]" <[email protected]>'].
