twthorn commented on code in PR #17552:
URL: https://github.com/apache/iceberg/pull/17552#discussion_r3753930918
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -143,12 +144,31 @@ protected Map<Integer, Long> controlTopicOffsets() {
}
protected void commitConsumerOffsets() {
+ Set<TopicPartition> partitions =
+ controlTopicOffsets().keySet().stream()
+ .map(k -> new TopicPartition(controlTopic, k))
+ .collect(Collectors.toSet());
+ Map<TopicPartition, OffsetAndMetadata> committed =
consumer.committed(partitions);
+
Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = Maps.newHashMap();
controlTopicOffsets()
.forEach(
- (k, v) ->
- offsetsToCommit.put(new TopicPartition(controlTopic, k), new
OffsetAndMetadata(v)));
- consumer.commitSync(offsetsToCommit);
+ (partition, offsetToCommit) -> {
+ TopicPartition tp = new TopicPartition(controlTopic, partition);
+ OffsetAndMetadata lastCommitted = committed.get(tp);
+ if (lastCommitted == null || offsetToCommit >
lastCommitted.offset()) {
+ offsetsToCommit.put(tp, new OffsetAndMetadata(offsetToCommit));
+ }
+ });
+ if (!offsetsToCommit.isEmpty()) {
+ LOG.info("Coordinator committing offsets: {}", offsetsToCommit);
+ consumer.commitSync(offsetsToCommit);
+ } else {
+ LOG.info(
Review Comment:
Switched to debug. On equal vs less than case, since we log the offsets in
the log line user can inspect to see which ones are equal or less than.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -143,12 +144,31 @@ protected Map<Integer, Long> controlTopicOffsets() {
}
protected void commitConsumerOffsets() {
+ Set<TopicPartition> partitions =
+ controlTopicOffsets().keySet().stream()
+ .map(k -> new TopicPartition(controlTopic, k))
+ .collect(Collectors.toSet());
+ Map<TopicPartition, OffsetAndMetadata> committed =
consumer.committed(partitions);
+
Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = Maps.newHashMap();
controlTopicOffsets()
Review Comment:
Since we have access to the attribute in this class and it's final I
switched to using it directly.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -143,12 +144,31 @@ protected Map<Integer, Long> controlTopicOffsets() {
}
protected void commitConsumerOffsets() {
+ Set<TopicPartition> partitions =
+ controlTopicOffsets().keySet().stream()
+ .map(k -> new TopicPartition(controlTopic, k))
+ .collect(Collectors.toSet());
+ Map<TopicPartition, OffsetAndMetadata> committed =
consumer.committed(partitions);
+
Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = Maps.newHashMap();
controlTopicOffsets()
.forEach(
- (k, v) ->
- offsetsToCommit.put(new TopicPartition(controlTopic, k), new
OffsetAndMetadata(v)));
- consumer.commitSync(offsetsToCommit);
+ (partition, offsetToCommit) -> {
+ TopicPartition tp = new TopicPartition(controlTopic, partition);
+ OffsetAndMetadata lastCommitted = committed.get(tp);
+ if (lastCommitted == null || offsetToCommit >
lastCommitted.offset()) {
Review Comment:
I added a comment regarding this, and agreed I'll target the general fix in
a follow-up.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -143,12 +144,31 @@ protected Map<Integer, Long> controlTopicOffsets() {
}
protected void commitConsumerOffsets() {
+ Set<TopicPartition> partitions =
+ controlTopicOffsets().keySet().stream()
+ .map(k -> new TopicPartition(controlTopic, k))
+ .collect(Collectors.toSet());
+ Map<TopicPartition, OffsetAndMetadata> committed =
consumer.committed(partitions);
Review Comment:
I'm open to add the local-only approach as in this type of incident it will
prevent the zombie for overwriting the offsets (since the zombie never receives
a new record).
However, there are some risks with local-only (e.g., delays in commit
processing) that can lead to failures. But since both have that risk, we pick
the one without the performance/reliability hit. The scenario I am considering
is this.
1. Coordinator A reads up to offset 90, commits offset 90, 90 written to
cache
2. Coordinator A reads up to offset 100, waiting for commit timer to fire
3. New coordinator created, coordinator B reads up to offset 150, commits
offset 150
4. Coordinator A commit timer fires, checks its local cache, 100 > 90,
resumes its commit and commits offset 100.
Result: offset reduced from 150 to 100. ValidationException when
reprocessing.
Thus I plan to handle this in a follow up PR, but will be a larger change so
I will create an issue with my proposal.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
