jolshan commented on code in PR #14985: URL: https://github.com/apache/kafka/pull/14985#discussion_r1434244004
########## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntimeTest.java: ########## @@ -1073,6 +1151,314 @@ public CoordinatorShardBuilder<MockCoordinatorShard, String> get() { ); } + @ParameterizedTest + @EnumSource(value = TransactionResult.class) + public void testScheduleTransactionCompletion(TransactionResult result) throws ExecutionException, InterruptedException, TimeoutException { + MockTimer timer = new MockTimer(); + MockPartitionWriter writer = new MockPartitionWriter(); + + CoordinatorRuntime<MockCoordinatorShard, String> runtime = + new CoordinatorRuntime.Builder<MockCoordinatorShard, String>() + .withTime(timer.time()) + .withTimer(timer) + .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT) + .withLoader(new MockCoordinatorLoader()) + .withEventProcessor(new DirectEventProcessor()) + .withPartitionWriter(writer) + .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier()) + .withCoordinatorRuntimeMetrics(mock(GroupCoordinatorRuntimeMetrics.class)) + .withCoordinatorMetrics(mock(GroupCoordinatorMetrics.class)) + .build(); + + // Schedule the loading. + runtime.scheduleLoadOperation(TP, 10); + + // Verify the initial state. + CoordinatorRuntime<MockCoordinatorShard, String>.CoordinatorContext ctx = runtime.contextOrThrow(TP); + assertEquals(0L, ctx.coordinator.lastWrittenOffset()); + assertEquals(0L, ctx.coordinator.lastCommittedOffset()); + assertEquals(Collections.singletonList(0L), ctx.coordinator.snapshotRegistry().epochsList()); + + // Transactional write #1. + CompletableFuture<String> write1 = runtime.scheduleTransactionalWriteOperation( + "write#1", + TP, + "transactional-id", + 100L, + (short) 5, + DEFAULT_WRITE_TIMEOUT, + state -> new CoordinatorResult<>(Arrays.asList("record1", "record2"), "response1") + ); + + // Verify that the write is not committed yet. + assertFalse(write1.isDone()); + + // The last written offset is updated. + assertEquals(2L, ctx.coordinator.lastWrittenOffset()); + // The last committed offset does not change. + assertEquals(0L, ctx.coordinator.lastCommittedOffset()); + // A new snapshot is created. + assertEquals(Arrays.asList(0L, 2L), ctx.coordinator.snapshotRegistry().epochsList()); + // Records have been replayed to the coordinator. They are stored in + // the pending set for now. + assertEquals(mkSet("record1", "record2"), ctx.coordinator.coordinator().pendingRecords( + 100L + )); + // Records have been written to the log. + assertEquals(Arrays.asList( + InMemoryPartitionWriter.LogEntry.value(100L, (short) 5, "record1"), + InMemoryPartitionWriter.LogEntry.value(100L, (short) 5, "record2") + ), writer.entries(TP)); + + // Complete transaction #1. + CompletableFuture<Void> complete1 = runtime.scheduleTransactionCompletion( + "complete#1", + TP, + 100L, + (short) 5, + 10, + result, + DEFAULT_WRITE_TIMEOUT + ); + + // Verify that the completion is not committed yet. + assertFalse(complete1.isDone()); + + // The last written offset is updated. + assertEquals(3L, ctx.coordinator.lastWrittenOffset()); + // The last committed offset does not change. + assertEquals(0L, ctx.coordinator.lastCommittedOffset()); + // A new snapshot is created. + assertEquals(Arrays.asList(0L, 2L, 3L), ctx.coordinator.snapshotRegistry().epochsList()); + // Records have been replayed to the coordinator. + if (result == TransactionResult.COMMIT) { + // They are now in the records set if committed. + assertEquals(mkSet("record1", "record2"), ctx.coordinator.coordinator().records()); + } else { + // Or they are gone if aborted. + assertEquals(Collections.emptySet(), ctx.coordinator.coordinator().records()); + } + + // Records have been written to the log. + assertEquals(Arrays.asList( + InMemoryPartitionWriter.LogEntry.value(100L, (short) 5, "record1"), + InMemoryPartitionWriter.LogEntry.value(100L, (short) 5, "record2"), + InMemoryPartitionWriter.LogEntry.control(100L, (short) 5, 10, result) + ), writer.entries(TP)); + + // Commit write #1. + writer.commit(TP, 2); + + // The write is completed. + assertTrue(write1.isDone()); + assertEquals("response1", write1.get(5, TimeUnit.SECONDS)); + + // Commit completion #1. + writer.commit(TP, 3); + + // The transaction is completed. + assertTrue(complete1.isDone()); + assertNull(complete1.get(5, TimeUnit.SECONDS)); + } + + @Test + public void testScheduleTransactionCompletionWhenWriteTimesOut() throws InterruptedException { + MockTimer timer = new MockTimer(); + MockPartitionWriter writer = new MockPartitionWriter(); + + CoordinatorRuntime<MockCoordinatorShard, String> runtime = + new CoordinatorRuntime.Builder<MockCoordinatorShard, String>() + .withTime(timer.time()) + .withTimer(timer) + .withDefaultWriteTimeOut(DEFAULT_WRITE_TIMEOUT) + .withLoader(new MockCoordinatorLoader()) + .withEventProcessor(new DirectEventProcessor()) + .withPartitionWriter(writer) + .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier()) + .withCoordinatorRuntimeMetrics(mock(GroupCoordinatorRuntimeMetrics.class)) + .withCoordinatorMetrics(mock(GroupCoordinatorMetrics.class)) + .build(); + + // Loads the coordinator. + runtime.scheduleLoadOperation(TP, 10); + + // Verify the initial state. + CoordinatorRuntime<MockCoordinatorShard, String>.CoordinatorContext ctx = runtime.contextOrThrow(TP); + assertEquals(0, ctx.coordinator.lastWrittenOffset()); + assertEquals(0, ctx.coordinator.lastCommittedOffset()); + assertEquals(Collections.singletonList(0L), ctx.coordinator.snapshotRegistry().epochsList()); + + // Complete #1. We should get a TimeoutException because the HWM will not advance. + CompletableFuture<Void> timedOutCompletion = runtime.scheduleTransactionCompletion( + "complete#1", + TP, + 100L, + (short) 5, + 10, + TransactionResult.COMMIT, + Duration.ofMillis(3) + ); + + timer.advanceClock(4); + + assertFutureThrows(timedOutCompletion, org.apache.kafka.common.errors.TimeoutException.class); Review Comment: synced offline here. We only rollback if the write fails. If the write completes we either wait for it to replicate or lose the info if the leader changes. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org