@Test
    public void testExternalZombieFencingRequestImmediateCompletion() throws Exception {
        expectHerderStartup();
        when(member.memberId()).thenReturn("leader");
        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V2);
        when(herder.connectorType(anyMap())).thenReturn(ConnectorType.SOURCE);

        expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
        SessionKey sessionKey = expectNewSessionKey();

        ClusterConfigState configState = exactlyOnceSnapshot(
                sessionKey,
                TASK_CONFIGS_MAP,
                Collections.singletonMap(CONN1, 2),
                Collections.singletonMap(CONN1, 5),
                Collections.singleton(CONN1)
        );
        expectConfigRefreshAndSnapshot(configState);

        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
        expectMemberPoll();

        // The future returned by Worker::fenceZombies
        KafkaFuture<Void> workerFencingFuture = mock(KafkaFuture.class);
        // The future tracked by the herder (which tracks the fencing performed by the worker and the possible followup write to the config topic)
        KafkaFuture<Void> herderFencingFuture = mock(KafkaFuture.class);

        // Immediately invoke callbacks that the herder sets up for when the worker fencing and writes to the config topic have completed
        ArgumentCaptor<KafkaFuture.BiConsumer<Void, Throwable>> herderFencingCallback = ArgumentCaptor.forClass(KafkaFuture.BiConsumer.class);
        when(herderFencingFuture.whenComplete(herderFencingCallback.capture())).thenAnswer(invocation -> {
            herderFencingCallback.getValue().accept(null, null);
            return null;
        });

        ArgumentCaptor<KafkaFuture.BaseFunction<Void, Void>> fencingFollowup = ArgumentCaptor.forClass(KafkaFuture.BaseFunction.class);
        when(workerFencingFuture.thenApply(fencingFollowup.capture())).thenAnswer(invocation -> {
            fencingFollowup.getValue().apply(null);
            return herderFencingFuture;
        });
        when(worker.fenceZombies(eq(CONN1), eq(2), eq(CONN1_CONFIG))).thenReturn(workerFencingFuture);

        doNothing().when(configBackingStore).putTaskCountRecord(CONN1, 1);

        expectHerderShutdown();

        startBackgroundHerder();

        FutureCallback<Void> fencing = new FutureCallback<>();
        herder.fenceZombieSourceTasks(CONN1, fencing);

        fencing.get(10, TimeUnit.SECONDS);

        stopBackgroundHerder();

        verifyNoMoreInteractions(herderFencingFuture, workerFencingFuture, member, worker, configBackingStore, statusBackingStore);
    }
Tests zombie fencing that completes extremely quickly, and causes all callback-related logic to be invoked
effectively as soon as it's put into place. This is not likely to occur in practice, but the test is valuable all the
same especially since it may shed light on potential deadlocks when the unlikely-but-not-impossible happens.
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java