Skip to content

Kafka Connect: Exclude zombie commitId from validThroughTs() #17080

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
thswlsqls wants to merge 1 commit into from
+40 −5
Closed

Kafka Connect: Exclude zombie commitId from validThroughTs() #17080

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions ...a-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,16 +144,21 @@ Map<TableReference, List<Envelope>> tableCommitMap() {
}

OffsetDateTime validThroughTs(boolean partialCommit) {
List<DataComplete> currentCommitReady =
readyBuffer.stream()
.filter(payload -> Objects.equals(currentCommitId, payload.commitId()))
.collect(Collectors.toList());

boolean hasValidThroughTs =
!partialCommit
&& readyBuffer.stream()
&& currentCommitReady.stream()
.flatMap(event -> event.assignments().stream())
.allMatch(offset -> offset.timestamp() != null);

OffsetDateTime result;
if (hasValidThroughTs) {
result =
readyBuffer.stream()
currentCommitReady.stream()
.flatMap(event -> event.assignments().stream())
.map(TopicPartitionOffset::timestamp)
.min(Comparator.naturalOrder())
Expand Down
36 changes: 33 additions & 3 deletions ...nnect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCommitState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,7 +113,11 @@ public void testIsCommitReadyIgnoresZombieCoordinatorPayloads() {

@Test
public void testGetValidThroughTs() {
CommitState commitState = new CommitState(mock(IcebergSinkConfig.class));
commitState.startNewCommit();

DataComplete payload1 = mock(DataComplete.class);
when(payload1.commitId()).thenReturn(commitState.currentCommitId());
TopicPartitionOffset tp1 = mock(TopicPartitionOffset.class);
OffsetDateTime ts1 = EventTestUtil.now();
when(tp1.timestamp()).thenReturn(ts1);
Expand All @@ -124,14 +128,12 @@ public void testGetValidThroughTs() {
when(payload1.assignments()).thenReturn(ImmutableList.of(tp1, tp2));

DataComplete payload2 = mock(DataComplete.class);
when(payload2.commitId()).thenReturn(commitState.currentCommitId());
TopicPartitionOffset tp3 = mock(TopicPartitionOffset.class);
OffsetDateTime ts3 = ts1.plusSeconds(2);
when(tp3.timestamp()).thenReturn(ts3);
when(payload2.assignments()).thenReturn(ImmutableList.of(tp3));

CommitState commitState = new CommitState(mock(IcebergSinkConfig.class));
commitState.startNewCommit();

commitState.addReady(wrapInEnvelope(payload1));
commitState.addReady(wrapInEnvelope(payload2));

Expand All @@ -140,6 +142,7 @@ public void testGetValidThroughTs() {

// null timestamp for one, so should not set a valid-through timestamp
DataComplete payload3 = mock(DataComplete.class);
when(payload3.commitId()).thenReturn(commitState.currentCommitId());
TopicPartitionOffset tp4 = mock(TopicPartitionOffset.class);
when(tp4.timestamp()).thenReturn(null);
when(payload3.assignments()).thenReturn(ImmutableList.of(tp4));
Expand All @@ -150,6 +153,33 @@ public void testGetValidThroughTs() {
assertThat(commitState.validThroughTs(true)).isNull();
}

@Test
public void testGetValidThroughTsIgnoresZombieCoordinatorPayloads() {
CommitState commitState = new CommitState(mock(IcebergSinkConfig.class));
commitState.startNewCommit();

// Stale DataComplete from a zombie Coordinator that started a different commit, with a null
// timestamp that would otherwise suppress the valid-through timestamp.
DataComplete zombiePayload = mock(DataComplete.class);
when(zombiePayload.commitId()).thenReturn(UUID.randomUUID());
TopicPartitionOffset zombieTp = mock(TopicPartitionOffset.class);
when(zombieTp.timestamp()).thenReturn(null);
when(zombiePayload.assignments()).thenReturn(ImmutableList.of(zombieTp));

DataComplete currentPayload = mock(DataComplete.class);
when(currentPayload.commitId()).thenReturn(commitState.currentCommitId());
TopicPartitionOffset currentTp = mock(TopicPartitionOffset.class);
OffsetDateTime ts = EventTestUtil.now();
when(currentTp.timestamp()).thenReturn(ts);
when(currentPayload.assignments()).thenReturn(ImmutableList.of(currentTp));

commitState.addReady(wrapInEnvelope(zombiePayload));
commitState.addReady(wrapInEnvelope(currentPayload));

// Only the current commit's payload counts toward the valid-through timestamp.
assertThat(commitState.validThroughTs(false)).isEqualTo(ts);
}

private Envelope wrapInEnvelope(Payload payload) {
Event event = mock(Event.class);
when(event.payload()).thenReturn(payload);
Expand Down