public Future<RecordMetadata> report(ProcessingContext<ConsumerRecord<byte[], byte[]>> context) {
        if (dlqTopicName.isEmpty()) {
            return CompletableFuture.completedFuture(null);
        }
        errorHandlingMetrics.recordDeadLetterQueueProduceRequest();

        if (context.original() == null) {
            errorHandlingMetrics.recordDeadLetterQueueProduceFailed();
            return CompletableFuture.completedFuture(null);
        }
        ConsumerRecord<byte[], byte[]> originalMessage = context.original();

        ProducerRecord<byte[], byte[]> producerRecord;
        if (originalMessage.timestamp() == RecordBatch.NO_TIMESTAMP) {
            producerRecord = new ProducerRecord<>(dlqTopicName, null,
                    originalMessage.key(), originalMessage.value(), originalMessage.headers());
        } else {
            producerRecord = new ProducerRecord<>(dlqTopicName, null, originalMessage.timestamp(),
                    originalMessage.key(), originalMessage.value(), originalMessage.headers());
        }

        if (connConfig.isDlqContextHeadersEnabled()) {
            populateContextHeaders(producerRecord, context);
        }

        return this.kafkaProducer.send(producerRecord, (metadata, exception) -> {
            if (exception != null) {
                log.error("Could not produce message to dead letter queue. topic=" + dlqTopicName, exception);
                errorHandlingMetrics.recordDeadLetterQueueProduceFailed();
            }
        });
    }
Write the raw records into a Kafka topic and return the producer future.

@param context processing context containing the raw record at {@link ProcessingContext#original()}.
@return the future associated with the writing of this record; never null
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/DeadLetterQueueReporter.java