private void verifyTaskGenerationAndOwnership(ConnectorTaskId id, int initialTaskGen) {
        log.debug("Reading to end of config topic to ensure it is still safe to bring up source task {} with exactly-once support", id);
        if (!refreshConfigSnapshot(Long.MAX_VALUE)) {
            throw new ConnectException("Failed to read to end of config topic");
        }

        FutureCallback<Void> verifyCallback = new FutureCallback<>();

        addRequest(
            () -> verifyTaskGenerationAndOwnership(id, initialTaskGen, verifyCallback),
            forwardErrorAndTickThreadStages(verifyCallback)
        );

        try {
            verifyCallback.get();
        } catch (InterruptedException e) {
            throw new ConnectException("Interrupted while performing preflight check for task " + id, e);
        } catch (ExecutionException e) {
            Throwable cause = e.getCause();
            throw ConnectUtils.maybeWrap(cause, "Failed to perform preflight check for task " + id);
        }
    }
@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());
    }