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

GH-251: Configurable inbound adapter converter

Browse Source 
Fixes https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/251

When using native decoding, the message emitted by the binder has no `id` or `timestamp`
headers. This can cause problems if the first SI endpoint uses a JDBC message store,
for example.

Add the ability to configure whether these headers are populated as well as the ability
to inject a custom message converter.
This commit is contained in:
Gary Russell
2017-11-10 15:13:48 -05:00
committed by Soby Chacko
parent 5db3ede9c4
commit 77cbfe2858
4 changed files with 92 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java
View File

@@ -23,6 +23,7 @@ import java.util.Map;
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Soby Chacko
* @author Gary Russell
*
* <p>
* Thanks to Laszlo Szabo for providing the initial patch for generic property support.
@@ -30,6 +31,28 @@ import java.util.Map;
*/
public class KafkaConsumerProperties {
public enum StartOffset {
earliest(-2L),
latest(-1L);
private final long referencePoint;
StartOffset(long referencePoint) {
this.referencePoint = referencePoint;
}
public long getReferencePoint() {
return this.referencePoint;
}
}
public enum StandardHeaders {
none,
id,
timestamp,
both
}
private boolean autoRebalanceEnabled = true;
private boolean autoCommitOffset = true;
@@ -48,6 +71,10 @@ public class KafkaConsumerProperties {
private String[] trustedPackages;
private StandardHeaders standardHeaders = StandardHeaders.none;
private String converterBeanName;
private Map<String, String> configuration = new HashMap<>();
public boolean isAutoCommitOffset() {
@@ -98,21 +125,6 @@ public class KafkaConsumerProperties {
this.autoRebalanceEnabled = autoRebalanceEnabled;
}
public enum StartOffset {
earliest(-2L),
latest(-1L);
private final long referencePoint;
StartOffset(long referencePoint) {
this.referencePoint = referencePoint;
}
public long getReferencePoint() {
return this.referencePoint;
}
}
public Map<String, String> getConfiguration() {
return this.configuration;
}
@@ -144,4 +156,20 @@ public class KafkaConsumerProperties {
public void setDlqProducerProperties(KafkaProducerProperties dlqProducerProperties) {
this.dlqProducerProperties = dlqProducerProperties;
}
public StandardHeaders getStandardHeaders() {
return this.standardHeaders;
}
public void setStandardHeaders(StandardHeaders standardHeaders) {
this.standardHeaders = standardHeaders;
}
public String getConverterBeanName() {
return this.converterBeanName;
}
public void setConverterBeanName(String converterBeanName) {
this.converterBeanName = converterBeanName;
}
}

10
spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc
View File

@@ -191,6 +191,16 @@ dlqProducerProperties::
All the properties available through kafka producer properties can be set through this property.
+
Default: Default Kafka producer properties.
standardHeaders::
Indicates which standard headers are populated by the inbound channel adapter.
`none`, `id`, `timestamp` or `both`.
Useful if using native deserialization and the first component to receive a message needs an `id` (such as an aggregator that is configured to use a JDBC message store).
+
Default: `none`
converterBeanName::
The name of a bean that implements `RecordMessageConverter`; used in the inbound channel adapter to replace the default `MessagingMessageConverter`.
+
Default: `null`
[[kafka-producer-properties]]
=== Kafka Producer Properties

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

@@ -43,6 +43,7 @@ import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
import org.springframework.cloud.stream.binder.BinderHeaders;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
@@ -51,6 +52,7 @@ import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
import org.springframework.cloud.stream.binder.HeaderMode;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties.StandardHeaders;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBindingProperties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner;
@@ -373,7 +375,25 @@ public class KafkaMessageChannelBinder extends
}
final KafkaMessageDrivenChannelAdapter<?, ?> kafkaMessageDrivenChannelAdapter = new KafkaMessageDrivenChannelAdapter<>(
messageListenerContainer);
MessagingMessageConverter messageConverter = new MessagingMessageConverter();
MessagingMessageConverter messageConverter;
if (extendedConsumerProperties.getExtension().getConverterBeanName() == null) {
messageConverter = new MessagingMessageConverter();
StandardHeaders standardHeaders = extendedConsumerProperties.getExtension().getStandardHeaders();
messageConverter.setGenerateMessageId(StandardHeaders.id.equals(standardHeaders)
|| StandardHeaders.both.equals(standardHeaders));
messageConverter.setGenerateTimestamp(StandardHeaders.timestamp.equals(standardHeaders)
|| StandardHeaders.both.equals(standardHeaders));
}
else {
try {
messageConverter = getApplicationContext().getBean(
extendedConsumerProperties.getExtension().getConverterBeanName(),
MessagingMessageConverter.class);
}
catch (NoSuchBeanDefinitionException e) {
throw new IllegalStateException("Converter bean not present in application context", e);
}
}
KafkaHeaderMapper mapper = null;
if (this.configurationProperties.getHeaderMapperBeanName() != null) {
mapper = getApplicationContext().getBean(this.configurationProperties.getHeaderMapperBeanName(),

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

@@ -74,11 +74,13 @@ import org.springframework.cloud.stream.binder.TestUtils;
import org.springframework.cloud.stream.binder.kafka.admin.KafkaAdminUtilsOperation;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties.StandardHeaders;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
import org.springframework.cloud.stream.binder.kafka.utils.KafkaTopicUtils;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.provisioning.ProvisioningException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.channel.DirectChannel;
@@ -96,6 +98,7 @@ import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.SendResult;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.kafka.support.converter.MessagingMessageConverter;
import org.springframework.kafka.test.core.BrokerAddress;
import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.kafka.test.utils.KafkaTestUtils;
@@ -1991,7 +1994,7 @@ public class KafkaBinderTests extends
}
@Test
@SuppressWarnings("unchecked")
@SuppressWarnings({ "unchecked", "rawtypes" })
public void testNativeSerializationWithCustomSerializerDeserializer() throws Exception {
Binding<?> producerBinding = null;
Binding<?> consumerBinding = null;
@@ -2019,6 +2022,7 @@ public class KafkaBinderTests extends
consumerProperties.getExtension().setAutoRebalanceEnabled(false);
consumerProperties.getExtension().getConfiguration().put("value.deserializer",
"org.apache.kafka.common.serialization.IntegerDeserializer");
consumerProperties.getExtension().setStandardHeaders(StandardHeaders.both);
consumerBinding = binder.bindConsumer(testTopicName, "test", moduleInputChannel, consumerProperties);
// Let the consumer actually bind to the producer before sending a msg
binderBindUnbindLatency();
@@ -2027,6 +2031,8 @@ public class KafkaBinderTests extends
assertThat(inbound).isNotNull();
assertThat(inbound.getPayload()).isEqualTo(10);
assertThat(inbound.getHeaders()).doesNotContainKey("contentType");
assertThat(inbound.getHeaders().getId()).isNotNull();
assertThat(inbound.getHeaders().getTimestamp()).isNotNull();
}
finally {
if (producerBinding != null) {
@@ -2039,7 +2045,7 @@ public class KafkaBinderTests extends
}
@Test
@SuppressWarnings("unchecked")
@SuppressWarnings({ "unchecked", "rawtypes" })
public void testNativeSerializationWithCustomSerializerDeserializerBytesPayload() throws Exception {
Binding<?> producerBinding = null;
Binding<?> consumerBinding = null;
@@ -2059,6 +2065,12 @@ public class KafkaBinderTests extends
invokeCreateTopic(zkUtils, testTopicName, 1, 1, new Properties());
configurationProperties.setAutoAddPartitions(true);
Binder binder = getBinder(configurationProperties);
ConfigurableApplicationContext context = TestUtils.getPropertyValue(binder, "binder.applicationContext",
ConfigurableApplicationContext.class);
MessagingMessageConverter converter = new MessagingMessageConverter();
converter.setGenerateMessageId(true);
converter.setGenerateTimestamp(true);
context.getBeanFactory().registerSingleton("testConverter", converter);
QueueChannel moduleInputChannel = new QueueChannel();
ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
producerProperties.setUseNativeEncoding(true);
@@ -2071,6 +2083,7 @@ public class KafkaBinderTests extends
consumerProperties.getExtension()
.getConfiguration()
.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
consumerProperties.getExtension().setConverterBeanName("testConverter");
consumerBinding = binder.bindConsumer(testTopicName, "test", moduleInputChannel, consumerProperties);
// Let the consumer actually bind to the producer before sending a msg
binderBindUnbindLatency();
@@ -2080,6 +2093,8 @@ public class KafkaBinderTests extends
assertThat(inbound.getPayload()).isEqualTo(new byte[1]);
assertThat(inbound.getHeaders()).containsKey("contentType");
assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()).isEqualTo("something/funky");
assertThat(inbound.getHeaders().getId()).isNotNull();
assertThat(inbound.getHeaders().getTimestamp()).isNotNull();
}
finally {
if (producerBinding != null) {
@@ -2270,7 +2285,7 @@ public class KafkaBinderTests extends
}
@Test
@SuppressWarnings("unchecked")
@SuppressWarnings({ "unchecked", "rawtypes" })
public void testSendAndReceiveWithRawMode() throws Exception {
Binder binder = getBinder();