@Test
    public void testTaskReconfigurationNoRetryWithTooManyTasks() {
        // initial tick
        when(member.memberId()).thenReturn("leader");
        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
        when(worker.isSinkConnector(CONN1)).thenReturn(Boolean.TRUE);
        expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
        expectConfigRefreshAndSnapshot(SNAPSHOT);

        when(worker.isRunning(CONN1)).thenReturn(true);
        when(worker.getPlugins()).thenReturn(plugins);

        herder.tick();
        // No requests are queued, so we shouldn't plan on waking up without external action
        // (i.e., rebalance, user request, or shutdown)
        // This helps indicate that no retriable operations (such as generating task configs after
        // a backoff period) are queued up by the worker
        verify(member, times(1)).poll(eq(Long.MAX_VALUE), any());

        // Process the task reconfiguration request in this tick
        int numTasks = MAX_TASKS + 5;
        SinkConnectorConfig sinkConnectorConfig = new SinkConnectorConfig(plugins, CONN1_CONFIG);
        // Fail to generate tasks because the connector provided too many task configs
        when(worker.connectorTaskConfigs(CONN1, sinkConnectorConfig))
                .thenThrow(new TooManyTasksException(CONN1, numTasks, MAX_TASKS));

        herder.requestTaskReconfiguration(CONN1);
        herder.tick();
        // We tried to generate task configs for the connector one time during this tick
        verify(worker, times(1)).connectorTaskConfigs(CONN1, sinkConnectorConfig);
        // Verifying again that no requests are queued
        verify(member, times(2)).poll(eq(Long.MAX_VALUE), any());
        verifyNoMoreInteractions(worker);

        time.sleep(DistributedHerder.RECONFIGURE_CONNECTOR_TASKS_BACKOFF_MAX_MS);
        herder.tick();
        // We ticked one more time, and no further attempt was made to generate task configs
        verifyNoMoreInteractions(worker);
        // And we don't have any requests queued
        verify(member, times(3)).poll(eq(Long.MAX_VALUE), any());
    }
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