Repository: apache/kafka
File: group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java (Java)

Code before:
            new CompletableFuture<>()
        );

        when(runtime.scheduleWriteAllOperation(
            ArgumentMatchers.eq("on-partition-deleted"),
            ArgumentMatchers.eq(Duration.ofMillis(5000)),
            ArgumentMatchers.any()
        )).thenReturn(offsetFutures);

        when(runtime.scheduleWriteAllOperation(
            ArgumentMatchers.eq("maybe-cleanup-share-group-state"),
            ArgumentMatchers.eq(Duration.ofMillis(5000)),
            ArgumentMatchers.any()
        )).thenReturn(shareFutures);

        // Run onMetadataUpdate in a separate thread.
        var resultFuture = CompletableFuture.runAsync(() -> {
            try {
                service.onMetadataUpdate(delta, newImage);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });

        // Verify method is blocked.
        assertFalse(resultFuture.isDone());

        // Complete all futures.
        offsetFutures.forEach(f -> f.complete(null));
        shareFutures.forEach(f -> f.complete(null));

        // Verify method completes.
        resultFuture.get(5, TimeUnit.SECONDS);

        // Verify operations were scheduled exactly once.
        verify(runtime, times(1)).scheduleWriteAllOperation(
            ArgumentMatchers.eq("on-partition-deleted"),
            ArgumentMatchers.eq(Duration.ofMillis(5000)),
            ArgumentMatchers.any()
        );
        verify(runtime, times(1)).scheduleWriteAllOperation(
            ArgumentMatchers.eq("maybe-cleanup-share-group-state"),
            ArgumentMatchers.eq(Duration.ofMillis(5000)),
            ArgumentMatchers.any()
        );
    }

    @Test
    public void testOnMetadataUpdateDoesNotScheduleOperationsWhenNoTopicsDeleted() {
        var runtime = mockRuntime();
        var service = new GroupCoordinatorServiceBuilder()

Review comment (bug):
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())`

Code after:
            new CompletableFuture<>(),
            new CompletableFuture<>()
        );

        when(runtime.scheduleWriteAllOperation(
            ArgumentMatchers.eq("on-partition-deleted"),
            ArgumentMatchers.eq(Duration.ofMillis(5000)),
            ArgumentMatchers.any()
        )).thenReturn(offsetFutures);

        when(runtime.scheduleWriteAllOperation(
            ArgumentMatchers.eq("maybe-cleanup-share-group-state"),
            ArgumentMatchers.eq(Duration.ofMillis(5000)),
            ArgumentMatchers.any()
        )).thenReturn(shareFutures);

        // Run onMetadataUpdate in a separate thread.
        var resultFuture = CompletableFuture.runAsync(() -> service.onMetadataUpdate(delta, newImage));

        // Wait for the operations to be scheduled and verify method is blocked.
        verify(runtime, timeout(5000).times(1)).scheduleWriteAllOperation(
            ArgumentMatchers.eq("on-partition-deleted"),
            ArgumentMatchers.eq(Duration.ofMillis(5000)),
            ArgumentMatchers.any()
        );
        verify(runtime, timeout(5000).times(1)).scheduleWriteAllOperation(
            ArgumentMatchers.eq("maybe-cleanup-share-group-state"),
            ArgumentMatchers.eq(Duration.ofMillis(5000)),
            ArgumentMatchers.any()
        );
        assertFalse(resultFuture.isDone());

        // Complete all futures.
        offsetFutures.forEach(f -> f.complete(null));
        shareFutures.forEach(f -> f.complete(null));

        // Verify method completes.
        resultFuture.get(5, TimeUnit.SECONDS);
    }

    @Test
    public void testOnMetadataUpdateDoesNotScheduleOperationsWhenNoTopicsDeleted() {
        var runtime = mockRuntime();
        var service = new GroupCoordinatorServiceBuilder()
            .setConfig(createConfig())
            .setRuntime(runtime)
            .build();
        service.startup(() -> 3);

        // Create an image with a topic and a delta with no deletions.
        var image = new MetadataImageBuilder()