You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by GitBox <gi...@apache.org> on 2020/05/31 04:02:03 UTC
[GitHub] [camel] DariusX opened a new pull request #3877: Camel 14935 kafka exception
DariusX opened a new pull request #3877:
URL: https://github.com/apache/camel/pull/3877
In KafkaConsumer, clear lastProcessedOffset if Exception on partition Revoked
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [camel] davsclaus merged pull request #3877: Camel 14935 kafka exception
Posted by GitBox <gi...@apache.org>.
davsclaus merged pull request #3877:
URL: https://github.com/apache/camel/pull/3877
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [camel] DariusX commented on a change in pull request #3877: Camel 14935 kafka exception
Posted by GitBox <gi...@apache.org>.
DariusX commented on a change in pull request #3877:
URL: https://github.com/apache/camel/pull/3877#discussion_r432950461
##########
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);
+ throw e;
+ } finally {
+ lastProcessedOffset.clear();
Review comment:
Thank you for pointing that out. I was going by the comments in the JIRA.
Since the partition is being revoked, I think it is fine to clear it out always, not just on the exception.
https://issues.apache.org/jira/browse/CAMEL-14935
But, I will double-check with the reporter of the JIRA too.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [camel] ceejay66 commented on a change in pull request #3877: Camel 14935 kafka exception
Posted by GitBox <gi...@apache.org>.
ceejay66 commented on a change in pull request #3877:
URL: https://github.com/apache/camel/pull/3877#discussion_r432991029
##########
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);
+ throw e;
+ } finally {
+ lastProcessedOffset.clear();
Review comment:
Yes removal is required on success. This fixes an issue where an exception on commit prevents the map entrry being removed.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [camel] ceejay66 commented on a change in pull request #3877: Camel 14935 kafka exception
Posted by GitBox <gi...@apache.org>.
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:
users@infra.apache.org
[GitHub] [camel] DariusX commented on pull request #3877: Camel 14935 kafka exception
Posted by GitBox <gi...@apache.org>.
DariusX commented on pull request #3877:
URL: https://github.com/apache/camel/pull/3877#issuecomment-638393248
Yes, same in 2.25.x
Created new PR for that: https://github.com/apache/camel/pull/3884
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [camel] DariusX commented on pull request #3877: Camel 14935 kafka exception
Posted by GitBox <gi...@apache.org>.
DariusX commented on pull request #3877:
URL: https://github.com/apache/camel/pull/3877#issuecomment-636549053
@davsclaus The code is fine, with the clearing in the finally block.
Can this move forward, please. Thx.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [camel] DariusX commented on a change in pull request #3877: Camel 14935 kafka exception
Posted by GitBox <gi...@apache.org>.
DariusX commented on a change in pull request #3877:
URL: https://github.com/apache/camel/pull/3877#discussion_r432996002
##########
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);
+ throw e;
+ } finally {
+ lastProcessedOffset.clear();
Review comment:
Thank you.
@davsclaus Yes, doing it in the finally is correct.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [camel] davsclaus commented on a change in pull request #3877: Camel 14935 kafka exception
Posted by GitBox <gi...@apache.org>.
davsclaus commented on a change in pull request #3877:
URL: https://github.com/apache/camel/pull/3877#discussion_r432911371
##########
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);
+ throw e;
+ } finally {
+ lastProcessedOffset.clear();
Review comment:
This finally block code is always exectued also for success. Is this really intended?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [camel] ceejay66 commented on a change in pull request #3877: Camel 14935 kafka exception
Posted by GitBox <gi...@apache.org>.
ceejay66 commented on a change in pull request #3877:
URL: https://github.com/apache/camel/pull/3877#discussion_r432991029
##########
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);
+ throw e;
+ } finally {
+ lastProcessedOffset.clear();
Review comment:
Yes removal is required on success as previously. Having the removal in the finally block fixes an issue where an exception on commit prevents the map entry being removed - leading to old offsets being committed.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [camel] davsclaus commented on pull request #3877: Camel 14935 kafka exception
Posted by GitBox <gi...@apache.org>.
davsclaus commented on pull request #3877:
URL: https://github.com/apache/camel/pull/3877#issuecomment-636658659
No its not correct, you clear the map even if there is NO exception.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [camel] DariusX closed pull request #3877: Camel 14935 kafka exception
Posted by GitBox <gi...@apache.org>.
DariusX closed pull request #3877:
URL: https://github.com/apache/camel/pull/3877
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [camel] DariusX commented on a change in pull request #3877: Camel 14935 kafka exception
Posted by GitBox <gi...@apache.org>.
DariusX commented on a change in pull request #3877:
URL: https://github.com/apache/camel/pull/3877#discussion_r432950461
##########
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);
+ throw e;
+ } finally {
+ lastProcessedOffset.clear();
Review comment:
Thank you for catching that. I'm going to decline the request and create a new one.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [camel] DariusX commented on a change in pull request #3877: Camel 14935 kafka exception
Posted by GitBox <gi...@apache.org>.
DariusX commented on a change in pull request #3877:
URL: https://github.com/apache/camel/pull/3877#discussion_r432950461
##########
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);
+ throw e;
+ } finally {
+ lastProcessedOffset.clear();
Review comment:
Thank you for pointing that out. I was going by the comments in the JIRA.
Since the partition is being revoked, I think it is fine to clear it out always, not just on the exception.
But, I will double-check with the reporter of the JIRA too.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [camel] davsclaus commented on a change in pull request #3877: Camel 14935 kafka exception
Posted by GitBox <gi...@apache.org>.
davsclaus commented on a change in pull request #3877:
URL: https://github.com/apache/camel/pull/3877#discussion_r433668035
##########
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:
What if you move the clear up here, and remove it from finally. Then its only cleared when there is an exception. And on success then the code below will only remove the current offset key, which seems correct to me.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [camel] ceejay66 commented on a change in pull request #3877: Camel 14935 kafka exception
Posted by GitBox <gi...@apache.org>.
ceejay66 commented on a change in pull request #3877:
URL: https://github.com/apache/camel/pull/3877#discussion_r432991029
##########
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);
+ throw e;
+ } finally {
+ lastProcessedOffset.clear();
Review comment:
Yes removal is required on success as previously. Having the removal in the finally block fixes an issue where an exception on commit prevents the map entry being removed - leading to a risk of old offsets being committed in a future rebalance scenario.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [camel] ceejay66 commented on a change in pull request #3877: Camel 14935 kafka exception
Posted by GitBox <gi...@apache.org>.
ceejay66 commented on a change in pull request #3877:
URL: https://github.com/apache/camel/pull/3877#discussion_r432991029
##########
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);
+ throw e;
+ } finally {
+ lastProcessedOffset.clear();
Review comment:
Yes removal is required on success as previously. This fixes an issue where an exception on commit prevents the map entry being removed - leading to old offsets being committed.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [camel] davsclaus commented on pull request #3877: Camel 14935 kafka exception
Posted by GitBox <gi...@apache.org>.
davsclaus commented on pull request #3877:
URL: https://github.com/apache/camel/pull/3877#issuecomment-637941806
@DariusX is this the same problem in camel-2.25.x branch? If so you are welcome to create a PR against this branch
----------------------------------------------------------------
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:
users@infra.apache.org