ceejay66 commented on a change in pull request #3877:
URL: https://github.com/apache/camel/pull/3877#discussion_r434241637
##########
File path:
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java
##########
@@ -460,7 +460,14 @@ public void onPartitionsRevoked(Collection<TopicPartition>
partitions) {
offset = -1L;
}
LOG.debug("Saving offset repository state {} from offsetKey {}
with offset: {}", threadId, offsetKey, offset);
- commitOffset(offsetRepository, partition, offset, true);
+ try {
+ commitOffset(offsetRepository, partition, offset, true);
+ } catch (java.lang.Exception e) {
+ LOG.error("Error saving offset repository state {} from
offsetKey {} with offset: {}", threadId, offsetKey, offset);
Review comment:
We don't want to clear the map as it is used by multiple consumers and
we will lose their state of last processed offset that may not be committed
yet. We just want to remove the entry from the map as it did originally but
ensure we do this in the exception case as well as the non exception case.
Moving the remove to the finally block gives us that behavior
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
