wombatu-kun commented on code in PR #17925:
URL: https://github.com/apache/iceberg/pull/17925#discussion_r3994512574
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -98,8 +100,10 @@ class Coordinator extends Channel {
this.catalog = catalog;
this.config = config;
- this.totalPartitionCount =
- members.stream().mapToInt(desc ->
desc.assignment().topicPartitions().size()).sum();
+ this.expectedPartitions =
+ members.stream()
+ .flatMap(member -> member.assignment().topicPartitions().stream())
Review Comment:
Collecting to a set also dedups the expected side: a partition claimed by
two `MemberDescription`s now counts once, which turns a commit that previously
waited out the timeout into a full watermarked one. Both new coordinator tests
pass a single member, so nothing pins that - add one with two members claiming
the same source partition.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java:
##########
@@ -114,24 +122,24 @@ boolean isCommitTimedOut() {
return false;
}
- boolean isCommitReady(int expectedPartitionCount) {
+ boolean isCommitReady(Set<TopicPartition> expectedPartitions) {
if (!isCommitInProgress()) {
return false;
}
- if (receivedPartitionCount >= expectedPartitionCount) {
+ if (reportedPartitions.containsAll(expectedPartitions)) {
LOG.info(
- "Commit {} ready, received responses for all {} partitions",
+ "Commit {} ready, received responses for all {} expected partitions",
currentCommitId,
- receivedPartitionCount);
+ expectedPartitions.size());
return true;
}
LOG.info(
- "Commit {} not ready, received responses for {} of {} partitions,
waiting for more",
+ "Commit {} not ready, received responses for {} of {} expected
partitions, waiting for more",
currentCommitId,
- receivedPartitionCount,
- expectedPartitionCount);
+ Sets.intersection(reportedPartitions, expectedPartitions).size(),
Review Comment:
`isCommitReady` now scans twice per DataComplete - `containsAll` over the
expected set, then `Sets.intersection` over the reported one only to fill a log
placeholder. Take `Sets.difference(expectedPartitions, reportedPartitions)`
once instead: `isEmpty()` gives the same readiness decision and `size()` gives
the outstanding count for the message.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -480,6 +489,158 @@ private void assertCommitComplete(int idx, UUID commitId,
OffsetDateTime ts) {
assertThat(commitCompletePayload.validThroughTs()).isEqualTo(ts);
}
+ @ParameterizedTest(name = "Repeated data starts at offset {0}")
+ @ValueSource(longs = {1, 3})
+ void repeatedDataCompleteWaitsForEveryExpectedPartition(long
repeatedWrittenOffset) {
+ MemberDescription member =
+ new MemberDescription(
+ "member",
+ Optional.empty(),
+ "client",
+ "host",
+ new MemberAssignment(
+ ImmutableSet.of(
+ new TopicPartition(SRC_TOPIC_NAME, 0), new
TopicPartition(SRC_TOPIC_NAME, 1))));
+ Coordinator coordinator = startCoordinator(ImmutableList.of(member));
+ TopicPartition controlPartition = new TopicPartition(CTL_TOPIC_NAME, 0);
+ long initialOffset = 1L;
+ consumer.commitSync(ImmutableMap.of(controlPartition, new
OffsetAndMetadata(initialOffset)));
+
+ coordinator.process();
+ UUID commitId =
+ ((StartCommit)
AvroUtil.decode(producer.history().get(0).value()).payload()).commitId();
+
+ DataFile dataFile = EventTestUtil.createDataFile();
+ Event written =
+ new Event(
+ config.connectGroupId(),
+ new DataWritten(
+ StructType.of(),
+ commitId,
+ TableReference.of("catalog", TABLE_IDENTIFIER, table.uuid()),
+ ImmutableList.of(dataFile),
+ ImmutableList.of()));
+ OffsetDateTime missingPartitionTimestamp = EventTestUtil.now();
+ OffsetDateTime firstPartitionTimestamp =
missingPartitionTimestamp.plusSeconds(1);
+ Event firstPartitionReady =
+ new Event(
+ config.connectGroupId(),
+ new DataComplete(
+ commitId,
+ ImmutableList.of(
+ new TopicPartitionOffset(SRC_TOPIC_NAME, 0, 1L,
firstPartitionTimestamp))));
+
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, initialOffset, "key",
AvroUtil.encode(written)));
+ consumer.addRecord(
+ new ConsumerRecord<>(
+ CTL_TOPIC_NAME, 0, initialOffset + 1, "key",
AvroUtil.encode(firstPartitionReady)));
+ coordinator.process();
+
+ assertCommitPending(controlPartition, initialOffset);
+
+ consumer.seek(controlPartition, repeatedWrittenOffset);
+ consumer.addRecord(
+ new ConsumerRecord<>(
+ CTL_TOPIC_NAME, 0, repeatedWrittenOffset, "key",
AvroUtil.encode(written)));
+ consumer.addRecord(
+ new ConsumerRecord<>(
+ CTL_TOPIC_NAME,
+ 0,
+ repeatedWrittenOffset + 1,
+ "key",
+ AvroUtil.encode(firstPartitionReady)));
+ coordinator.process();
+
+ assertCommitPending(controlPartition, initialOffset);
+
+ long missingPartitionOffset = repeatedWrittenOffset + 2;
+ addReadyRecord(
+ missingPartitionOffset,
+ commitId,
+ new TopicPartitionOffset(SRC_TOPIC_NAME, 1, 1L,
missingPartitionTimestamp));
+ coordinator.process();
+
+ table.refresh();
+ assertThat(producer.history()).hasSize(3);
+ assertCommitTable(1, commitId, missingPartitionTimestamp);
+ assertCommitComplete(2, commitId, missingPartitionTimestamp);
+ assertThat(table.snapshots()).hasSize(1);
+ assertThat(
+ SnapshotChanges.builderFor(table)
+ .snapshot(table.currentSnapshot())
+ .build()
+ .addedDataFiles())
+ .extracting(DataFile::location)
+ .containsExactly(dataFile.location());
+
+ long committedOffset = missingPartitionOffset + 1;
+ assertThat(table.currentSnapshot().summary())
+ .containsEntry(COMMIT_ID_SNAPSHOT_PROP, commitId.toString())
+ .containsEntry(OFFSETS_SNAPSHOT_PROP, String.format("{\"0\":%d}",
committedOffset))
+ .containsEntry(VALID_THROUGH_TS_SNAPSHOT_PROP,
missingPartitionTimestamp.toString());
+
assertThat(consumer.committed(ImmutableSet.of(controlPartition)).get(controlPartition).offset())
+ .isEqualTo(committedOffset);
+ }
+
+ @Test
+ void unexpectedPartitionDoesNotCompleteCommit() {
+ MemberDescription member =
Review Comment:
This member fixture is byte-identical to the one in
`repeatedDataCompleteWaitsForEveryExpectedPartition`. Hoist it to a helper that
takes the source partitions and returns the `MemberDescription`.
--
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]
