void populateContextHeaders(ProducerRecord<byte[], byte[]> producerRecord, ProcessingContext<ConsumerRecord<byte[], byte[]>> context) {
        Headers headers = producerRecord.headers();
        if (context.original() != null) {
            headers.add(ERROR_HEADER_ORIG_TOPIC, toBytes(context.original().topic()));
            headers.add(ERROR_HEADER_ORIG_PARTITION, toBytes(context.original().partition()));
            headers.add(ERROR_HEADER_ORIG_OFFSET, toBytes(context.original().offset()));
        }

        headers.add(ERROR_HEADER_CONNECTOR_NAME, toBytes(connectorTaskId.connector()));
        headers.add(ERROR_HEADER_TASK_ID, toBytes(String.valueOf(connectorTaskId.task())));
        headers.add(ERROR_HEADER_STAGE, toBytes(context.stage().name()));
        headers.add(ERROR_HEADER_EXECUTING_CLASS, toBytes(context.executingClass().getName()));
        if (context.error() != null) {
            headers.add(ERROR_HEADER_EXCEPTION, toBytes(context.error().getClass().getName()));
            headers.add(ERROR_HEADER_EXCEPTION_MESSAGE, toBytes(context.error().getMessage()));
            byte[] trace;
            if ((trace = stacktrace(context.error())) != null) {
                headers.add(ERROR_HEADER_EXCEPTION_STACK_TRACE, trace);
            }
        }
    }
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