twthorn commented on code in PR #17552:
URL: https://github.com/apache/iceberg/pull/17552#discussion_r3787029815
##########
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:
Nice catch, since tp only used in the conditional we can defer it's creation
to later.
##########
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:
Thanks, missed this the log level on this one! And updated the message
##########
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:
Good point, updated as well.
##########
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:
Nice catch, updated and added the limitation test (tests reinforces
documentation)
--
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]
