From 52d218988bf0091483de13120bbd93e85b7e2ec2 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 20 Mar 2015 15:56:31 -0400 Subject: [PATCH] INTEXT-154: Namespace Support for New Features JIRA: https://jira.spring.io/browse/INTEXT-154, https://jira.spring.io/browse/INTEXT-138 --- ...afkaMessageDrivenChannelAdapterParser.java | 10 +- .../KafkaMessageDrivenChannelAdapter.java | 30 ++++- .../kafka/support/KafkaHeaders.java | 4 +- .../xml/spring-integration-kafka-1.0.xsd | 42 +++++++ ...rivenChannelAdapterParserTests-context.xml | 65 ++++++++++- ...essageDrivenChannelAdapterParserTests.java | 110 +++++++++++++++++- 6 files changed, 251 insertions(+), 10 deletions(-) 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 8def6edf5c..365750daa7 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 @@ -20,7 +20,6 @@ import org.w3c.dom.Element; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.AbstractChannelAdapterParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; @@ -93,6 +92,15 @@ public class KafkaMessageDrivenChannelAdapterParser extends AbstractChannelAdapt IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "key-decoder"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "payload-decoder"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + "auto-commit", "autoCommitOffset"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + "use-context-message-builder", "useMessageBuilderFactory"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + "set-id", "generateMessageId"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + "set-timestamp", "generateTimestamp"); + 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 5e9da2b004..d0b14cd270 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 @@ -73,6 +73,14 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport imp this.payloadDecoder = payloadDecoder; } + /** + * Automatically commit the offsets when 'true'. When 'false', the + * adapter inserts a 'kafka_acknowledgment` header allowing the user to manually + * commit the offset using the {@link Acknowledgment#acknowledge()} method. + * Default 'true'. + * + * @param autoCommitOffset false to not auto-commit (default true). + */ public void setAutoCommitOffset(boolean autoCommitOffset) { this.autoCommitOffset = autoCommitOffset; } @@ -80,7 +88,8 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport imp /** * 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 only works in conjunction with {@link #setUseMessageBuilderFactory(boolean)}. + * 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 @@ -93,7 +102,8 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport imp /** * 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 only works in conjunction with {@link #setUseMessageBuilderFactory(boolean)}. + * 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 @@ -220,7 +230,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport imp /** * Special subclass of {@link Message}. It is used for lower message generation overhead, unless the default - * strategy of the superclass is set via {@link #setUseMessageBuilderFactory(boolean)} + * strategy of the outer class is set via {@link #setUseMessageBuilderFactory(boolean)} * @since 1.1 */ private class KafkaMessage implements Message { @@ -244,6 +254,20 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport imp return this.messageHeaders; } + @Override + public String toString() { + StringBuilder sb = new StringBuilder(getClass().getSimpleName()); + sb.append(" [payload="); + if (this.payload instanceof byte[]) { + sb.append("byte[").append(((byte[]) this.payload).length).append("]"); + } + else { + sb.append(this.payload); + } + sb.append(", headers=").append(this.messageHeaders).append("]"); + return sb.toString(); + } + } @SuppressWarnings("serial") diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaHeaders.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaHeaders.java index 7d40175401..ace9e965e3 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaHeaders.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaHeaders.java @@ -29,9 +29,9 @@ public abstract class KafkaHeaders { public static final String MESSAGE_KEY = PREFIX + "messageKey"; - public static final String PARTITION_ID = PREFIX + "_partitionId"; + public static final String PARTITION_ID = PREFIX + "partitionId"; - public static final String OFFSET = PREFIX + "_offset"; + public static final String OFFSET = PREFIX + "offset"; public static final String NEXT_OFFSET = PREFIX + "nextOffset"; diff --git a/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd b/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd index 362cc66c97..e7bbb0e41f 100644 --- a/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd +++ b/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd @@ -651,6 +651,48 @@ + + + + When 'true', the adapter automatically commits the offset. When 'false', the adapter + inserts a 'kafka_acknowledgment` header allowing the user to manually commit the offset + using the `Acknowledgment.acknowledge()` method. Default 'true'. + + + + + + + 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). + + + 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 da0bcb0dd2..336777cf3d 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 @@ -49,8 +49,71 @@ queue-size="${queue.size:1024}" concurrency="${concurrency:10}" max-fetch="${max.fetch:1000}" - topics="${topics:foo,bar}"/> + topics="${topics:foo,bar}" /> + + + + +