private void awaitStopTask(ConnectorTaskId taskId, long timeout) {
        try (LoggingContext loggingContext = LoggingContext.forTask(taskId)) {
            WorkerTask<?, ?> task = tasks.remove(taskId);
            if (task == null) {
                log.warn("Ignoring await stop request for non-present task {}", taskId);
                return;
            }

            if (!task.awaitStop(timeout)) {
                log.error("Graceful stop of task {} failed.", task.id());
                task.cancel();
            } else {
                log.debug("Graceful stop of task {} succeeded.", task.id());
            }

            try {
                task.removeMetrics();
            } finally {
                connectorStatusMetricsGroup.recordTaskRemoved(taskId);
            }
        }
    }
@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