SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Check HeaderMapper bean with a well known name (#753)

Browse Source 
* Check HeaderMapper bean with a well known name

* If a custom bean name for header mapper is not provided through the binder property headerMapperBeanName,
  then look for a header mapper bean with the name kafkaHeaderMapper.
* Add tests to verify

Resolves #749

* Addressing PR review comments
This commit is contained in:
Soby Chacko
2019-09-27 11:08:15 -04:00
committed by Gary Russell
parent e549090787
commit db5a303431
3 changed files with 203 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
docs/src/main/asciidoc/overview.adoc
View File

@@ -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.

50
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -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<String, Object> 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<String, Object> 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;
}

180
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

@@ -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<KafkaConsumerProperties> 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<Message<byte[]>> inboundMessageRef = new AtomicReference<>();
moduleInputChannel.subscribe(message1 -> {
try {
inboundMessageRef.set((Message<byte[]>) 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<String, Object> target) {
if (source.headers("custom-header").iterator().hasNext()) {
target.put("custom-header", source.headers("custom-header").iterator().next().value());
}
}
});
ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
DirectChannel moduleOutputChannel = createBindableChannel("output",
createProducerBindingProperties(producerProperties));
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
DirectChannel moduleInputChannel = createBindableChannel("input",
createConsumerBindingProperties(consumerProperties));
Binding<MessageChannel> producerBinding = binder.bindProducer("bar.0",
moduleOutputChannel, producerProperties);
Binding<MessageChannel> 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<String, Object> target) {
if (source.headers("custom-header").iterator().hasNext()) {
target.put("custom-header", source.headers("custom-header").iterator().next().value());
}
}
});
ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
DirectChannel moduleOutputChannel = createBindableChannel("output",
createProducerBindingProperties(producerProperties));
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
DirectChannel moduleInputChannel = createBindableChannel("input",
createConsumerBindingProperties(consumerProperties));
Binding<MessageChannel> producerBinding = binder.bindProducer("bar.0",
moduleOutputChannel, producerProperties);
Binding<MessageChannel> 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<Message<byte[]>> inboundMessageRef = new AtomicReference<>();
moduleInputChannel.subscribe(message1 -> {
try {
inboundMessageRef.set((Message<byte[]>) 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();
}