@Test public void testWritingDomainMessageAsKafkaMessageShouldAppendDomainHeaders() { GenericDomainEventMessage<String> expected = domainMessage(); ProducerRecord<String, byte[]> senderMessage = testSubject.createKafkaMessage(expected, SOME_TOPIC); assertDomainHeaders(expected, senderMessage.headers()); }
@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) ); }
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) ); } }
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); }
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); }
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);  }