lianetm commented on code in PR #21263:
URL: https://github.com/apache/kafka/pull/21263#discussion_r2669944348
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -3151,107 +3153,90 @@ public void
testCompleteTransactionWithUnexpectedPartition() {
}
@Test
- public void testOnPartitionsDeleted() {
- CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
- GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ public void testOnMetadataUpdateWhenNotStarted() {
+ var runtime = mockRuntime();
+ var service = new GroupCoordinatorServiceBuilder()
.setConfig(createConfig())
.setRuntime(runtime)
.build();
- service.startup(() -> 3);
- MetadataImage image = new MetadataImageBuilder()
+ var image = new MetadataImageBuilder()
.addTopic(Uuid.randomUuid(), "foo", 1)
.build();
+ var delta = new MetadataDelta(image);
- service.onMetadataUpdate(new MetadataDelta(image), image);
-
- when(runtime.scheduleWriteAllOperation(
- ArgumentMatchers.eq("on-partition-deleted"),
- ArgumentMatchers.eq(Duration.ofMillis(5000)),
- ArgumentMatchers.any()
- )).thenReturn(Arrays.asList(
- CompletableFuture.completedFuture(null),
- CompletableFuture.completedFuture(null),
-
FutureUtils.failedFuture(Errors.COORDINATOR_LOAD_IN_PROGRESS.exception())
- ));
-
- when(runtime.scheduleWriteAllOperation(
- ArgumentMatchers.eq("maybe-cleanup-share-group-state"),
- ArgumentMatchers.eq(Duration.ofMillis(5000)),
- ArgumentMatchers.any()
- )).thenReturn(Arrays.asList(
- CompletableFuture.completedFuture(null),
- CompletableFuture.completedFuture(null),
- CompletableFuture.completedFuture(null)
- ));
-
- // The exception is logged and swallowed.
- assertDoesNotThrow(() ->
- service.onPartitionsDeleted(
- List.of(new TopicPartition("foo", 0)),
- BufferSupplier.NO_CACHING
- )
- );
- }
-
- @Test
- public void testOnPartitionsDeletedWhenServiceIsNotStarted() {
- CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
- GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
- .setConfig(createConfig())
- .setRuntime(runtime)
- .build();
-
- assertThrows(CoordinatorNotAvailableException.class, () ->
service.onPartitionsDeleted(
- List.of(new TopicPartition("foo", 0)),
- BufferSupplier.NO_CACHING
- ));
+ assertThrows(CoordinatorNotAvailableException.class,
+ () -> service.onMetadataUpdate(delta, image));
}
@Test
- public void testOnPartitionsDeletedCleanupShareGroupState() {
- CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
- GroupCoordinatorService service = new GroupCoordinatorServiceBuilder()
+ public void testOnMetadataUpdateSchedulesOperationsWhenTopicsDeleted()
throws ExecutionException, InterruptedException, TimeoutException {
+ var runtime = mockRuntime();
+ var service = new GroupCoordinatorServiceBuilder()
.setConfig(createConfig())
.setRuntime(runtime)
.build();
service.startup(() -> 3);
- MetadataImage image = new MetadataImageBuilder()
- .addTopic(Uuid.randomUuid(), "foo", 1)
+ var topicId = Uuid.randomUuid();
+ var initialImage = new MetadataImageBuilder()
+ .addTopic(topicId, "foo", 1)
.build();
- service.onMetadataUpdate(new MetadataDelta(image), image);
+ // Create a delta that deletes the topic.
+ var delta = new MetadataDelta(initialImage);
+ delta.replay(new RemoveTopicRecord().setTopicId(topicId));
+ var newImage = delta.apply(new MetadataProvenance(1, 0, 0L, true));
+
+ // Use incomplete futures to verify method blocks.
+ var offsetFutures = List.of(
+ new CompletableFuture<>(),
+ new CompletableFuture<>(),
+ new CompletableFuture<>()
+ );
+ var shareFutures = List.of(
+ new CompletableFuture<>(),
+ new CompletableFuture<>(),
+ new CompletableFuture<>()
+ );
- // No error in partition deleted callback
when(runtime.scheduleWriteAllOperation(
ArgumentMatchers.eq("on-partition-deleted"),
ArgumentMatchers.eq(Duration.ofMillis(5000)),
ArgumentMatchers.any()
- )).thenReturn(List.of(
- CompletableFuture.completedFuture(null),
- CompletableFuture.completedFuture(null),
- CompletableFuture.completedFuture(null)
- ));
+ )).thenReturn(offsetFutures);
when(runtime.scheduleWriteAllOperation(
ArgumentMatchers.eq("maybe-cleanup-share-group-state"),
ArgumentMatchers.eq(Duration.ofMillis(5000)),
ArgumentMatchers.any()
- )).thenReturn(List.of(
- CompletableFuture.completedFuture(null),
- CompletableFuture.completedFuture(null),
-
CompletableFuture.failedFuture(Errors.COORDINATOR_LOAD_IN_PROGRESS.exception())
- ));
+ )).thenReturn(shareFutures);
+
+ // Run onMetadataUpdate in a separate thread.
+ var resultFuture = CompletableFuture.runAsync(() -> {
+ try {
+ service.onMetadataUpdate(delta, newImage);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
- // The exception is logged and swallowed.
- assertDoesNotThrow(() ->
- service.onPartitionsDeleted(
- List.of(new TopicPartition("foo", 0)),
- BufferSupplier.NO_CACHING
- )
- );
+ // Verify method is blocked.
+ assertFalse(resultFuture.isDone());
Review Comment:
Is this enough to be sure that we're blocking? I guess the future here could
be incomplete only because the async execution of `onMetadataUpdate` hasn't
made it til the end where it's supposed to block/join on the write operations
(not because it is blocked waiting on the operations completion). If so, this
wouldn't catch a bug like the one you pointed out above with the missing join().
What about we verify that we did schedule the operations before this assert?
~ `verify(runtime, timeout(xxx).times(2)).scheduleWriteAllOperation) +
assertFalse(resultFuture.isDone())`
--
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]
