Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-115: Use SK MessageConverter from MDCAdapter

Browse Source 
Fixes: GH-115 (https://github.com/spring-projects/spring-integration-kafka/issues/115)

* Use the `MessagingMessageListenerAdapter` with its `MessageConverter` capabilities
directly from the `KafkaMessageDrivenChannelAdapter` instead of the custom headers logic.
* Remove redundant properties in favor of `MessageConverter` injection
* Provide parser and `KafkaMessageDrivenChannelAdapter` tests to ensure that `MessageConverter` injection works properly
* Remove unnecessary testing POJOs
This commit is contained in:
Artem Bilan
2016-04-07 23:00:03 -04:00
committed by Artem Bilan
parent f9adcbd3ec
commit caae6487e6
10 changed files with 50 additions and 363 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java
View File

@@ -51,12 +51,7 @@ public class KafkaMessageDrivenChannelAdapterParser extends AbstractChannelAdapt
builder.addPropertyReference("outputChannel", channelName);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
"use-context-message-builder", "useMessageBuilderFactory");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
"set-id", "generateMessageId");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
"set-timestamp", "generateTimestamp");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
return builder.getBeanDefinition();
}

129
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java
View File

@@ -16,24 +16,15 @@
package org.springframework.integration.kafka.inbound;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.integration.context.OrderlyShutdownCapable;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.MutableMessageBuilderFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.converter.MessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
/**
@@ -44,77 +35,25 @@ import org.springframework.util.Assert;
*
* @author Marius Bogoevici
* @author Gary Russell
*
* TODO: Use the MessagingMessageConverter from spring-kafka
* @author Artem Bilan
*
*/
public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSupport implements OrderlyShutdownCapable {
private final AbstractMessageListenerContainer<K, V> messageListenerContainer;
private boolean generateMessageId = false;
private boolean generateTimestamp = false;
private boolean useMessageBuilderFactory = false;
private final MessagingMessageListenerAdapter<K, V> listener = new IntegrationMessageListener();
public KafkaMessageDrivenChannelAdapter(AbstractMessageListenerContainer<K, V> messageListenerContainer) {
Assert.notNull(messageListenerContainer, "messageListenerContainer is required");
Assert.isNull(messageListenerContainer.getMessageListener(), "Container must not already have a listener");
this.messageListenerContainer = messageListenerContainer;
this.messageListenerContainer.setAutoStartup(false);
this.messageListenerContainer.setMessageListener(this.listener);
}
/**
* Generate {@link Message} {@code ids} for produced messages. If set to {@code false}
* , will try to use a default value. By default set to {@code false}. Note that this
* option is only guaranteed to work when {@link #setUseMessageBuilderFactory(boolean)
* useMessageBuilderFactory} is false (default). If the latter is set to {@code true},
* then some {@link MessageBuilderFactory} implementations such as
* {@link DefaultMessageBuilderFactory} may ignore it.
* @param generateMessageId true if a message id should be generated
* @since 1.1
*/
public void setGenerateMessageId(boolean generateMessageId) {
this.generateMessageId = generateMessageId;
}
/**
* Generate {@code timestamp} for produced messages. If set to {@code false}, -1 is
* used instead. By default set to {@code false}. Note that this option is only
* guaranteed to work when {@link #setUseMessageBuilderFactory(boolean)
* useMessageBuilderFactory} is false (default). If the latter is set to {@code true},
* then some {@link MessageBuilderFactory} implementations such as
* {@link DefaultMessageBuilderFactory} may ignore it.
* @param generateTimestamp true if a timestamp should be generated
* @since 1.1
*/
public void setGenerateTimestamp(boolean generateTimestamp) {
this.generateTimestamp = generateTimestamp;
}
/**
* Use the {@link MessageBuilderFactory} returned by
* {@link #getMessageBuilderFactory()} to create messages.
* @param useMessageBuilderFactory true if the {@link MessageBuilderFactory} returned
* by {@link #getMessageBuilderFactory()} should be used.
* @since 1.1
*/
public void setUseMessageBuilderFactory(boolean useMessageBuilderFactory) {
this.useMessageBuilderFactory = useMessageBuilderFactory;
}
@Override
protected void onInit() {
this.messageListenerContainer.setMessageListener(
!AckMode.MANUAL.equals(this.messageListenerContainer.getAckMode())
? new AutoAcknowledgingChannelForwardingMessageListener()
: new AcknowledgingChannelForwardingMessageListener());
if (!this.generateMessageId && !this.generateTimestamp
&& (getMessageBuilderFactory() instanceof DefaultMessageBuilderFactory)) {
setMessageBuilderFactory(new MutableMessageBuilderFactory());
}
super.onInit();
public void setMessageConverter(MessageConverter messageConverter) {
this.listener.setMessageConverter(messageConverter);
}
@Override
@@ -143,60 +82,16 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
return getPhase();
}
private Message<V> toMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment) {
private class IntegrationMessageListener extends MessagingMessageListenerAdapter<K, V> {
KafkaMessageHeaders kafkaMessageHeaders = new KafkaMessageHeaders(this.generateMessageId,
this.generateTimestamp);
Map<String, Object> rawHeaders = kafkaMessageHeaders.getRawHeaders();
rawHeaders.put(KafkaHeaders.MESSAGE_KEY, record.key());
rawHeaders.put(KafkaHeaders.TOPIC, record.topic());
rawHeaders.put(KafkaHeaders.PARTITION_ID, record.partition());
rawHeaders.put(KafkaHeaders.OFFSET, record.offset());
if (acknowledgment != null) {
rawHeaders.put(KafkaHeaders.ACKNOWLEDGMENT, acknowledgment);
IntegrationMessageListener() {
super(null);
}
if (this.useMessageBuilderFactory) {
return getMessageBuilderFactory()
.withPayload(record.value())
.copyHeaders(kafkaMessageHeaders)
.build();
}
else {
return MessageBuilder.createMessage(record.value(), kafkaMessageHeaders);
}
}
private class AutoAcknowledgingChannelForwardingMessageListener implements MessageListener<K, V> {
@Override
public void onMessage(ConsumerRecord<K, V> record) {
sendMessage(toMessage(record, null));
}
}
private class AcknowledgingChannelForwardingMessageListener implements AcknowledgingMessageListener<K, V> {
@Override
public void onMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment) {
sendMessage(toMessage(record, acknowledgment));
}
}
@SuppressWarnings("serial")
private static final class KafkaMessageHeaders extends MessageHeaders {
private KafkaMessageHeaders(boolean generateId, boolean generateTimestamp) {
super(null, generateId ? null : ID_VALUE_NONE, generateTimestamp ? null : -1L);
}
@Override
public Map<String, Object> getRawHeaders() {
return super.getRawHeaders();
Message<?> message = toMessagingMessage(record, acknowledgment);
sendMessage(message);
}
}

35
spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-2.0.xsd
View File

@@ -151,37 +151,16 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="use-context-message-builder" type="xsd:string">
<xsd:attribute name="message-converter" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
When 'true', use the configured message builder from the integration application context
(bean:"messageBuilderFactory", which defaults to a "DefaultMessageBuilderFactory").
When 'false', use an internal, optimized, message creation algorithm. Default 'false'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="set-id" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
When 'true', generate a unique ID header for each new message.
Default 'false'. Set to `true` if messages contain collections and you use a splitter
later in the flow, or a router where a message is sent to multiple recipients with
'apply-sequence' set to 'true'. In this cases, the message id is used as the correlation
id of the resulting messages. If 'use-context-message-builder' is 'true', this
setting may be ignored by the configured message builder factory (this is the case
for the 'DefaultMessageBuilderFactory', where an ID is always generated).
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="set-timestamp" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
When 'true', generate a TIMESTAMP header for each new message.
Default 'false' (TIMESTAMP header is set to -1).
If 'use-context-message-builder' is 'true', this
setting may be ignored by the configured message builder factory (this is the case
for the 'DefaultMessageBuilderFactory', where a timestamp header is always generated).
An 'org.springframework.kafka.support.converter.MessageConverter' bean reference.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.kafka.support.converter.MessageConverter"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>

3
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml
View File

@@ -20,8 +20,11 @@
phase="100"
send-timeout="5000"
channel="nullChannel"
message-converter="messageConverter"
error-channel="errorChannel" />
<bean id="messageConverter" class="org.springframework.kafka.support.converter.MessagingMessageConverter"/>
<bean id="container1" class="org.springframework.kafka.listener.KafkaMessageListenerContainer">
<constructor-arg>
<bean class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">

31
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java
View File

@@ -18,21 +18,28 @@ package org.springframework.integration.kafka.inbound;
import static org.assertj.core.api.Assertions.assertThat;
import java.lang.reflect.Type;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.junit.ClassRule;
import org.junit.Test;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.converter.MessagingMessageConverter;
import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
/**
* @author Gary Russell
@@ -44,10 +51,8 @@ public class MessageDrivenAdapterTests {
private static String topic1 = "testTopic1";
private static String topic2 = "testTopic2";
@ClassRule
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2);
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1);
@Test
public void testInbound() throws Exception {
@@ -60,15 +65,33 @@ public class MessageDrivenAdapterTests {
QueueChannel out = new QueueChannel();
adapter.setOutputChannel(out);
adapter.afterPropertiesSet();
adapter.setMessageConverter(new MessagingMessageConverter() {
@Override
public Message<?> toMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment, Type type) {
Message<?> message = super.toMessage(record, acknowledgment, type);
return MessageBuilder.fromMessage(message).setHeader("testHeader", "testValue").build();
}
});
adapter.start();
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<Integer, String>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic("testTopic1");
template.send("foo");
template.send(1, "foo");
Message<?> received = out.receive(10000);
assertThat(received).isNotNull();
MessageHeaders headers = received.getHeaders();
assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo(1);
assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo("testTopic1");
assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0);
assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo(0L);
assertThat(headers.get("testHeader")).isEqualTo("testValue");
adapter.stop();
}

40
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/NonSerializableTestKey.java
View File

@@ -1,40 +0,0 @@
/*
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.test.utils;
/**
* @author Soby Chacko
* @since 0.5
*/
public class NonSerializableTestKey {
private final String keyPart1;
private final String keyPart2;
public NonSerializableTestKey(final String keyPart1, final String keyPart2) {
this.keyPart1 = keyPart1;
this.keyPart2 = keyPart2;
}
public String getKeyPart1() {
return keyPart1;
}
public String getKeyPart2() {
return keyPart2;
}
}

39
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/NonSerializableTestPayload.java
View File

@@ -1,39 +0,0 @@
/*
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.test.utils;
/**
* @author Soby Chacko
* @since 0.5
*/
public class NonSerializableTestPayload {
private final String part1;
private final String part2;
public NonSerializableTestPayload(final String part1, final String part2) {
this.part1 = part1;
this.part2 = part2;
}
public String getPart1() {
return part1;
}
public String getPart2() {
return part2;
}
}

43
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/TestKey.java
View File

@@ -1,43 +0,0 @@
/*
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.test.utils;
import java.io.Serializable;
/**
* @author Soby Chacko
* @since 0.5
*/
public class TestKey implements Serializable {
private static final long serialVersionUID = -6415387283545560656L;
private final String keyPart1;
private final String keyPart2;
public TestKey(final String keyPart1, final String keyPart2) {
this.keyPart1 = keyPart1;
this.keyPart2 = keyPart2;
}
public String getKeyPart1() {
return keyPart1;
}
public String getKeyPart2() {
return keyPart2;
}
}

43
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/TestObject.java
View File

@@ -1,43 +0,0 @@
/*
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.test.utils;
/**
* @author Soby Chacko
* @since 0.5
*/
public class TestObject {
public String testData1;
public int testData2;
public String getTestData1() {
return testData1;
}
public void setTestData1(final String testData1) {
this.testData1 = testData1;
}
public int getTestData2() {
return testData2;
}
public void setTestData2(final int testData2) {
this.testData2 = testData2;
}
}

43
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/TestPayload.java
View File

@@ -1,43 +0,0 @@
/*
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.kafka.test.utils;
import java.io.Serializable;
/**
* @author Soby Chacko
* @since 0.5
*/
public class TestPayload implements Serializable {
private static final long serialVersionUID = -8560378224929007403L;
private final String part1;
private final String part2;
public TestPayload(final String part1, final String part2) {
this.part1 = part1;
this.part2 = part2;
}
public String getPart1() {
return part1;
}
public String getPart2() {
return part2;
}
}