@Test
    public void testPollDurationOnSlowConnectorOperations() {
        connectProtocolVersion = CONNECT_PROTOCOL_V1;
        // If an operation during tick() takes some amount of time, that time should count against the rebalance delay
        final int rebalanceDelayMs = 20000;
        final long operationDelayMs = 10000;
        final long maxPollWaitMs = rebalanceDelayMs - operationDelayMs;
        when(member.memberId()).thenReturn("member");
        when(member.currentProtocolVersion()).thenReturn(connectProtocolVersion);
        when(worker.isSinkConnector(CONN1)).thenReturn(Boolean.TRUE);

        // Assign the connector to this worker, and have it start
        expectRebalance(Collections.emptyList(), Collections.emptyList(), ConnectProtocol.Assignment.NO_ERROR, 1, singletonList(CONN1), Collections.emptyList(), rebalanceDelayMs);
        expectConfigRefreshAndSnapshot(SNAPSHOT);
        ArgumentCaptor<Callback<TargetState>> onFirstStart = ArgumentCaptor.forClass(Callback.class);
        doAnswer(invocation -> {
            time.sleep(operationDelayMs);
            onFirstStart.getValue().onCompletion(null, TargetState.STARTED);
            return true;
        }).when(worker).startConnector(eq(CONN1), eq(CONN1_CONFIG), any(), eq(herder), eq(TargetState.STARTED), onFirstStart.capture());
        expectExecuteTaskReconfiguration(true, conn1SinkConfig, invocation -> TASK_CONFIGS);

        herder.tick();

        // Rebalance again due to config update
        expectRebalance(Collections.emptyList(), Collections.emptyList(), ConnectProtocol.Assignment.NO_ERROR, 1, singletonList(CONN1), Collections.emptyList(), rebalanceDelayMs);
        when(configBackingStore.snapshot()).thenReturn(SNAPSHOT_UPDATED_CONN1_CONFIG);
        doNothing().when(worker).stopAndAwaitConnector(CONN1);

        ArgumentCaptor<Callback<TargetState>> onSecondStart = ArgumentCaptor.forClass(Callback.class);
        doAnswer(invocation -> {
            time.sleep(operationDelayMs);
            onSecondStart.getValue().onCompletion(null, TargetState.STARTED);
            return true;
        }).when(worker).startConnector(eq(CONN1), eq(CONN1_CONFIG_UPDATED), any(), eq(herder), eq(TargetState.STARTED), onSecondStart.capture());
        expectExecuteTaskReconfiguration(true, conn1SinkConfigUpdated, invocation -> TASK_CONFIGS);

        configUpdateListener.onConnectorConfigUpdate(CONN1); // read updated config
        herder.tick();

        // Third tick should resolve all outstanding requests
        expectRebalance(Collections.emptyList(), Collections.emptyList(), ConnectProtocol.Assignment.NO_ERROR, 1, singletonList(CONN1), Collections.emptyList(), rebalanceDelayMs);
        // which includes querying the connector task configs after the update
        expectExecuteTaskReconfiguration(true, conn1SinkConfigUpdated, invocation -> {
            time.sleep(operationDelayMs);
            return TASK_CONFIGS;
        });
        herder.tick();

        // We should poll for less than the delay - time to start the connector, meaning that a long connector start
        // does not delay the poll timeout
        verify(member, times(3)).poll(leq(maxPollWaitMs), any());
        verify(worker, times(2)).startConnector(eq(CONN1), any(), any(), eq(herder), eq(TargetState.STARTED), any());
        verifyNoMoreInteractions(member, worker, configBackingStore);
    }
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