@Test
    public void testExternalZombieFencingRequestDelayedCompletion() throws Exception {
        final String conn3 = "sourceC";
        final Map<String, Integer> tasksPerConnector = new HashMap<>();
        tasksPerConnector.put(CONN1, 5);
        tasksPerConnector.put(CONN2, 3);
        tasksPerConnector.put(conn3, 12);

        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();

        Map<String, Integer> taskCountRecords = new HashMap<>();
        taskCountRecords.put(CONN1, 2);
        taskCountRecords.put(CONN2, 3);
        taskCountRecords.put(conn3, 5);
        Map<String, Integer> taskConfigGenerations = new HashMap<>();
        taskConfigGenerations.put(CONN1, 3);
        taskConfigGenerations.put(CONN2, 4);
        taskConfigGenerations.put(conn3, 2);
        Set<String> pendingFencing = new HashSet<>(Arrays.asList(CONN1, CONN2, conn3));
        ClusterConfigState configState = exactlyOnceSnapshot(
                sessionKey,
                TASK_CONFIGS_MAP,
                taskCountRecords,
                taskConfigGenerations,
                pendingFencing,
                tasksPerConnector
        );
        expectConfigRefreshAndSnapshot(configState);

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

        // The callbacks that the herder has accrued for outstanding fencing futures, which will be completed after
        // a successful round of fencing and a task record write to the config topic
        Map<String, ArgumentCaptor<KafkaFuture.BiConsumer<Void, Throwable>>> herderFencingCallbacks = new HashMap<>();
        // The callbacks that the herder has installed for after a successful round of zombie fencing, but before writing
        // a task record to the config topic
        Map<String, ArgumentCaptor<KafkaFuture.BaseFunction<Void, Void>>> workerFencingFollowups = new HashMap<>();

        Map<String, CountDownLatch> callbacksInstalled = new HashMap<>();
        tasksPerConnector.keySet().forEach(connector -> {
            // 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);

            ArgumentCaptor<KafkaFuture.BiConsumer<Void, Throwable>> herderFencingCallback = ArgumentCaptor.forClass(KafkaFuture.BiConsumer.class);
            herderFencingCallbacks.put(connector, herderFencingCallback);

            // Don't immediately invoke callbacks that the herder sets up for when the worker fencing and writes to the config topic have completed
            // Instead, wait for them to be installed, then invoke them explicitly after the fact on a thread separate from the herder's tick thread
            when(herderFencingFuture.whenComplete(herderFencingCallback.capture())).thenReturn(null);

            ArgumentCaptor<KafkaFuture.BaseFunction<Void, Void>> fencingFollowup = ArgumentCaptor.forClass(KafkaFuture.BaseFunction.class);
            CountDownLatch callbackInstalled = new CountDownLatch(1);
            workerFencingFollowups.put(connector, fencingFollowup);
            callbacksInstalled.put(connector, callbackInstalled);
            when(workerFencingFuture.thenApply(fencingFollowup.capture())).thenAnswer(invocation -> {
                callbackInstalled.countDown();
                return herderFencingFuture;
            });

            when(worker.fenceZombies(eq(connector), eq(taskCountRecords.get(connector)), any())).thenReturn(workerFencingFuture);
        });

        tasksPerConnector.forEach((connector, taskCount) -> doNothing().when(configBackingStore).putTaskCountRecord(eq(connector), eq(taskCount)));

        expectHerderShutdown();

        startBackgroundHerder();

        List<FutureCallback<Void>> stackedFencingRequests = new ArrayList<>();
        tasksPerConnector.forEach((connector, numStackedRequests) -> {
            List<FutureCallback<Void>> connectorFencingRequests = IntStream.range(0, numStackedRequests)
                    .mapToObj(i -> new FutureCallback<Void>())
                    .collect(Collectors.toList());

            connectorFencingRequests.forEach(fencing ->
                    herder.fenceZombieSourceTasks(connector, fencing)
            );

            stackedFencingRequests.addAll(connectorFencingRequests);
        });

        callbacksInstalled.forEach((connector, latch) -> {
            try {
                assertTrue(latch.await(10, TimeUnit.SECONDS));
                workerFencingFollowups.get(connector).getValue().apply(null);
                herderFencingCallbacks.get(connector).getAllValues().forEach(cb -> cb.accept(null, null));
            } catch (InterruptedException e) {
                fail("Unexpectedly interrupted");
            }
        });

        for (FutureCallback<Void> fencing : stackedFencingRequests) {
            fencing.get(10, TimeUnit.SECONDS);
        }

        stopBackgroundHerder();

        // We should only perform a single physical zombie fencing for each connector; all the subsequent requests should be stacked onto the first one
        tasksPerConnector.keySet().forEach(connector -> verify(worker).fenceZombies(eq(connector), eq(taskCountRecords.get(connector)), any()));
        verifyNoMoreInteractions(member, worker, configBackingStore, statusBackingStore);
    }
Issues multiple rapid fencing requests for a handful of connectors, each of which takes a little while to complete.
This mimics what might happen when a few connectors are reconfigured in quick succession and each task for the
connector needs to hit the leader with a fencing request during its preflight check.
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java