KafkaController { ProducerRecord<byte[], byte[]> record(String topic, String key, byte[] body) { var record = new ProducerRecord<>(topic, Strings.bytes(key), body); Headers headers = record.headers(); headers.add(MessageHeaders.HEADER_CLIENT, Strings.bytes(KafkaController.class.getSimpleName())); headers.add(MessageHeaders.HEADER_TRACE, Strings.bytes("true")); ActionLog actionLog = LogManager.CURRENT_ACTION_LOG.get(); headers.add(MessageHeaders.HEADER_CORRELATION_ID, Strings.bytes(actionLog.correlationId())); headers.add(MessageHeaders.HEADER_REF_ID, Strings.bytes(actionLog.id)); return record; } KafkaController(MessageProducer producer); Response publish(Request request);  }
@Test void record() { ProducerRecord<byte[], byte[]> record = controller.record("topic", "key", new byte[0]); assertThat(record.headers().lastHeader(MessageHeaders.HEADER_TRACE).value()).isEqualTo(Strings.bytes("true")); }