@Test
    public void testConnectorOffsets() throws Exception {
        when(member.memberId()).thenReturn("leader");
        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
        expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
        expectConfigRefreshAndSnapshot(SNAPSHOT);

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

        herder.tick();

        when(configBackingStore.snapshot()).thenReturn(SNAPSHOT);
        ConnectorOffsets offsets = new ConnectorOffsets(Collections.singletonList(new ConnectorOffset(
                Collections.singletonMap("partitionKey", "partitionValue"),
                Collections.singletonMap("offsetKey", "offsetValue"))));

        ArgumentCaptor<Callback<ConnectorOffsets>> callbackCapture = ArgumentCaptor.forClass(Callback.class);
        doAnswer(invocation -> {
            callbackCapture.getValue().onCompletion(null, offsets);
            return null;
        }).when(worker).connectorOffsets(eq(CONN1), eq(CONN1_CONFIG), callbackCapture.capture());

        FutureCallback<ConnectorOffsets> cb = new FutureCallback<>();
        herder.connectorOffsets(CONN1, cb);
        herder.tick();
        assertEquals(offsets, cb.get(1000, TimeUnit.MILLISECONDS));

        verifyNoMoreInteractions(worker, member, 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