public static DeadLetterQueueReporter createAndSetup(Map<String, Object> adminProps,
                                                         ConnectorTaskId id,
                                                         SinkConnectorConfig sinkConfig, Map<String, Object> producerProps,
                                                         ErrorHandlingMetrics errorHandlingMetrics) {
        String topic = sinkConfig.dlqTopicName();

        try (Admin admin = Admin.create(adminProps)) {
            if (!admin.listTopics().names().get().contains(topic)) {
                log.error("Topic {} doesn't exist. Will attempt to create topic.", topic);
                NewTopic schemaTopicRequest = new NewTopic(topic, DLQ_NUM_DESIRED_PARTITIONS, sinkConfig.dlqTopicReplicationFactor());
                admin.createTopics(singleton(schemaTopicRequest)).all().get();
            }
        } catch (InterruptedException e) {
            throw new ConnectException("Could not initialize dead letter queue with topic=" + topic, e);
        } catch (ExecutionException e) {
            if (!(e.getCause() instanceof TopicExistsException)) {
                throw new ConnectException("Could not initialize dead letter queue with topic=" + topic, e);
            }
        }

        KafkaProducer<byte[], byte[]> dlqProducer = new KafkaProducer<>(producerProps);
        return new DeadLetterQueueReporter(dlqProducer, sinkConfig, id, errorHandlingMetrics);
    }
Write the original consumed record into a dead letter queue. The dead letter queue is a Kafka topic located
on the same cluster used by the worker to maintain internal topics. Each connector is typically configured
with its own Kafka topic dead letter queue. By default, the topic name is not set, and if the
connector config doesn't specify one, this feature is disabled.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/DeadLetterQueueReporter.java