private void expectRebalance(final Collection<String> revokedConnectors,
                                 final List<ConnectorTaskId> revokedTasks,
                                 final short error,
                                 final long offset,
                                 String leader,
                                 String leaderUrl,
                                 final List<String> assignedConnectors,
                                 final List<ConnectorTaskId> assignedTasks,
                                 int delay,
                                 boolean isLeader) {
        ArgumentCaptor<Supplier<UncheckedCloseable>> onPoll = ArgumentCaptor.forClass(Supplier.class);
        doAnswer(invocation -> {
            onPoll.getValue().get().close();
            ExtendedAssignment assignment;
            if (!revokedConnectors.isEmpty() || !revokedTasks.isEmpty()) {
                rebalanceListener.onRevoked(leader, revokedConnectors, revokedTasks);
            }

            if (connectProtocolVersion == CONNECT_PROTOCOL_V0) {
                assignment = new ExtendedAssignment(
                        connectProtocolVersion, error, leader, leaderUrl, offset,
                        assignedConnectors, assignedTasks,
                        Collections.emptyList(), Collections.emptyList(), 0);
            } else {
                assignment = new ExtendedAssignment(
                        connectProtocolVersion, error, leader, leaderUrl, offset,
                        assignedConnectors, assignedTasks,
                        new ArrayList<>(revokedConnectors), new ArrayList<>(revokedTasks), delay);
            }
            rebalanceListener.onAssigned(assignment, 3);
            time.sleep(100L);
            return null;
        }).when(member).ensureActive(onPoll.capture());

        if (isLeader) {
            doNothing().when(configBackingStore).claimWritePrivileges();
        }

        if (!revokedConnectors.isEmpty()) {
            for (String connector : revokedConnectors) {
                doNothing().when(worker).stopAndAwaitConnector(connector);
            }
        }

        if (!revokedTasks.isEmpty()) {
            doNothing().when(worker).stopAndAwaitTask(any(ConnectorTaskId.class));
        }

        if (!revokedConnectors.isEmpty()) {
            doNothing().when(statusBackingStore).flush();
        }

        doNothing().when(member).wakeup();
    }
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