laskoviymishka commented on code in PR #17552:
URL: https://github.com/apache/iceberg/pull/17552#discussion_r3774837532
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -316,6 +320,171 @@ private void triggerCommitCycle(Coordinator coordinator) {
coordinator.process();
}
+ @Test
+ public void testCommitConsumerOffsetsDoesNotRewind() {
+ when(config.commitIntervalMs()).thenReturn(0);
+ when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(), clientFactory,
context);
+ coordinator.start();
+ initConsumer();
+
+ TopicPartition ctl = new TopicPartition(CTL_TOPIC_NAME, 0);
+
+ long healthyWatermark = 100L;
+ coordinator.controlTopicOffsets().put(0, healthyWatermark);
+ coordinator.commitConsumerOffsets();
+
+ coordinator.controlTopicOffsets().put(0, 5L);
+ coordinator.commitConsumerOffsets();
+
+ OffsetAndMetadata committedOffsetAndMetadata =
+ consumer.committed(ImmutableSet.of(ctl)).get(ctl);
+ long committed = committedOffsetAndMetadata == null ? 0L :
committedOffsetAndMetadata.offset();
+
+ assertThat(committed)
+ .as("commitConsumerOffsets should not rewind the shared -coord
consumer group offsets")
+ .isEqualTo(healthyWatermark);
+ }
+
+ @Test
+ public void testCommitConsumerDuplicateDoesNotCommit() {
+ when(config.commitIntervalMs()).thenReturn(0);
+ when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(), clientFactory,
context);
+ coordinator.start();
+ initConsumer();
+
+ TopicPartition ctl = new TopicPartition(CTL_TOPIC_NAME, 0);
+
+ long healthyWatermark = 100L;
+ coordinator.controlTopicOffsets().put(0, healthyWatermark);
+ coordinator.commitConsumerOffsets();
+
+ long nextWatermark = healthyWatermark + 5;
+ consumer.commitSync(ImmutableMap.of(ctl, new
OffsetAndMetadata(nextWatermark)));
+
+ coordinator.controlTopicOffsets().put(0, 100L);
+ coordinator.commitConsumerOffsets();
+
+ OffsetAndMetadata committedOffsetAndMetadata =
+ consumer.committed(ImmutableSet.of(ctl)).get(ctl);
+ long committed = committedOffsetAndMetadata == null ? 0L :
committedOffsetAndMetadata.offset();
+
+ assertThat(committed)
+ .as(
+ "commitConsumerOffsets should not rewind offsets when consumer
group was updated by another coordinator")
+ .isEqualTo(nextWatermark);
+ }
+
+ @Test
+ public void testCommitNewConsumerAdvances() {
+ when(config.commitIntervalMs()).thenReturn(0);
+ when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(), clientFactory,
context);
+ coordinator.start();
+ initConsumer();
+
+ long newWatermark = 5L;
+
+ TopicPartition ctl = new TopicPartition(CTL_TOPIC_NAME, 0);
+ coordinator.controlTopicOffsets().put(0, newWatermark);
+
+ coordinator.commitConsumerOffsets();
+
+ OffsetAndMetadata committedOffsetAndMetadata =
+ consumer.committed(ImmutableSet.of(ctl)).get(ctl);
+ long committed = committedOffsetAndMetadata == null ? 0L :
committedOffsetAndMetadata.offset();
+
+ assertThat(committed)
+ .as("commitConsumerOffsets should advance offsets on its first commit")
+ .isEqualTo(newWatermark);
+ }
+
+ @Test
+ public void testCommitConsumerAdvances() {
+ when(config.commitIntervalMs()).thenReturn(0);
+ when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(), clientFactory,
context);
+ coordinator.start();
+ initConsumer();
+
+ long healthyWatermark = 100L;
+ TopicPartition ctl = new TopicPartition(CTL_TOPIC_NAME, 0);
+
+ coordinator.controlTopicOffsets().put(0, healthyWatermark);
+ coordinator.commitConsumerOffsets();
+
+ long watermarkToCommit = 105L;
+ coordinator.controlTopicOffsets().put(0, watermarkToCommit);
+ coordinator.commitConsumerOffsets();
+
+ OffsetAndMetadata committedOffsetAndMetadata =
+ consumer.committed(ImmutableSet.of(ctl)).get(ctl);
+ long committed = committedOffsetAndMetadata == null ? 0L :
committedOffsetAndMetadata.offset();
+
+ assertThat(committed)
+ .as("commitConsumerOffsets should advance offsets when its value is
greater")
+ .isEqualTo(watermarkToCommit);
+ }
+
+ @Test
+ public void testCommitConsumerMixedPartitionsRewindOrAdvance() {
+ when(config.commitIntervalMs()).thenReturn(0);
+ when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(), clientFactory,
context);
+ coordinator.start();
+ initConsumer();
+
+ long healthyWatermark0 = 100L;
+ long healthWatermark1 = 200L;
+ TopicPartition ctl0 = new TopicPartition(CTL_TOPIC_NAME, 0);
+ TopicPartition ctl1 = new TopicPartition(CTL_TOPIC_NAME, 1);
+
+ consumer.rebalance(ImmutableList.of(ctl0, ctl1));
+ consumer.updateBeginningOffsets(ImmutableMap.of(ctl0, 0L, ctl1, 0L));
+
+ coordinator.controlTopicOffsets().put(0, healthyWatermark0);
+ coordinator.controlTopicOffsets().put(1, healthWatermark1);
+ coordinator.commitConsumerOffsets();
+
+ long watermarkToCommit = 105L;
+ long watermarkToSkip = 195L;
+ coordinator.controlTopicOffsets().put(0, watermarkToCommit);
+ coordinator.controlTopicOffsets().put(1, watermarkToSkip);
+ coordinator.commitConsumerOffsets();
+
+ Map<TopicPartition, OffsetAndMetadata> committedOffsetAndMetadata =
+ consumer.committed(ImmutableSet.of(ctl0, ctl1));
+
+ long committed0 =
+ committedOffsetAndMetadata == null ? 0L :
committedOffsetAndMetadata.get(ctl0).offset();
Review Comment:
Follow-up nit, non-blocking: the null guard only checks the whole map, not
the individual entries — if `committed()` returns a non-null map missing
`ctl0`/`ctl1`, `.get(ctl0).offset()` NPEs and masks the real assertion failure.
Pulling each entry out first and defaulting to 0L (same shape as the
single-partition tests above) is a bit sturdier.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -142,13 +143,35 @@ protected Map<Integer, Long> controlTopicOffsets() {
return controlTopicOffsets;
}
+ /**
+ * Commit consumer offsets. Only commits offsets if it has not committed
offsets before or the
+ * value is greater than the cached offset.
+ *
+ * <p>Note: there is a risk that two parallel coordinators may overwrite
each other's offsets, to
+ * be fixed in a follow-up PR.
+ */
protected void commitConsumerOffsets() {
Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = Maps.newHashMap();
- controlTopicOffsets()
- .forEach(
- (k, v) ->
- offsetsToCommit.put(new TopicPartition(controlTopic, k), new
OffsetAndMetadata(v)));
- consumer.commitSync(offsetsToCommit);
+ controlTopicOffsets.forEach(
+ (partition, offsetToCommit) -> {
+ TopicPartition tp = new TopicPartition(controlTopic, partition);
+ Long lastCommittedOffset = committedOffsets.get(tp.partition());
Review Comment:
Follow-up nit, non-blocking: `tp.partition()` here just returns `partition`,
so `committedOffsets.get(partition)` reads the same and skips the round-trip
through the freshly-built `TopicPartition`. `tp` is still needed for the `put`
below, so keep it.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -142,13 +143,35 @@ protected Map<Integer, Long> controlTopicOffsets() {
return controlTopicOffsets;
}
+ /**
+ * Commit consumer offsets. Only commits offsets if it has not committed
offsets before or the
+ * value is greater than the cached offset.
+ *
+ * <p>Note: there is a risk that two parallel coordinators may overwrite
each other's offsets, to
+ * be fixed in a follow-up PR.
+ */
protected void commitConsumerOffsets() {
Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = Maps.newHashMap();
- controlTopicOffsets()
- .forEach(
- (k, v) ->
- offsetsToCommit.put(new TopicPartition(controlTopic, k), new
OffsetAndMetadata(v)));
- consumer.commitSync(offsetsToCommit);
+ controlTopicOffsets.forEach(
+ (partition, offsetToCommit) -> {
+ TopicPartition tp = new TopicPartition(controlTopic, partition);
+ Long lastCommittedOffset = committedOffsets.get(tp.partition());
+ if (lastCommittedOffset == null || offsetToCommit >
lastCommittedOffset) {
+ offsetsToCommit.put(tp, new OffsetAndMetadata(offsetToCommit));
+ }
+ });
+ if (!offsetsToCommit.isEmpty()) {
+ LOG.info("Coordinator committing offsets: {}", offsetsToCommit);
Review Comment:
Follow-up nit, non-blocking: this runs every commit cycle, so an INFO line
per interval gets noisy — `LOG.debug` would match the skip path just below. And
since this is the abstract `Channel`, "Committing consumer offsets: {}" avoids
mis-attributing to Coordinator for any other subclass.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -316,6 +320,171 @@ private void triggerCommitCycle(Coordinator coordinator) {
coordinator.process();
}
+ @Test
+ public void testCommitConsumerOffsetsDoesNotRewind() {
+ when(config.commitIntervalMs()).thenReturn(0);
+ when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(), clientFactory,
context);
+ coordinator.start();
+ initConsumer();
+
+ TopicPartition ctl = new TopicPartition(CTL_TOPIC_NAME, 0);
+
+ long healthyWatermark = 100L;
+ coordinator.controlTopicOffsets().put(0, healthyWatermark);
+ coordinator.commitConsumerOffsets();
+
+ coordinator.controlTopicOffsets().put(0, 5L);
+ coordinator.commitConsumerOffsets();
+
+ OffsetAndMetadata committedOffsetAndMetadata =
+ consumer.committed(ImmutableSet.of(ctl)).get(ctl);
+ long committed = committedOffsetAndMetadata == null ? 0L :
committedOffsetAndMetadata.offset();
+
+ assertThat(committed)
+ .as("commitConsumerOffsets should not rewind the shared -coord
consumer group offsets")
+ .isEqualTo(healthyWatermark);
+ }
+
+ @Test
+ public void testCommitConsumerDuplicateDoesNotCommit() {
+ when(config.commitIntervalMs()).thenReturn(0);
+ when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(), clientFactory,
context);
+ coordinator.start();
+ initConsumer();
+
+ TopicPartition ctl = new TopicPartition(CTL_TOPIC_NAME, 0);
+
+ long healthyWatermark = 100L;
+ coordinator.controlTopicOffsets().put(0, healthyWatermark);
+ coordinator.commitConsumerOffsets();
+
+ long nextWatermark = healthyWatermark + 5;
+ consumer.commitSync(ImmutableMap.of(ctl, new
OffsetAndMetadata(nextWatermark)));
+
+ coordinator.controlTopicOffsets().put(0, 100L);
+ coordinator.commitConsumerOffsets();
+
+ OffsetAndMetadata committedOffsetAndMetadata =
+ consumer.committed(ImmutableSet.of(ctl)).get(ctl);
+ long committed = committedOffsetAndMetadata == null ? 0L :
committedOffsetAndMetadata.offset();
+
+ assertThat(committed)
+ .as(
+ "commitConsumerOffsets should not rewind offsets when consumer
group was updated by another coordinator")
Review Comment:
Follow-up, non-blocking: this description says we don't rewind when another
coordinator advanced the group, but the test passes because the retry value
(100) equals what this instance already cached, so the equality guard skips it
— the broker's 105 doesn't factor into why we skip. I'd reword it to what it
actually verifies (the local-cache duplicate skip). If you want to pin the
deferred window down later, a companion case with a value strictly between the
cache and the broker — cache 100 / broker 105 / local 103 — would commit 103
and rewind under the current impl, so it documents the gap the follow-up will
close. Fine to leave for the follow-up.
--
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]
