MessagePublisherImpl implements MessagePublisher<T> { @Override public void publish(String key, T value) { publish(topic, key, value); } MessagePublisherImpl(MessageProducer producer, String topic, Class<T> messageClass); @Override void publish(String key, T value); @Override void publish(String topic, String key, T value);  }
@Test void publish() { ActionLog actionLog = logManager.begin("begin", null); actionLog.correlationIds = List.of("correlationId"); var message = new TestMessage(); message.stringField = "value"; publisher.publish(message); verify(producer).send(argThat(record -> { assertThat(record.key()).isNull(); assertThat(new String(record.headers().lastHeader(MessageHeaders.HEADER_CORRELATION_ID).value(), UTF_8)).isEqualTo("correlationId"); assertThat(new String(record.headers().lastHeader(MessageHeaders.HEADER_REF_ID).value(), UTF_8)).isEqualTo(actionLog.id); return true; })); publisher.publish("topic", "key", message); verify(producer).send(argThat(record -> Arrays.equals(Strings.bytes("key"), record.key()) && "topic".equals(record.topic()))); logManager.end("end"); }