diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java index 816f0b3c90..202ade8ba6 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java @@ -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(); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java index fee9f08559..85c79c7889 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java @@ -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 extends MessageProducerSupport implements OrderlyShutdownCapable { private final AbstractMessageListenerContainer messageListenerContainer; - private boolean generateMessageId = false; - - private boolean generateTimestamp = false; - - private boolean useMessageBuilderFactory = false; + private final MessagingMessageListenerAdapter listener = new IntegrationMessageListener(); public KafkaMessageDrivenChannelAdapter(AbstractMessageListenerContainer 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 extends MessageProducerSuppo return getPhase(); } - private Message toMessage(ConsumerRecord record, Acknowledgment acknowledgment) { + private class IntegrationMessageListener extends MessagingMessageListenerAdapter { - KafkaMessageHeaders kafkaMessageHeaders = new KafkaMessageHeaders(this.generateMessageId, - this.generateTimestamp); - - Map 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 { - - @Override - public void onMessage(ConsumerRecord record) { - sendMessage(toMessage(record, null)); - } - - } - - private class AcknowledgingChannelForwardingMessageListener implements AcknowledgingMessageListener { - @Override public void onMessage(ConsumerRecord 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 getRawHeaders() { - return super.getRawHeaders(); + Message message = toMessagingMessage(record, acknowledgment); + sendMessage(message); } } diff --git a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-2.0.xsd b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-2.0.xsd index 8cd912deb2..2706848f18 100644 --- a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-2.0.xsd +++ b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-2.0.xsd @@ -151,37 +151,16 @@ - + - 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'. - - - - - - - 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). - - - - - - - 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. + + + + + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml index 7f68dc5894..57d906b25e 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml @@ -20,8 +20,11 @@ phase="100" send-timeout="5000" channel="nullChannel" + message-converter="messageConverter" error-channel="errorChannel" /> + + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java index 04d3ba458e..2d10594f20 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java @@ -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 senderProps = KafkaTestUtils.producerProps(embeddedKafka); ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); KafkaTemplate 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(); } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/NonSerializableTestKey.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/NonSerializableTestKey.java deleted file mode 100644 index f37a744b02..0000000000 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/NonSerializableTestKey.java +++ /dev/null @@ -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; - } - -} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/NonSerializableTestPayload.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/NonSerializableTestPayload.java deleted file mode 100644 index 1061de369a..0000000000 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/NonSerializableTestPayload.java +++ /dev/null @@ -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; - } -} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/TestKey.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/TestKey.java deleted file mode 100644 index cbeffa5d14..0000000000 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/TestKey.java +++ /dev/null @@ -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; - } -} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/TestObject.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/TestObject.java deleted file mode 100644 index 301038e3bf..0000000000 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/TestObject.java +++ /dev/null @@ -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; - } - -} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/TestPayload.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/TestPayload.java deleted file mode 100644 index e99561453c..0000000000 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/test/utils/TestPayload.java +++ /dev/null @@ -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; - } -}