You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2024/06/20 07:53:55 UTC
(kafka) branch trunk updated: KAFKA-16973; Fix caught-up condition (#16367)
This is an automated email from the ASF dual-hosted git repository.
dajac pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new ee550c4b770 KAFKA-16973; Fix caught-up condition (#16367)
ee550c4b770 is described below
commit ee550c4b770402a35f2972bbbe6b75d58af8c938
Author: David Jacot <dj...@confluent.io>
AuthorDate: Thu Jun 20 09:53:48 2024 +0200
KAFKA-16973; Fix caught-up condition (#16367)
When a write operation does not have any records, the coordinator runtime checked whether the state machine is caught-up to decide whether the operation should wait until the state machine is committed up to the operation point or the operation should be completed. The current implementation assumes that there will always be a pending write operation waiting in the deferred queue when the state machine is not fully caught-up yet. This is true except when the state machine is just load [...]
This patch fixes the issue by always comparing the last written offset and the last committed offset.
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../group/runtime/CoordinatorRuntime.java | 6 +-
.../group/runtime/CoordinatorRuntimeTest.java | 83 ++++++++++++++++++++++
2 files changed, 85 insertions(+), 4 deletions(-)
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java
index ac594ebea05..21c5d624a6b 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java
@@ -60,7 +60,6 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalInt;
-import java.util.OptionalLong;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.RejectedExecutionException;
@@ -900,9 +899,8 @@ public class CoordinatorRuntime<S extends CoordinatorShard<U>, U> implements Aut
if (currentBatch != null) {
currentBatch.deferredEvents.add(event);
} else {
- OptionalLong pendingOffset = deferredEventQueue.highestPendingOffset();
- if (pendingOffset.isPresent()) {
- deferredEventQueue.add(pendingOffset.getAsLong(), event);
+ if (coordinator.lastCommittedOffset() < coordinator.lastWrittenOffset()) {
+ deferredEventQueue.add(coordinator.lastWrittenOffset(), event);
} else {
event.complete(null);
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntimeTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntimeTest.java
index 616a4f646ad..7a2adb3e8b0 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntimeTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntimeTest.java
@@ -3756,6 +3756,89 @@ public class CoordinatorRuntimeTest {
assertNotEquals(coordinator, ctx.coordinator);
}
+ @Test
+ public void testWriteOpIsNotReleasedWhenStateMachineIsNotCaughtUpAfterLoad() throws ExecutionException, InterruptedException, TimeoutException {
+ MockTimer timer = new MockTimer();
+ MockPartitionWriter writer = new MockPartitionWriter();
+ CoordinatorLoader<String> loader = new CoordinatorLoader<String>() {
+ @Override
+ public CompletableFuture<LoadSummary> load(
+ TopicPartition tp,
+ CoordinatorPlayback<String> coordinator
+ ) {
+ coordinator.replay(
+ 0,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ "record#0"
+ );
+
+ coordinator.replay(
+ 0,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ "record#1"
+ );
+
+ coordinator.updateLastWrittenOffset(2L);
+ coordinator.updateLastCommittedOffset(1L);
+
+ return CompletableFuture.completedFuture(new LoadSummary(
+ 0L,
+ 0L,
+ 0L,
+ 2,
+ 1
+ ));
+ }
+
+ @Override
+ public void close() {}
+ };
+
+ CoordinatorRuntime<MockCoordinatorShard, String> runtime =
+ new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
+ .withTime(timer.time())
+ .withTimer(timer)
+ .withDefaultWriteTimeOut(Duration.ofMillis(20))
+ .withLoader(loader)
+ .withEventProcessor(new DirectEventProcessor())
+ .withPartitionWriter(writer)
+ .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
+ .withCoordinatorRuntimeMetrics(mock(GroupCoordinatorRuntimeMetrics.class))
+ .withCoordinatorMetrics(mock(GroupCoordinatorMetrics.class))
+ .withSerializer(new StringSerializer())
+ .withAppendLingerMs(10)
+ .build();
+
+ // Schedule the loading.
+ runtime.scheduleLoadOperation(TP, 10);
+
+ // Verify the initial state.
+ CoordinatorRuntime<MockCoordinatorShard, String>.CoordinatorContext ctx = runtime.contextOrThrow(TP);
+ assertEquals(2L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(1L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(Collections.singletonList(2L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // Schedule a write operation that does not generate any records.
+ CompletableFuture<String> write = runtime.scheduleWriteOperation("write#1", TP, Duration.ofMillis(20),
+ state -> new CoordinatorResult<>(Collections.emptyList(), "response1"));
+
+ // The write operation should not be done.
+ assertFalse(write.isDone());
+
+ // Advance the last committed offset.
+ ctx.highWatermarklistener.onHighWatermarkUpdated(TP, 2L);
+
+ // Verify the state.
+ assertEquals(2L, ctx.coordinator.lastWrittenOffset());
+ assertEquals(2L, ctx.coordinator.lastCommittedOffset());
+ assertEquals(Collections.singletonList(2L), ctx.coordinator.snapshotRegistry().epochsList());
+
+ // The write operation should be completed.
+ assertEquals("response1", write.get(5, TimeUnit.SECONDS));
+ }
+
private static <S extends CoordinatorShard<U>, U> ArgumentMatcher<CoordinatorPlayback<U>> coordinatorMatcher(
CoordinatorRuntime<S, U> runtime,
TopicPartition tp