nahidupa commented on code in PR #17925:
URL: https://github.com/apache/iceberg/pull/17925#discussion_r3967733540
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -129,6 +133,138 @@ public void testCommitNoFiles() {
assertThat(table.snapshots()).isEmpty();
}
+ @Test
+ public void testControlPartitionsRevokedResetsInFlightCommit() {
+ 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();
+
+ // begin a commit and buffer a worker response, but withhold DATA_COMPLETE
so the commit
+ // stays in flight
+ coordinator.process();
+ assertThat(producer.history()).hasSize(1);
+ UUID commitId =
+ ((StartCommit)
AvroUtil.decode(producer.history().get(0).value()).payload()).commitId();
+
+ Event commitResponse =
+ new Event(
+ config.connectGroupId(),
+ new DataWritten(
+ StructType.of(),
+ commitId,
+ TableReference.of("catalog", TableIdentifier.of("db", "tbl"),
null),
+ ImmutableList.of(EventTestUtil.createDataFile()),
+ ImmutableList.of()));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key",
AvroUtil.encode(commitResponse)));
+ coordinator.process();
+
+ // still mid-commit: no further event emitted
+ assertThat(producer.history()).hasSize(1);
+
+ // a control-topic rebalance revokes the partition; the in-flight commit
must be discarded
+ consumer.rebalance(ImmutableList.of());
+
+ // with the in-flight commit reset, the coordinator is free to start a
brand new commit on the
+ // next cycle. Without the reset it would still consider commit `commitId`
in progress and emit
+ // nothing here.
+ coordinator.process();
+
+ assertThat(producer.history()).hasSize(2);
+ Event newStart = AvroUtil.decode(producer.history().get(1).value());
+ assertThat(newStart.type()).isEqualTo(PayloadType.START_COMMIT);
+ assertThat(((StartCommit)
newStart.payload()).commitId()).isNotEqualTo(commitId);
+ }
+
+ /**
+ * A control-topic rebalance revokes the coordinator's assignment and,
because the coordinator's
+ * consumer resumes from its last <em>committed</em> offset, the coordinator
re-reads every
+ * control-topic record it had already consumed. This test models that
rewind: after a rebalance,
+ * the same {@code DataWritten}/{@code DataComplete} pair for a commit is
delivered again.
+ *
+ * <p>The coordinator expects responses from {@code totalPartitionCount}
partitions (two here).
+ * Before the rebalance it has heard from exactly one (partition 0), so the
commit is in flight
+ * with a readiness count of one. With the in-flight commit state reset (the
fix), the re-read
+ * {@code DataComplete} is a stale event for a commit that no longer exists,
so {@link
+ * CommitState#addReady} ignores it and the coordinator never concludes it
has heard from both
+ * partitions. Without the reset, the stale {@code DataComplete} carries the
same commit id as the
+ * still-in-flight commit, so it is counted a second time and the
coordinator fires a commit that
+ * is missing half its data and stamps a watermark the table does not yet
satisfy.
+ */
+ @Test
+ public void testControlPartitionsRevokedRewindDoesNotDoubleCount() {
+ when(config.commitIntervalMs()).thenReturn(0);
+ when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+ // two source partitions, so a commit is only ready once both have reported
+ MemberAssignment assignment =
+ new MemberAssignment(
+ ImmutableSet.of(
+ new TopicPartition(SRC_TOPIC_NAME, 0), new
TopicPartition(SRC_TOPIC_NAME, 1)));
+ MemberDescription member =
+ new MemberDescription(null, Optional.empty(), null, null, assignment);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(member),
clientFactory, context);
+ coordinator.start();
+ initConsumer();
+
+ // begin a commit and deliver partition 0's DataWritten + DataComplete, so
the commit is in
+ // flight with a readiness count of one (of two)
+ coordinator.process();
+ assertThat(producer.history()).hasSize(1);
+ UUID commitId =
+ ((StartCommit)
AvroUtil.decode(producer.history().get(0).value()).payload()).commitId();
+
+ OffsetDateTime ts = EventTestUtil.now();
+ Event dataWritten =
+ new Event(
+ config.connectGroupId(),
+ new DataWritten(
+ StructType.of(),
+ commitId,
+ TableReference.of("catalog", TableIdentifier.of("db", "tbl"),
null),
+ ImmutableList.of(EventTestUtil.createDataFile()),
+ ImmutableList.of()));
+ Event dataComplete =
+ new Event(
+ config.connectGroupId(),
+ new DataComplete(
+ commitId, ImmutableList.of(new
TopicPartitionOffset(SRC_TOPIC_NAME, 0, 3L, ts))));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key",
AvroUtil.encode(dataWritten)));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key",
AvroUtil.encode(dataComplete)));
+ coordinator.process();
+
+ // still mid-commit: only the StartCommit has been emitted
+ assertThat(producer.history()).hasSize(1);
+
+ // a control-topic rebalance revokes the partition; the in-flight commit
must be discarded
+ consumer.rebalance(ImmutableList.of());
+
+ // the consumer rewinds and re-delivers the same DataWritten +
DataComplete pair
+ consumer.rebalance(ImmutableList.of(new TopicPartition(CTL_TOPIC_NAME,
0)));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key",
AvroUtil.encode(dataWritten)));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key",
AvroUtil.encode(dataComplete)));
+ coordinator.process();
+
+ // The coordinator has heard from exactly one partition (partition 0),
delivered twice. With the
+ // reset, the re-read pair is stale and ignored, so no CommitToTable is
emitted. Without the
+ // reset, the stale DataComplete would push the readiness count to 2 and a
CommitToTable (and
+ // CommitComplete) would appear here.
+ assertThat(producer.history())
+ .noneMatch(record -> AvroUtil.decode(record.value()).type() ==
PayloadType.COMMIT_TO_TABLE);
Review Comment:
The new `testReplayedDataCompleteStillCommitsTheFileExactlyOnce` adds the
positive assertion: with two source partitions, duplicate
`DataWritten`/`DataComplete` payloads from partition 0 do not create a
snapshot; after partition 1 reports, the table has one snapshot and its added
file locations contain exactly the expected file.
There is an important limit to that coverage: the duplicate payloads are
delivered at newer Kafka offsets, not by rewinding and reconsuming the original
offsets. The test therefore establishes one file addition within this modeled
commit, not the complete rebalance/recovery guarantee requested here. A
same-offset rewind case and non-null watermark assertions remain coverage gaps.
The current test does fail at the early-snapshot assertion when partition
deduplication is disabled.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -121,6 +121,20 @@ protected void consumeAvailable(Duration pollDuration) {
while (!records.isEmpty()) {
records.forEach(
record -> {
+ // A rebalance can reassign this partition to the same consumer,
which then resumes
+ // from the group's committed offset -- behind the position this
channel already
+ // reached. Skipping the re-delivered records before the offset
update keeps
+ // controlTopicOffsets monotonic, and skipping before dispatch
keeps a replayed
+ // DataComplete from being counted toward readiness a second time.
+ Long nextOffset = controlTopicOffsets.get(record.partition());
Review Comment:
Rebuilt on main at `f05bf491a` as `2e9d2e927`. The only production delta is
now in `CommitState`: count distinct source topic/partition identities for the
active commit instead of summing assignment entries. `Channel.java`, including
the merged `Long::max` handling, is unchanged.
This leads with the double-counting invariant in `addReady` and removes the
duplicated dispatch guard from this PR. It complements dispatch protection
rather than replacing it; neither this change nor its tests establish that all
rebalance or recovery cases are safe.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestChannel.java:
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.tuple;
+import static org.mockito.Mockito.mock;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.UUID;
+import org.apache.iceberg.connect.IcebergSinkConfig;
+import org.apache.iceberg.connect.events.AvroUtil;
+import org.apache.iceberg.connect.events.Event;
+import org.apache.iceberg.connect.events.StartCommit;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.connect.sink.SinkTaskContext;
+import org.junit.jupiter.api.Test;
+
+class TestChannel extends ChannelTestBase {
Review Comment:
The duplicated dispatch guard and its proposed `TestChannel` changes were
removed. The rebuilt PR leaves both `Channel.java` and the merged `TestChannel`
unchanged, including the `hasSize(7)` dispatch expectation. Any intentional
change to that expectation belongs with the PR carrying the guard.
--
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]
