@Test
    public void testModifyOffsetsSourceConnectorExactlyOnceEnabledZombieFencingFailure() {
        // Setup herder with exactly-once support for source connectors enabled
        herder = exactlyOnceHerder();
        rebalanceListener = herder.new RebalanceListener(time);

        // Get the initial assignment
        when(member.memberId()).thenReturn("leader");
        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
        expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
        expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
        expectMemberPoll();

        herder.tick();

        // Now handle the reset connector offsets request
        expectMemberEnsureActive();
        when(herder.connectorType(any())).thenReturn(ConnectorType.SOURCE);

        // Expect a round of zombie fencing to occur
        KafkaFuture<Void> workerFencingFuture = mock(KafkaFuture.class);
        KafkaFuture<Void> herderFencingFuture = mock(KafkaFuture.class);
        when(worker.fenceZombies(CONN1, SNAPSHOT_STOPPED_CONN1.taskCountRecord(CONN1), CONN1_CONFIG)).thenReturn(workerFencingFuture);
        when(workerFencingFuture.thenApply(any(KafkaFuture.BaseFunction.class))).thenReturn(herderFencingFuture);

        ArgumentCaptor<KafkaFuture.BiConsumer<Void, Throwable>> herderFencingCallback = ArgumentCaptor.forClass(KafkaFuture.BiConsumer.class);
        when(herderFencingFuture.whenComplete(herderFencingCallback.capture())).thenAnswer(invocation -> {
            herderFencingCallback.getValue().accept(null, new ConnectException("Failed to perform zombie fencing"));
            return null;
        });

        FutureCallback<Message> callback = new FutureCallback<>();
        herder.resetConnectorOffsets(CONN1, callback);
        // Process the zombie fencing request that is queued up first
        herder.tick();
        ExecutionException e = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
        assertEquals(ConnectException.class, e.getCause().getClass());
        assertEquals("Failed to perform zombie fencing for source connector prior to modifying offsets",
                e.getCause().getMessage());

        // Two fencing callbacks are added - one is in ZombieFencing::start itself to remove the connector from the active
        // fencing list. The other is the callback passed from DistributedHerder::modifyConnectorOffsets in order to
        // queue up the actual reset offsets request if the zombie fencing succeeds.
        verify(herderFencingFuture, times(2)).whenComplete(any());
        verifyNoMoreInteractions(workerFencingFuture, herderFencingFuture, member, worker);
    }
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