diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index e94f76a21..ced216b27 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -136,6 +136,7 @@ Default: See individual producer properties. spring.cloud.stream.kafka.binder.headerMapperBeanName:: The bean name of a `KafkaHeaderMapper` used for mapping `spring-messaging` headers to and from Kafka headers. Use this, for example, if you wish to customize the trusted packages in a `DefaultKafkaHeaderMapper` that uses JSON deserialization for the headers. +If this custom `KafkaHeaderMapper` bean is not made available to the binder using this property, then the binder will look for a header mapper bean with the name `kafkaBinderHeaderMapper` before falling back to a default header mapper. + Default: none. diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 0cfd8899c..c5306ee56 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -51,6 +51,7 @@ import org.apache.kafka.common.header.internals.RecordHeaders; import org.apache.kafka.common.serialization.ByteArrayDeserializer; import org.apache.kafka.common.serialization.ByteArraySerializer; +import org.springframework.beans.BeansException; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.NoSuchBeanDefinitionException; import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder; @@ -377,12 +378,23 @@ public class KafkaMessageChannelBinder extends if (StringUtils.hasText(producerProperties.getExtension().getRecordMetadataChannel())) { handler.setSendSuccessChannelName(producerProperties.getExtension().getRecordMetadataChannel()); } + KafkaHeaderMapper mapper = null; if (this.configurationProperties.getHeaderMapperBeanName() != null) { mapper = getApplicationContext().getBean( this.configurationProperties.getHeaderMapperBeanName(), KafkaHeaderMapper.class); } + if (mapper == null) { + //First, try to see if there is a bean named headerMapper registered by other frameworks using the binder (for e.g. spring cloud sleuth) + try { + mapper = getApplicationContext().getBean("kafkaBinderHeaderMapper", KafkaHeaderMapper.class); + } + catch (BeansException be) { + // Pass through + } + } + /* * Even if the user configures a bean, we must not use it if the header mode is * not the default (headers); setting the mapper to null disables populating @@ -899,23 +911,29 @@ public class KafkaMessageChannelBinder extends KafkaHeaderMapper.class); } if (mapper == null) { - DefaultKafkaHeaderMapper headerMapper = new DefaultKafkaHeaderMapper() { - - @Override - public void toHeaders(Headers source, Map headers) { - super.toHeaders(source, headers); - if (headers.size() > 0) { - headers.put(BinderHeaders.NATIVE_HEADERS_PRESENT, Boolean.TRUE); - } - } - - }; - String[] trustedPackages = extendedConsumerProperties.getExtension() - .getTrustedPackages(); - if (!StringUtils.isEmpty(trustedPackages)) { - headerMapper.addTrustedPackages(trustedPackages); + //First, try to see if there is a bean named headerMapper registered by other frameworks using the binder (for e.g. spring cloud sleuth) + try { + mapper = getApplicationContext().getBean("kafkaBinderHeaderMapper", KafkaHeaderMapper.class); + } + catch (BeansException be) { + DefaultKafkaHeaderMapper headerMapper = new DefaultKafkaHeaderMapper() { + + @Override + public void toHeaders(Headers source, Map headers) { + super.toHeaders(source, headers); + if (headers.size() > 0) { + headers.put(BinderHeaders.NATIVE_HEADERS_PRESENT, Boolean.TRUE); + } + } + + }; + String[] trustedPackages = extendedConsumerProperties.getExtension() + .getTrustedPackages(); + if (!StringUtils.isEmpty(trustedPackages)) { + headerMapper.addTrustedPackages(trustedPackages); + } + mapper = headerMapper; } - mapper = headerMapper; } return mapper; } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 7fd451cb4..836067681 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -52,6 +52,8 @@ import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.KafkaFuture; import org.apache.kafka.common.errors.TopicExistsException; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeader; import org.apache.kafka.common.record.TimestampType; import org.apache.kafka.common.serialization.ByteArrayDeserializer; import org.apache.kafka.common.serialization.ByteArraySerializer; @@ -114,6 +116,7 @@ import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.listener.MessageListenerContainer; import org.springframework.kafka.support.Acknowledgment; +import org.springframework.kafka.support.KafkaHeaderMapper; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.SendResult; import org.springframework.kafka.support.TopicPartitionOffset; @@ -133,7 +136,6 @@ import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; -import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.SettableListenableFuture; @@ -316,7 +318,7 @@ public class KafkaBinderTests extends @SuppressWarnings({ "rawtypes", "unchecked" }) @Test - public void testTrustedPackages() throws Exception { + public void testDefaultHeaderMapper() throws Exception { Binder binder = getBinder(); BindingProperties producerBindingProperties = createProducerBindingProperties( @@ -326,7 +328,7 @@ public class KafkaBinderTests extends ExtendedConsumerProperties consumerProperties = createConsumerProperties(); consumerProperties.getExtension() - .setTrustedPackages(new String[] { "org.springframework.util" }); + .setTrustedPackages(new String[] { "org.springframework.cloud.stream.binder.kafka" }); DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); @@ -339,9 +341,92 @@ public class KafkaBinderTests extends consumerProperties); binderBindUnbindLatency(); + final Pojo pojoHeader = new Pojo("testing"); Message message = org.springframework.integration.support.MessageBuilder .withPayload("foo") .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN) + .setHeader("foo", pojoHeader).build(); + + moduleOutputChannel.send(message); + CountDownLatch latch = new CountDownLatch(1); + AtomicReference> inboundMessageRef = new AtomicReference<>(); + moduleInputChannel.subscribe(message1 -> { + try { + inboundMessageRef.set((Message) message1); + } + finally { + latch.countDown(); + } + }); + Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); + + Assertions.assertThat(inboundMessageRef.get()).isNotNull(); + Assertions.assertThat(inboundMessageRef.get().getPayload()) + .isEqualTo("foo".getBytes()); + Assertions + .assertThat(inboundMessageRef.get().getHeaders() + .get(MessageHeaders.CONTENT_TYPE)) + .isEqualTo(MimeTypeUtils.TEXT_PLAIN); + Assertions.assertThat(inboundMessageRef.get().getHeaders().get("foo")) + .isInstanceOf(Pojo.class); + Pojo actual = (Pojo) inboundMessageRef.get().getHeaders().get("foo"); + Assertions.assertThat(actual.field).isEqualTo(pojoHeader.field); + producerBinding.unbind(); + consumerBinding.unbind(); + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Test + public void testCustomHeaderMapper() throws Exception { + + KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); + binderConfiguration.setHeaderMapperBeanName("headerMapper"); + + KafkaTopicProvisioner kafkaTopicProvisioner = new KafkaTopicProvisioner( + binderConfiguration, new TestKafkaProperties()); + try { + kafkaTopicProvisioner.afterPropertiesSet(); + } + catch (Exception e) { + throw new RuntimeException(e); + } + KafkaTestBinder binder = new KafkaTestBinder(binderConfiguration, kafkaTopicProvisioner); + ((GenericApplicationContext) binder.getApplicationContext()).registerBean("headerMapper", + KafkaHeaderMapper.class, () -> new KafkaHeaderMapper() { + @Override + public void fromHeaders(MessageHeaders headers, Headers target) { + target.add(new RecordHeader("custom-header", "foobar".getBytes())); + } + + @Override + public void toHeaders(Headers source, Map target) { + if (source.headers("custom-header").iterator().hasNext()) { + target.put("custom-header", source.headers("custom-header").iterator().next().value()); + } + + } + }); + + ExtendedProducerProperties producerProperties = createProducerProperties(); + + DirectChannel moduleOutputChannel = createBindableChannel("output", + createProducerBindingProperties(producerProperties)); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + + DirectChannel moduleInputChannel = createBindableChannel("input", + createConsumerBindingProperties(consumerProperties)); + + Binding producerBinding = binder.bindProducer("bar.0", + moduleOutputChannel, producerProperties); + + Binding consumerBinding = binder.bindConsumer("bar.0", + "testSendAndReceiveNoOriginalContentType", moduleInputChannel, + consumerProperties); + binderBindUnbindLatency(); + + Message message = org.springframework.integration.support.MessageBuilder + .withPayload("foo") .setHeader("foo", MimeTypeUtils.TEXT_PLAIN).build(); moduleOutputChannel.send(message); @@ -360,16 +445,87 @@ public class KafkaBinderTests extends Assertions.assertThat(inboundMessageRef.get()).isNotNull(); Assertions.assertThat(inboundMessageRef.get().getPayload()) .isEqualTo("foo".getBytes()); - Assertions.assertThat(inboundMessageRef.get().getHeaders() - .get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull(); - Assertions - .assertThat(inboundMessageRef.get().getHeaders() - .get(MessageHeaders.CONTENT_TYPE)) - .isEqualTo(MimeTypeUtils.TEXT_PLAIN); Assertions.assertThat(inboundMessageRef.get().getHeaders().get("foo")) - .isInstanceOf(MimeType.class); - MimeType actual = (MimeType) inboundMessageRef.get().getHeaders().get("foo"); - Assertions.assertThat(actual).isEqualTo(MimeTypeUtils.TEXT_PLAIN); + .isNull(); + Assertions.assertThat(inboundMessageRef.get().getHeaders().get("custom-header")) + .isEqualTo("foobar".getBytes()); + producerBinding.unbind(); + consumerBinding.unbind(); + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Test + public void testWellKnownHeaderMapperWithBeanNameKafkaHeaderMapper() throws Exception { + + KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); + + KafkaTopicProvisioner kafkaTopicProvisioner = new KafkaTopicProvisioner( + binderConfiguration, new TestKafkaProperties()); + try { + kafkaTopicProvisioner.afterPropertiesSet(); + } + catch (Exception e) { + throw new RuntimeException(e); + } + KafkaTestBinder binder = new KafkaTestBinder(binderConfiguration, kafkaTopicProvisioner); + ((GenericApplicationContext) binder.getApplicationContext()).registerBean("kafkaBinderHeaderMapper", + KafkaHeaderMapper.class, () -> new KafkaHeaderMapper() { + @Override + public void fromHeaders(MessageHeaders headers, Headers target) { + target.add(new RecordHeader("custom-header", "foobar".getBytes())); + } + + @Override + public void toHeaders(Headers source, Map target) { + if (source.headers("custom-header").iterator().hasNext()) { + target.put("custom-header", source.headers("custom-header").iterator().next().value()); + } + + } + }); + + ExtendedProducerProperties producerProperties = createProducerProperties(); + + DirectChannel moduleOutputChannel = createBindableChannel("output", + createProducerBindingProperties(producerProperties)); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + + DirectChannel moduleInputChannel = createBindableChannel("input", + createConsumerBindingProperties(consumerProperties)); + + Binding producerBinding = binder.bindProducer("bar.0", + moduleOutputChannel, producerProperties); + + Binding consumerBinding = binder.bindConsumer("bar.0", + "testSendAndReceiveNoOriginalContentType", moduleInputChannel, + consumerProperties); + binderBindUnbindLatency(); + + Message message = org.springframework.integration.support.MessageBuilder + .withPayload("foo") + .setHeader("foo", MimeTypeUtils.TEXT_PLAIN).build(); + + moduleOutputChannel.send(message); + CountDownLatch latch = new CountDownLatch(1); + AtomicReference> inboundMessageRef = new AtomicReference<>(); + moduleInputChannel.subscribe(message1 -> { + try { + inboundMessageRef.set((Message) message1); + } + finally { + latch.countDown(); + } + }); + Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); + + Assertions.assertThat(inboundMessageRef.get()).isNotNull(); + Assertions.assertThat(inboundMessageRef.get().getPayload()) + .isEqualTo("foo".getBytes()); + Assertions.assertThat(inboundMessageRef.get().getHeaders().get("foo")) + .isNull(); + Assertions.assertThat(inboundMessageRef.get().getHeaders().get("custom-header")) + .isEqualTo("foobar".getBytes()); producerBinding.unbind(); consumerBinding.unbind(); }