private List<ErrorReporter<ConsumerRecord<byte[], byte[]>>> sinkTaskReporters(ConnectorTaskId id, SinkConnectorConfig connConfig,
                                                     ErrorHandlingMetrics errorHandlingMetrics,
                                                     Class<? extends Connector> connectorClass) {
        ArrayList<ErrorReporter<ConsumerRecord<byte[], byte[]>>> reporters = new ArrayList<>();
        LogReporter<ConsumerRecord<byte[], byte[]>> logReporter = new LogReporter.Sink(id, connConfig, errorHandlingMetrics);
        reporters.add(logReporter);

        // check if topic for dead letter queue exists
        String topic = connConfig.dlqTopicName();
        if (topic != null && !topic.isEmpty()) {
            Map<String, Object> producerProps = baseProducerConfigs(id.connector(), "connector-dlq-producer-" + id, config, connConfig, connectorClass,
                                                                connectorClientConfigOverridePolicy, kafkaClusterId);
            Map<String, Object> adminProps = adminConfigs(id.connector(), "connector-dlq-adminclient-", config, connConfig, connectorClass, connectorClientConfigOverridePolicy, kafkaClusterId, ConnectorType.SINK);
            DeadLetterQueueReporter reporter = DeadLetterQueueReporter.createAndSetup(adminProps, id, connConfig, producerProps, errorHandlingMetrics);

            reporters.add(reporter);
        }

        return reporters;
    }
@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