@Test
    public void testVerifyTaskGeneration() {
        Map<String, Integer> taskConfigGenerations = new HashMap<>();
        herder.configState = new ClusterConfigState(
                1,
                null,
                Collections.singletonMap(CONN1, 3),
                Collections.singletonMap(CONN1, CONN1_CONFIG),
                Collections.singletonMap(CONN1, TargetState.STARTED),
                TASK_CONFIGS_MAP,
                Collections.emptyMap(),
                taskConfigGenerations,
                Collections.singletonMap(CONN1, new AppliedConnectorConfig(CONN1_CONFIG)),
                Collections.emptySet(),
                Collections.emptySet());

        Callback<Void> verifyCallback = mock(Callback.class);

        herder.assignment = new ExtendedAssignment(
                (short) 2, (short) 0, "leader", "leaderUrl", 0,
                Collections.emptySet(), Collections.singleton(TASK1),
                Collections.emptySet(), Collections.emptySet(), 0);

        assertThrows(ConnectException.class, () -> herder.verifyTaskGenerationAndOwnership(TASK1, 0, verifyCallback));
        assertThrows(ConnectException.class, () -> herder.verifyTaskGenerationAndOwnership(TASK1, 1, verifyCallback));
        assertThrows(ConnectException.class, () -> herder.verifyTaskGenerationAndOwnership(TASK1, 2, verifyCallback));

        taskConfigGenerations.put(CONN1, 0);
        herder.verifyTaskGenerationAndOwnership(TASK1, 0, verifyCallback);
        assertThrows(ConnectException.class, () -> herder.verifyTaskGenerationAndOwnership(TASK1, 1, verifyCallback));
        assertThrows(ConnectException.class, () -> herder.verifyTaskGenerationAndOwnership(TASK1, 2, verifyCallback));

        taskConfigGenerations.put(CONN1, 1);
        assertThrows(ConnectException.class, () -> herder.verifyTaskGenerationAndOwnership(TASK1, 0, verifyCallback));
        herder.verifyTaskGenerationAndOwnership(TASK1, 1, verifyCallback);
        assertThrows(ConnectException.class, () -> herder.verifyTaskGenerationAndOwnership(TASK1, 2, verifyCallback));

        taskConfigGenerations.put(CONN1, 2);
        assertThrows(ConnectException.class, () -> herder.verifyTaskGenerationAndOwnership(TASK1, 0, verifyCallback));
        assertThrows(ConnectException.class, () -> herder.verifyTaskGenerationAndOwnership(TASK1, 1, verifyCallback));
        herder.verifyTaskGenerationAndOwnership(TASK1, 2, verifyCallback);

        taskConfigGenerations.put(CONN1, 3);
        assertThrows(ConnectException.class, () -> herder.verifyTaskGenerationAndOwnership(TASK1, 0, verifyCallback));
        assertThrows(ConnectException.class, () -> herder.verifyTaskGenerationAndOwnership(TASK1, 1, verifyCallback));
        assertThrows(ConnectException.class, () -> herder.verifyTaskGenerationAndOwnership(TASK1, 2, verifyCallback));

        ConnectorTaskId unassignedTask = new ConnectorTaskId(CONN2, 0);
        taskConfigGenerations.put(unassignedTask.connector(), 1);
        assertThrows(ConnectException.class, () -> herder.verifyTaskGenerationAndOwnership(unassignedTask, 0, verifyCallback));
        assertThrows(ConnectException.class, () -> herder.verifyTaskGenerationAndOwnership(unassignedTask, 1, verifyCallback));
        assertThrows(ConnectException.class, () -> herder.verifyTaskGenerationAndOwnership(unassignedTask, 2, verifyCallback));

        verify(verifyCallback, times(3)).onCompletion(isNull(), isNull());
    }
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