### Question:
DefaultKafkaMessageConverter implements KafkaMessageConverter<String, byte[]> { @Override public ProducerRecord<String, byte[]> createKafkaMessage(EventMessage<?> eventMessage, String topic) { SerializedObject<byte[]> serializedObject = eventMessage.serializePayload(serializer, byte[].class); return new ProducerRecord<>( topic, null, null, recordKey(eventMessage), serializedObject.getData(), toHeaders(eventMessage, serializedObject, headerValueMapper) ); } @SuppressWarnings("WeakerAccess") protected  DefaultKafkaMessageConverter(Builder builder); static Builder builder(); @Override ProducerRecord<String, byte[]> createKafkaMessage(EventMessage<?> eventMessage, String topic); @Override Optional<EventMessage<?>> readKafkaMessage(ConsumerRecord<String, byte[]> consumerRecord);  }### Answer:
@Test public void testKafkaKeyGenerationEventMessageShouldBeNull() { ProducerRecord<String, byte[]> evt = testSubject.createKafkaMessage(eventMessage(), SOME_TOPIC); assertThat(evt.key()).isNull(); }
@Test public void testKafkaKeyGenerationDomainMessageShouldBeAggregateIdentifier() { ProducerRecord<String, byte[]> domainEvt = testSubject.createKafkaMessage(domainMessage(), SOME_TOPIC); assertThat(domainEvt.key()).isEqualTo((domainMessage().getAggregateIdentifier())); }
@Test public void testWritingEventMessageAsKafkaMessageShouldAppendEventHeaders() { EventMessage<?> expected = eventMessage(); ProducerRecord<String, byte[]> senderMessage = testSubject.createKafkaMessage(expected, SOME_TOPIC); SerializedObject<byte[]> serializedObject = expected.serializePayload(serializer, byte[].class); assertEventHeaders("key", expected, serializedObject, senderMessage.headers()); }
@Test public void testWritingDomainMessageAsKafkaMessageShouldAppendDomainHeaders() { GenericDomainEventMessage<String> expected = domainMessage(); ProducerRecord<String, byte[]> senderMessage = testSubject.createKafkaMessage(expected, SOME_TOPIC); assertDomainHeaders(expected, senderMessage.headers()); }
@Test public void testWritingEventMessageShouldBeReadAsEventMessage() { EventMessage<?> expected = eventMessage(); ProducerRecord<String, byte[]> senderMessage = testSubject.createKafkaMessage(expected, SOME_TOPIC); EventMessage<?> actual = receiverMessage(senderMessage); assertEventMessage(actual, expected); }
@Test public void testWritingDomainEventMessageShouldBeReadAsDomainMessage() { DomainEventMessage<?> expected = domainMessage(); ProducerRecord<String, byte[]> senderMessage = testSubject.createKafkaMessage(expected, SOME_TOPIC); EventMessage<?> actual = receiverMessage(senderMessage); assertEventMessage(actual, expected); assertDomainMessage((DomainEventMessage<?>) actual, expected); }