@Test
    public void testExternalZombieFencingRequestAsynchronousFailure() 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);
        // The callbacks that the herder has accrued for outstanding fencing futures
        ArgumentCaptor<KafkaFuture.BiConsumer<Void, Throwable>> herderFencingCallbacks = ArgumentCaptor.forClass(KafkaFuture.BiConsumer.class);

        when(worker.fenceZombies(eq(CONN1), eq(2), eq(CONN1_CONFIG))).thenReturn(workerFencingFuture);

        when(workerFencingFuture.thenApply(any(KafkaFuture.BaseFunction.class))).thenReturn(herderFencingFuture);

        CountDownLatch callbacksInstalled = new CountDownLatch(2);
        when(herderFencingFuture.whenComplete(herderFencingCallbacks.capture())).thenAnswer(invocation -> {
            callbacksInstalled.countDown();
            return null;
        });

        expectHerderShutdown();

        startBackgroundHerder();

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

        assertTrue(callbacksInstalled.await(10, TimeUnit.SECONDS));

        Exception fencingException = new AuthorizationException("you didn't say the magic word");
        herderFencingCallbacks.getAllValues().forEach(cb -> cb.accept(null, fencingException));

        ExecutionException exception = assertThrows(ExecutionException.class, () -> fencing.get(10, TimeUnit.SECONDS));
        assertInstanceOf(ConnectException.class, exception.getCause());

        stopBackgroundHerder();

        verifyNoMoreInteractions(herderFencingFuture, workerFencingFuture, member, worker, configBackingStore, statusBackingStore);
    }
The herder tries to perform a round of fencing and is able to retrieve a future from worker::fenceZombies, but the attempt
fails at a later point.
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java