@Test
    public void testModifyOffsetsSourceConnectorExactlyOnceEnabled() throws Exception {
        // 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 alter 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, null);
            return null;
        });

        ArgumentCaptor<Callback<Message>> workerCallbackCapture = ArgumentCaptor.forClass(Callback.class);
        Message msg = new Message("The offsets for this connector have been altered successfully");

        Map<Map<String, ?>, Map<String, ?>> offsets = Collections.singletonMap(
                Collections.singletonMap("partitionKey", "partitionValue"),
                Collections.singletonMap("offsetKey", "offsetValue"));
        doAnswer(invocation -> {
            workerCallbackCapture.getValue().onCompletion(null, msg);
            return null;
        }).when(worker).modifyConnectorOffsets(eq(CONN1), eq(CONN1_CONFIG), eq(offsets), workerCallbackCapture.capture());

        FutureCallback<Message> callback = new FutureCallback<>();
        herder.alterConnectorOffsets(CONN1, offsets, callback);
        // Process the zombie fencing request that is queued up first followed by the actual alter offsets request
        herder.tick();
        assertEquals(msg, callback.get(1000L, TimeUnit.MILLISECONDS));

        // Handle the second alter connector offsets request
        expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1_FENCED);
        FutureCallback<Message> callback2 = new FutureCallback<>();
        herder.alterConnectorOffsets(CONN1, offsets, callback2);
        herder.tick();
        assertEquals(msg, callback2.get(1000L, TimeUnit.MILLISECONDS));

        // 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 alter offsets request if the zombie fencing succeeds.
        verify(herderFencingFuture, times(2)).whenComplete(any());

        // No zombie fencing request to the worker is expected in the second alter connector offsets request since we already
        // did a round of zombie fencing the first time and no new tasks came up in the meanwhile.
        verify(worker, times(1)).fenceZombies(eq(CONN1), eq(SNAPSHOT_STOPPED_CONN1.taskCountRecord(CONN1)), eq(CONN1_CONFIG));
        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