### Question:
DeadLetterTopicExceptionHandler implements DeserializationExceptionHandler { @Override public DeserializationHandlerResponse handle(final ProcessorContext context, final ConsumerRecord<byte[], byte[]> record, final Exception exception) { RecordCollector collector = ((ProcessorContextImpl) context).recordCollector(); Headers headers = record.headers(); headers.add(ExceptionHeader.ERROR_EXCEPTION_STACKTRACE, toByteArray(getStacktrace(exception))); headers.add(ExceptionHeader.ERROR_EXCEPTION_MESSAGE, toByteArray(exception.getMessage())); headers.add(ExceptionHeader.ERROR_EXCEPTION_CLASS_NAME, toByteArray(exception.getClass().getName())); headers.add(ExceptionHeader.ERROR_TIMESTAMP, toByteArray(Time.SYSTEM.milliseconds())); headers.add(ExceptionHeader.ERROR_APPLICATION_ID, toByteArray(applicationId)); headers.add(ExceptionHeader.ERROR_RECORD_TOPIC, toByteArray(record.topic())); headers.add(ExceptionHeader.ERROR_RECORD_PARTITION, toByteArray(record.partition())); headers.add(ExceptionHeader.ERROR_RECORD_OFFSET, toByteArray(record.offset())); customHeaders.forEach(headers::add); final String outputTopic = (config.outputTopic() != null) ? config.outputTopic() : record.topic() + OUTPUT_TOPIC_DEFAULT_SUFFIX; LOG.debug( "Sending rejected record from topic={}, partition={}, offset={} into topic {}", record.topic(), record.partition(), record.offset(), outputTopic ); if (internalProducer != null) { ProducerRecord<byte[], byte[]> producerRecord = new ProducerRecord<>( outputTopic, null, record.timestamp(), record.key(), record.value(), headers ); internalProducer.send(producerRecord, new Callback() { @Override public void onCompletion(final RecordMetadata metadata, final Exception exception) { LOG.error("Fail to send corrupted record into topic {}. Ignored record.", outputTopic, exception); } }); } else { collector.send(outputTopic, record.key(), record.value(), headers, record.timestamp(), serializer, serializer, null); } return getDeserializationHandlerResponse(exception); }  @Override void configure(final Map<String, ?> configs); @Override DeserializationHandlerResponse handle(final ProcessorContext context,
                                                 final ConsumerRecord<byte[], byte[]> record,
                                                 final Exception exception);  }### Answer:
@Test public void test() { StringSerializer serializer = new StringSerializer(); byte[] key = serializer.serialize(null, "test-key"); byte[] value = serializer.serialize(null, "test-value"); ConsumerRecord<byte[], byte[]> consumerRecord = new ConsumerRecord<>( "test-topic", 0, 1L, key, value); handler.handle(context, consumerRecord, new StreamsException(TEST_ERROR_MESSAGE)); Assertions.assertEquals("test-topic-rejected", collector.capturedTopic); Assertions.assertEquals(key, collector.capturedKey); Assertions.assertEquals(value, collector.capturedValue); Assertions.assertEquals(TEST_ERROR_MESSAGE, new String(collector.capturedHeaders.lastHeader(ExceptionHeader.ERROR_EXCEPTION_MESSAGE).value())); Assertions.assertEquals(StreamsException.class.getName(), new String(collector.capturedHeaders.lastHeader(ExceptionHeader.ERROR_EXCEPTION_CLASS_NAME).value())); }