@Test
    public void testModifyOffsetsConnectorNotInStoppedState() {
        // 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);

        herder.tick();

        // Now handle the connector offsets modification request
        FutureCallback<Message> callback = new FutureCallback<>();
        herder.modifyConnectorOffsets(CONN1, null, callback);
        herder.tick();
        ExecutionException e = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
        assertInstanceOf(BadRequestException.class, e.getCause());
    }
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