@Test
    public void testExternalZombieFencingRequestSynchronousFailure() 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();

        Exception fencingException = new KafkaException("whoops!");
        when(worker.fenceZombies(eq(CONN1), eq(2), eq(CONN1_CONFIG))).thenThrow(fencingException);

        expectHerderShutdown();

        startBackgroundHerder();

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

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

        stopBackgroundHerder();

        verifyNoMoreInteractions(member, worker, configBackingStore, statusBackingStore);
    }
The herder tries to perform a round of fencing, but fails synchronously while invoking Worker::fenceZombies
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java