@SuppressWarnings("unchecked") @Test public void testReadingMessage_WhenKafkaReturnNullHeaders_ShouldReturnEmptyMessage() { ConsumerRecord source = mock(ConsumerRecord.class); when(source.headers()).thenReturn(null); assertThat(testSubject.readKafkaMessage(source).isPresent()).isFalse(); }
@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(); }
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(); } }
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); }
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); }
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);  }