private void stopTask(ConnectorTaskId taskId) {
        try (LoggingContext loggingContext = LoggingContext.forTask(taskId)) {
            WorkerTask<?, ?> task = tasks.get(taskId);
            if (task == null) {
                log.warn("Ignoring stop request for unowned task {}", taskId);
                return;
            }

            log.info("Stopping task {}", task.id());
            if (task instanceof WorkerSourceTask)
                sourceTaskOffsetCommitter.ifPresent(committer -> committer.remove(task.id()));

            try (LoaderSwap loaderSwap = plugins.withClassLoader(task.loader())) {
                task.stop();
            }
        }
    }
@return the {@link ProducerConfig#TRANSACTIONAL_ID_CONFIG transactional ID} to use for a task that writes
records and/or offsets in a transaction. Not to be confused with {@link DistributedConfig#transactionalProducerId()},
which is not used by tasks at all, but instead, by the worker itself.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java