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);
        }
    }
Request task configs from the connector and write them to the config storage in case the configs are detected to
have changed. This method retries infinitely with exponential backoff in case of any errors.

@param initialRequestTime the time in milliseconds when the original request was made (i.e. before any retries)
@param connName the name of the connector
@param exponentialBackoff {@link ExponentialBackoff} used to calculate the retry backoff duration
@param attempts the number of retry attempts that have been made
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java