### Question:
DefaultKafkaMessageConverter implements KafkaMessageConverter<String, byte[]> { @Override public Optional<EventMessage<?>> readKafkaMessage(ConsumerRecord<String, byte[]> consumerRecord) { try { Headers headers = consumerRecord.headers(); if (isAxonMessage(headers)) { byte[] messageBody = consumerRecord.value(); SerializedMessage<?> message = extractSerializedMessage(headers, messageBody); return buildMessage(headers, message); } } catch (Exception e) { logger.trace("Error converting ConsumerRecord [{}] to an EventMessage", consumerRecord, e); } return Optional.empty(); } @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:
@SuppressWarnings("unchecked") @Test public void testReadingMessage_WhenKafkaReturnNullHeaders_ShouldReturnEmptyMessage() { ConsumerRecord source = mock(ConsumerRecord.class); when(source.headers()).thenReturn(null); assertThat(testSubject.readKafkaMessage(source).isPresent()).isFalse(); }
@SuppressWarnings("unchecked") @Test public void testReadingMessageMissingAxonHeaderShouldReturnEmptyMessage() { ConsumerRecord msgWithoutHeaders = new ConsumerRecord("foo", 0, 0, "abc", 1); assertThat(testSubject.readKafkaMessage(msgWithoutHeaders).isPresent()).isFalse(); }
@SuppressWarnings("unchecked") @Test public void testReadingMessagePayloadDifferentThanByteShouldReturnEmptyMessage() { EventMessage<Object> eventMessage = eventMessage(); SerializedObject serializedObject = mock(SerializedObject.class); when(serializedObject.getType()).thenReturn(new SimpleSerializedType("foo", null)); Headers headers = toHeaders(eventMessage, serializedObject, byteMapper()); ConsumerRecord payloadDifferentThanByte = new ConsumerRecord( "foo", 0, 0, NO_TIMESTAMP, NO_TIMESTAMP_TYPE, -1L, NULL_SIZE, NULL_SIZE, 1, "123", headers ); assertThat(testSubject.readKafkaMessage(payloadDifferentThanByte).isPresent()).isFalse(); }