From 22b1e70e1346f665fc7ba336719b0b580c7b42b2 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 3 Apr 2019 12:49:51 -0400 Subject: [PATCH] GH-195: Add XML Schema support for new components Resolves https://github.com/spring-projects/spring-integration-kafka/issues/195 - gateways - message source Polishing Don't start adapters in parser tests Fix XML filename for test --- .../xml/KafkaInboundChannelAdapterParser.java | 61 ++ .../config/xml/KafkaInboundGatewayParser.java | 64 ++ .../config/xml/KafkaNamespaceHandler.java | 3 + .../KafkaOutboundChannelAdapterParser.java | 47 +- .../xml/KafkaOutboundGatewayParser.java | 52 ++ .../kafka/config/xml/KafkaParsingUtils.java | 86 +++ .../KafkaMessageDrivenChannelAdapter.java | 1 + .../kafka/inbound/KafkaMessageSource.java | 22 +- .../outbound/KafkaProducerMessageHandler.java | 5 +- .../config/spring-integration-kafka-3.2.xsd | 730 ++++++++++++------ .../kafka/config/xml/AllXmlTests-context.xml | 110 +++ .../kafka/config/xml/AllXmlTests.java | 55 ++ ...boundChannelAdapterParserTests-context.xml | 59 ++ ...KafkaInboundChannelAdapterParserTests.java | 70 ++ .../xml/KafkaInboundGatewayTests-context.xml | 54 ++ .../config/xml/KafkaInboundGatewayTests.java | 69 ++ ...afkaOutboundGatewayParserTests-context.xml | 42 + .../xml/KafkaOutboundGatewayParserTests.java | 75 ++ .../kafka/inbound/MessageSourceTests.java | 7 +- 19 files changed, 1340 insertions(+), 272 deletions(-) create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParser.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayParser.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParser.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaParsingUtils.java create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests-context.xml create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests.java create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests-context.xml create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests.java create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests-context.xml create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests.java diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParser.java new file mode 100644 index 0000000000..ff0fd4cf64 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParser.java @@ -0,0 +1,61 @@ +/* + * Copyright 2019 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 + * + * https://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.config.xml; + +import org.w3c.dom.Element; + +import org.springframework.beans.BeanMetadataElement; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.kafka.inbound.KafkaMessageSource; +import org.springframework.util.StringUtils; + +/** + * Parser for the inbound channel adapter. + * + * @author Gary Russell + * @since 3.2 + * + */ +public class KafkaInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser { + + @Override + protected boolean shouldGenerateIdAsFallback() { + return true; + } + + @Override + protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(KafkaMessageSource.class); + builder.addConstructorArgReference(element.getAttribute("consumer-factory")); + String attribute = element.getAttribute("ack-factory"); + if (StringUtils.hasText(attribute)) { + builder.addConstructorArgReference(attribute); + } + builder.addConstructorArgValue(element.getAttribute("topics")); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "client-id"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "group-id"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "payload-type"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "raw-header", "rawMessageHeader"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "rebalance-listener"); + return builder.getBeanDefinition(); + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayParser.java new file mode 100644 index 0000000000..1f835a192b --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayParser.java @@ -0,0 +1,64 @@ +/* + * Copyright 2019 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 + * + * https://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.config.xml; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.integration.config.xml.AbstractInboundGatewayParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.kafka.inbound.KafkaInboundGateway; + +/** + * Inbound gateway parser. + * + * @author Gary Russell + * @since 3.2 + * + */ +public class KafkaInboundGatewayParser extends AbstractInboundGatewayParser { + + + @Override + protected Class getBeanClass(Element element) { + return KafkaInboundGateway.class; + } + + @Override + protected void doPostProcess(BeanDefinitionBuilder builder, Element element) { + builder.addConstructorArgReference(element.getAttribute("listener-container")); + builder.addConstructorArgReference(element.getAttribute("kafka-template")); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "retry-template"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "recovery-callback"); + } + + @Override + protected boolean shouldGenerateIdAsFallback() { + return true; + } + + @Override + protected boolean isEligibleAttribute(String attributeName) { + return super.isEligibleAttribute(attributeName) + && !attributeName.equals("listener-container") + && !attributeName.equals("kafka-template"); + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java index ee8e9ba792..19513fb910 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java @@ -32,6 +32,9 @@ public class KafkaNamespaceHandler extends AbstractIntegrationNamespaceHandler { public void init() { registerBeanDefinitionParser("outbound-channel-adapter", new KafkaOutboundChannelAdapterParser()); registerBeanDefinitionParser("message-driven-channel-adapter", new KafkaMessageDrivenChannelAdapterParser()); + registerBeanDefinitionParser("outbound-gateway", new KafkaOutboundGatewayParser()); + registerBeanDefinitionParser("inbound-gateway", new KafkaInboundGatewayParser()); + registerBeanDefinitionParser("inbound-channel-adapter", new KafkaInboundChannelAdapterParser()); } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java index b41b9b423f..3e4a774301 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java @@ -18,12 +18,10 @@ package org.springframework.integration.kafka.config.xml; import org.w3c.dom.Element; -import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser; -import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; /** @@ -43,50 +41,7 @@ public class KafkaOutboundChannelAdapterParser extends AbstractOutboundChannelAd final BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(KafkaProducerMessageHandler.class); - final String kafkaTemplateBeanName = element.getAttribute("kafka-template"); - - builder.addConstructorArgReference(kafkaTemplateBeanName); - - BeanDefinition topicExpressionDef = - IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("topic", "topic-expression", - parserContext, element, false); - if (topicExpressionDef != null) { - builder.addPropertyValue("topicExpression", topicExpressionDef); - } - - BeanDefinition messageKeyExpressionDef = - IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("message-key", - "message-key-expression", parserContext, element, false); - if (messageKeyExpressionDef != null) { - builder.addPropertyValue("messageKeyExpression", messageKeyExpressionDef); - } - - BeanDefinition partitionIdExpressionDef = - IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("partition-id", - "partition-id-expression", parserContext, element, false); - if (partitionIdExpressionDef != null) { - builder.addPropertyValue("partitionIdExpression", partitionIdExpressionDef); - } - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "sync"); - - BeanDefinition sendTimeoutExpressionDef = - IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("send-timeout", - "send-timeout-expression", parserContext, element, false); - if (sendTimeoutExpressionDef != null) { - builder.addPropertyValue("sendTimeoutExpression", sendTimeoutExpressionDef); - } - - BeanDefinition timestampExpressionDef = - IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("timestamp-expression", element); - - if (timestampExpressionDef != null) { - builder.addPropertyValue("timestampExpression", timestampExpressionDef); - } - - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy"); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "send-failure-channel"); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "send-success-channel"); - + KafkaParsingUtils.commonOutboundProperties(element, parserContext, builder); return builder.getBeanDefinition(); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParser.java new file mode 100644 index 0000000000..58d7d3e830 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParser.java @@ -0,0 +1,52 @@ +/* + * Copyright 2019 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 + * + * https://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.config.xml; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractConsumerEndpointParser; +import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; + +/** + * Parser for the outbound gateway. + * + * @author Gary Russell + * @since 3.2 + * + */ +public class KafkaOutboundGatewayParser extends AbstractConsumerEndpointParser { + + @Override + protected boolean shouldGenerateIdAsFallback() { + return true; + } + + @Override + protected String getInputChannelAttributeName() { + return "request-channel"; + } + + @Override + protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(KafkaProducerMessageHandler.class); + KafkaParsingUtils.commonOutboundProperties(element, parserContext, builder); + return builder; + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaParsingUtils.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaParsingUtils.java new file mode 100644 index 0000000000..3f2b6d4246 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaParsingUtils.java @@ -0,0 +1,86 @@ +/* + * Copyright 2019 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 + * + * https://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.config.xml; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; + +/** + * Utilities to assist with parsing XML. + * + * @author Gary Russell + * @since 3.2 + * + */ +public final class KafkaParsingUtils { + + private KafkaParsingUtils() { + super(); + } + + public static void commonOutboundProperties(final Element element, final ParserContext parserContext, + final BeanDefinitionBuilder builder) { + + final String kafkaTemplateBeanName = element.getAttribute("kafka-template"); + builder.addConstructorArgReference(kafkaTemplateBeanName); + + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "send-failure-channel"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "send-success-channel"); + + BeanDefinition topicExpressionDef = + IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("topic", "topic-expression", + parserContext, element, false); + if (topicExpressionDef != null) { + builder.addPropertyValue("topicExpression", topicExpressionDef); + } + + BeanDefinition messageKeyExpressionDef = + IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("message-key", + "message-key-expression", parserContext, element, false); + if (messageKeyExpressionDef != null) { + builder.addPropertyValue("messageKeyExpression", messageKeyExpressionDef); + } + + BeanDefinition partitionIdExpressionDef = + IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("partition-id", + "partition-id-expression", parserContext, element, false); + if (partitionIdExpressionDef != null) { + builder.addPropertyValue("partitionIdExpression", partitionIdExpressionDef); + } + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "sync"); + + BeanDefinition sendTimeoutExpressionDef = + IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("send-timeout", + "send-timeout-expression", parserContext, element, false); + if (sendTimeoutExpressionDef != null) { + builder.addPropertyValue("sendTimeoutExpression", sendTimeoutExpressionDef); + } + + BeanDefinition timestampExpressionDef = + IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("timestamp-expression", element); + + if (timestampExpressionDef != null) { + builder.addPropertyValue("timestampExpression", timestampExpressionDef); + } + } + +} 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 a1175c0a3d..317fac6c67 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 @@ -112,6 +112,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo */ public KafkaMessageDrivenChannelAdapter(AbstractMessageListenerContainer messageListenerContainer, ListenerMode mode) { + Assert.notNull(messageListenerContainer, "messageListenerContainer is required"); Assert.isNull(messageListenerContainer.getContainerProperties().getMessageListener(), "Container must not already have a listener"); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java index e280b3e1b0..85f1a2ee3f 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java @@ -50,6 +50,7 @@ import org.springframework.integration.endpoint.Pausable; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.listener.ConsumerAwareRebalanceListener; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.converter.KafkaMessageHeaders; @@ -94,8 +95,6 @@ public class KafkaMessageSource extends AbstractMessageSource impl private final Supplier minTimeoutProvider = () -> Duration.ofMillis(Math.max(this.pollTimeout.toMillis() * 20, MIN_ASSIGN_TIMEOUT)); - private final Log logger = LogFactory.getLog(getClass()); - private final ConsumerFactory consumerFactory; private final KafkaAckCallbackFactory ackCallbackFactory; @@ -118,6 +117,8 @@ public class KafkaMessageSource extends AbstractMessageSource impl private ConsumerRebalanceListener rebalanceListener; + private ConsumerAwareRebalanceListener consumerAwareRebalanceListener; + private boolean rawMessageHeader; private Duration commitTimeout; @@ -145,6 +146,7 @@ public class KafkaMessageSource extends AbstractMessageSource impl Assert.notNull(consumerFactory, "'consumerFactory' must not be null"); Assert.notNull(ackCallbackFactory, "'ackCallbackFactory' must not be null"); + Assert.isTrue(topics != null && topics.length > 0, "At least one topic is required"); this.consumerFactory = fixOrRejectConsumerFactory(consumerFactory); this.ackCallbackFactory = ackCallbackFactory; this.topics = topics; @@ -223,6 +225,9 @@ public class KafkaMessageSource extends AbstractMessageSource impl */ public void setRebalanceListener(ConsumerRebalanceListener rebalanceListener) { this.rebalanceListener = rebalanceListener; + if (rebalanceListener instanceof ConsumerAwareRebalanceListener) { + this.consumerAwareRebalanceListener = (ConsumerAwareRebalanceListener) rebalanceListener; + } } @Override @@ -377,6 +382,7 @@ public class KafkaMessageSource extends AbstractMessageSource impl protected void createConsumer() { synchronized (this.consumerMonitor) { this.consumer = this.consumerFactory.createConsumer(this.groupId, this.clientId, null); + boolean isConsumerAware = this.consumerAwareRebalanceListener != null; this.consumer.subscribe(Arrays.asList(this.topics), new ConsumerRebalanceListener() { @Override @@ -385,7 +391,11 @@ public class KafkaMessageSource extends AbstractMessageSource impl if (KafkaMessageSource.this.logger.isInfoEnabled()) { KafkaMessageSource.this.logger.info("Partitions revoked: " + partitions); } - if (KafkaMessageSource.this.rebalanceListener != null) { + if (isConsumerAware) { + KafkaMessageSource.this.consumerAwareRebalanceListener.onPartitionsRevokedAfterCommit( + KafkaMessageSource.this.consumer, partitions); + } + else if (KafkaMessageSource.this.rebalanceListener != null) { KafkaMessageSource.this.rebalanceListener.onPartitionsRevoked(partitions); } } @@ -397,7 +407,11 @@ public class KafkaMessageSource extends AbstractMessageSource impl if (KafkaMessageSource.this.logger.isInfoEnabled()) { KafkaMessageSource.this.logger.info("Partitions assigned: " + partitions); } - if (KafkaMessageSource.this.rebalanceListener != null) { + if (isConsumerAware) { + KafkaMessageSource.this.consumerAwareRebalanceListener.onPartitionsAssigned( + KafkaMessageSource.this.consumer, partitions); + } + else if (KafkaMessageSource.this.rebalanceListener != null) { KafkaMessageSource.this.rebalanceListener.onPartitionsAssigned(partitions); } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java index 8d5782f1ce..6c4e10b242 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java @@ -132,8 +132,6 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes private Type replyPayloadType = Object.class; - private volatile boolean noOutputChannel; - public KafkaProducerMessageHandler(final KafkaTemplate kafkaTemplate) { Assert.notNull(kafkaTemplate, "kafkaTemplate cannot be null"); this.kafkaTemplate = kafkaTemplate; @@ -360,6 +358,9 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes String topic = this.topicExpression != null ? this.topicExpression.getValue(this.evaluationContext, message, String.class) : messageHeaders.get(KafkaHeaders.TOPIC, String.class); + if (topic == null) { + topic = this.kafkaTemplate.getDefaultTopic(); + } Assert.state(StringUtils.hasText(topic), "The 'topic' can not be empty or null"); diff --git a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.2.xsd b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.2.xsd index 2addbf8262..443c4121c3 100644 --- a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.2.xsd +++ b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.2.xsd @@ -27,155 +27,224 @@ - - - - + + + + + + + + + + + + + Defines the Consumer Endpoint for the KafkaProducerMessageHandler + that writes the contents of the Message to kafka broker and receives + a reply. + + + + + + + + + + Unique ID for this adapter. + + + + + + + + + + + + + + + + + Message Channel to which Messages should be sent in order to have them sent to Kafka. + + + + + + + + + + + + Specify whether this outbound gateway must return a non-null value. This value is + 'true' by default, and a ReplyRequiredException will be thrown when + the underlying service returns a null value. + + + + + + + Message Channel to which replies should be sent after being received from Kafka. + + + + + + + + + + + + + + + + + + + + + + Defines a Polling Channel Adapter for the + 'org.springframework.integration.kafka.inbound.KafkaMessageSource' + for polling a Kafka topic. + + + + + + - + - + + A reference to a ConsumerFactory. + - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + - Specifies a timeout in milliseconds for how long the 'KafkaProducerMessageHandler' - should wait wait for send operation results. Defaults to 10 seconds. - The timeout is applied only in 'sync' mode. + A reference to a ConsumerFactory. + + + + + + + + + + + + Comma-delimited list of topic names. + + + + + + + The 'group.id' Kafka Consumer property; overrides the value in the consumer factory. + + + + + + + The 'client.id' Kafka Consumer property; overrides the value in the consumer factory. + + + + + + + The time to block in poll() in milliseconds; default 50ms. - + - Specifies an expression that is evaluated to determine a timeout in milliseconds - for how long the 'KafkaProducerMessageHandler' - should wait wait for send operation results. Defaults to 10 seconds. - The timeout is applied only in 'sync' mode. + A reference to a RecordMessageConverter; default 'MessagingMessageConverter'. + + + + + + + + + + + + The payload type to which convert the record data; only applies if he message-converter + type-aware converter, such as a JSON converter. + Use a SpEL expression, e.g. '#{T(java.lang.String)}'. - + - Specifies the order for invocation when this endpoint is connected as a - subscriber to a SubscribableChannel. + A reference to a 'ConsumerRebalanceListener'. + + + + + + + + + + + + Set to true to add the raw 'ConsumerRecord' in the 'KafkaHeaders.RAW_DATA' header. - + - - - - - - - - - - - - - - - - - - - - - @@ -186,102 +255,98 @@ - - - - - Maximum amount of time in milliseconds to wait when sending a message to the channel - if such channel may block. For example, a Queue Channel can block until space is available - if its maximum capacity has been reached. - - - - - - - Message Channel to which error Messages should be sent. - - - - - - - - - - - - An 'org.springframework.kafka.listener.AbstractMessageListenerContainer' bean reference. - - - - - - - - - - - - An 'org.springframework.kafka.support.converter.MessageConverter' bean reference. - if mode = 'record' must be a 'RecordMessageConverter'; if mode = 'batch' must be - a `BatchMessageConverter`. Defaults to the default implementation for each mode. - - - - - - - - - - - - 'record' or 'batch' - default 'record' - one converted ConsumerRecord per message, when - 'batch' then the payload is a collection of converted ConsumerRecords. - - - - - - - - - - Set the payload type to convert to when using a type-aware message converter such as the - StringJsonMessageConverter. Fully qualified class name; defaults to 'java.lang.Object'. - - - - - - - A retry template for retrying deliveries; an 'error-channel' is not allowed - when a retry template is provided; configure a 'recovery-callback' such as an - 'ErrorMessageSendingRecoverer' when using a retry template. - - - - - - - - - - - - Used in conjunction with a 'retry-template'; in most cases this will be - an 'ErrorMessageSendingRecoverer'. Omitting this element will cause an - exception to be thrown to the listener container after retries are exhausted. - - - - - - - - - + + + + + + + 'record' or 'batch' - default 'record' - one converted ConsumerRecord per message, when + 'batch' then the payload is a collection of converted ConsumerRecords. + + + + + + + + + + Maximum amount of time in milliseconds to wait when sending a message to the channel + if such channel may block. For example, a Queue Channel can block until space is available + if its maximum capacity has been reached. + + + + + + + + + + + + Defines the Message Producing Endpoint for the KafkaInboundGateway. + + + + + + + + + Unique ID for this gateway. + + + + + + + + + Message Channel to which converted Messages should be sent. + + + + + + + + + + + + Maximum amount of time in milliseconds to wait when sending a message to the channel + if such channel may block. For example, a Queue Channel can block until space is available + if its maximum capacity has been reached. + + + + + + + Message Channel where reply Messages will be expected. + + + + + + + + + + + + + + + @@ -292,6 +357,223 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + Specifies a timeout in milliseconds for how long the 'KafkaProducerMessageHandler' + should wait wait for send operation results. Defaults to 10 seconds. + The timeout is applied only in 'sync' mode. + + + + + + + + + + Specifies an expression that is evaluated to determine a timeout in milliseconds + for how long the 'KafkaProducerMessageHandler' + should wait wait for send operation results. Defaults to 10 seconds. + The timeout is applied only in 'sync' mode. + + + + + + + Specifies the order for invocation when this endpoint is connected as a + subscriber to a SubscribableChannel. + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + Message Channel to which error Messages should be sent. + + + + + + + + + + + + An 'org.springframework.kafka.listener.AbstractMessageListenerContainer' bean reference. + + + + + + + + + + + + An 'org.springframework.kafka.support.converter.MessageConverter' bean reference. + if mode = 'record' must be a 'RecordMessageConverter'; if mode = 'batch' must be + a `BatchMessageConverter`. Defaults to the default implementation for each mode. + + + + + + + + + + + + Set the payload type to convert to when using a type-aware message converter such as the + StringJsonMessageConverter. Fully qualified class name; defaults to 'java.lang.Object'. + + + + + + + A retry template for retrying deliveries; an 'error-channel' is not allowed + when a retry template is provided; configure a 'recovery-callback' such as an + 'ErrorMessageSendingRecoverer' when using a retry template. + + + + + + + + + + + + Used in conjunction with a 'retry-template'; in most cases this will be + an 'ErrorMessageSendingRecoverer'. Omitting this element will cause an + exception to be thrown to the listener container after retries are exhausted. + + + + + + + + + + + @@ -307,4 +589,18 @@ + + + + + + + + + + + + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml new file mode 100644 index 0000000000..779d4d62c3 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml @@ -0,0 +1,110 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java new file mode 100644 index 0000000000..f5fe1a059f --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java @@ -0,0 +1,55 @@ +/* + * Copyright 2019 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 + * + * https://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.config.xml; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.messaging.Message; +import org.springframework.messaging.PollableChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +/** + * @author Gary Russell + * @since 3.2 + * + */ +@SpringJUnitConfig +@DirtiesContext +@EmbeddedKafka(topics = { "one", "two", "three", "four" }) +public class AllXmlTests { + + @Autowired + private KafkaTemplate template; + + @Autowired + private PollableChannel lastChannel; + + @Test + public void testEndToEnd() { + this.template.send("one", "foo"); + Message received = this.lastChannel.receive(30_000); + assertThat(received).isNotNull(); + assertThat(received.getPayload()).isEqualTo("fooonetwothreefour"); + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests-context.xml new file mode 100644 index 0000000000..ce8c20f00f --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests-context.xml @@ -0,0 +1,59 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests.java new file mode 100644 index 0000000000..7e2a7c7c55 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests.java @@ -0,0 +1,70 @@ +/* + * Copyright 2019 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 + * + * https://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.config.xml; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.ApplicationContext; +import org.springframework.integration.kafka.inbound.KafkaMessageSource; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +/** + * @author Gary Russell + * @since 3.2 + * + */ +@SpringJUnitConfig +@DirtiesContext +public class KafkaInboundChannelAdapterParserTests { + + @Autowired + @Qualifier("adapter1.source") + private KafkaMessageSource source1; + + @Autowired + @Qualifier("adapter2.source") + private KafkaMessageSource source2; + + @Autowired + private ApplicationContext context; + + @Test + public void testProps() { + assertThat(TestUtils.getPropertyValue(this.source1, "topics")).isEqualTo(new String[] { "topic1" }); + assertThat(TestUtils.getPropertyValue(this.source1, "consumerFactory")) + .isSameAs(this.context.getBean("consumerFactory")); + assertThat(TestUtils.getPropertyValue(this.source1, "ackCallbackFactory")) + .isSameAs(this.context.getBean("ackFactory")); + assertThat(TestUtils.getPropertyValue(this.source1, "clientId")).isEqualTo("client"); + assertThat(TestUtils.getPropertyValue(this.source1, "groupId")).isEqualTo("group"); + assertThat(TestUtils.getPropertyValue(this.source1, "messageConverter")) + .isSameAs(this.context.getBean("converter")); + assertThat(TestUtils.getPropertyValue(this.source1, "payloadType")).isEqualTo(String.class); + assertThat(TestUtils.getPropertyValue(this.source1, "rawMessageHeader", Boolean.class)).isTrue(); + assertThat(TestUtils.getPropertyValue(this.source1, "rebalanceListener")) + .isSameAs(this.context.getBean("rebal")); + + assertThat(TestUtils.getPropertyValue(this.source2, "topics")).isEqualTo(new String[] { "topic1", "topic2" }); + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests-context.xml new file mode 100644 index 0000000000..d7d3d9689f --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests-context.xml @@ -0,0 +1,54 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests.java new file mode 100644 index 0000000000..feac4c59ac --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests.java @@ -0,0 +1,69 @@ +/* + * Copyright 2019 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 + * + * https://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.config.xml; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; +import org.springframework.integration.kafka.inbound.KafkaInboundGateway; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.kafka.listener.KafkaMessageListenerContainer; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +/** + * @author Gary Russell + * @since 3.2 + * + */ +@SpringJUnitConfig +@DirtiesContext +public class KafkaInboundGatewayTests { + + @Autowired + private KafkaInboundGateway gateway1; + + @Autowired + private ApplicationContext context; + + @Test + public void testProps() { + assertThat(this.gateway1.isAutoStartup()).isFalse(); + assertThat(this.gateway1.isRunning()).isFalse(); + assertThat(this.gateway1.getPhase()).isEqualTo(100); + assertThat(TestUtils.getPropertyValue(this.gateway1, "requestChannelName")).isEqualTo("nullChannel"); + assertThat(TestUtils.getPropertyValue(this.gateway1, "replyChannelName")).isEqualTo("errorChannel"); + KafkaMessageListenerContainer container = + TestUtils.getPropertyValue(this.gateway1, "messageListenerContainer", + KafkaMessageListenerContainer.class); + assertThat(container).isNotNull(); + assertThat(TestUtils.getPropertyValue(this.gateway1, "listener.fallbackType")) + .isEqualTo(String.class); + assertThat(TestUtils.getPropertyValue(this.gateway1, "errorMessageStrategy")) + .isSameAs(this.context.getBean("ems")); + assertThat(TestUtils.getPropertyValue(this.gateway1, "retryTemplate")) + .isSameAs(this.context.getBean("retryTemplate")); + assertThat(TestUtils.getPropertyValue(this.gateway1, "recoveryCallback")) + .isSameAs(this.context.getBean("recoveryCallback")); + assertThat(TestUtils.getPropertyValue(this.gateway1, "messagingTemplate.sendTimeout")).isEqualTo(5000L); + assertThat(TestUtils.getPropertyValue(this.gateway1, "replyTimeout")).isEqualTo(43L); + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests-context.xml new file mode 100644 index 0000000000..74fe603b4e --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests-context.xml @@ -0,0 +1,42 @@ + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests.java new file mode 100644 index 0000000000..acee42b7f9 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests.java @@ -0,0 +1,75 @@ +/* + * Copyright 2019 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 + * + * https://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.config.xml; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.ApplicationContext; +import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; +import org.springframework.integration.support.DefaultErrorMessageStrategy; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +/** + * @author Gary Russell + * @since 3.2 + * + */ +@SpringJUnitConfig +@DirtiesContext +public class KafkaOutboundGatewayParserTests { + + @Autowired + @Qualifier("allProps.handler") + KafkaProducerMessageHandler messageHandler; + + @Autowired + private ApplicationContext context; + + @Test + public void testProps() { + assertThat(TestUtils.getPropertyValue(this.messageHandler, "errorMessageStrategy")).isInstanceOf(EMS.class); + assertThat(TestUtils.getPropertyValue(this.messageHandler, "kafkaTemplate")) + .isSameAs(this.context.getBean("template")); + assertThat(this.messageHandler.getOrder()).isEqualTo(23); + assertThat(TestUtils.getPropertyValue(this.messageHandler, "topicExpression.expression")).isEqualTo("'topic'"); + assertThat(TestUtils.getPropertyValue(this.messageHandler, "messageKeyExpression.expression")) + .isEqualTo("'key'"); + assertThat(TestUtils.getPropertyValue(this.messageHandler, "partitionIdExpression.expression")).isEqualTo("2"); + assertThat(TestUtils.getPropertyValue(this.messageHandler, "sync", Boolean.class)).isTrue(); + assertThat(TestUtils.getPropertyValue(this.messageHandler, "sendTimeoutExpression.expression")).isEqualTo("44"); + assertThat(TestUtils.getPropertyValue(this.messageHandler, "timestampExpression.expression")) + .isEqualTo("T(System).currentTimeMillis()"); + + assertThat(TestUtils.getPropertyValue(this.messageHandler, "errorMessageStrategy")) + .isSameAs(this.context.getBean("ems")); + assertThat(TestUtils.getPropertyValue(this.messageHandler, "sendFailureChannel")) + .isSameAs(this.context.getBean("failures")); + assertThat(TestUtils.getPropertyValue(this.messageHandler, "sendSuccessChannel")) + .isSameAs(this.context.getBean("successes")); + } + + public static class EMS extends DefaultErrorMessageStrategy { + + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java index 8b7324a572..0346fc153d 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java @@ -403,17 +403,18 @@ public class MessageSourceTests { @SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testMaxPollRecords() { - KafkaMessageSource source = new KafkaMessageSource(new DefaultKafkaConsumerFactory<>(Collections.emptyMap())); + KafkaMessageSource source = new KafkaMessageSource(new DefaultKafkaConsumerFactory<>(Collections.emptyMap()), + "topic"); assertThat((TestUtils.getPropertyValue(source, "consumerFactory.configs", Map.class) .get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG))).isEqualTo(1); source = new KafkaMessageSource(new DefaultKafkaConsumerFactory<>( - Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2))); + Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2)), "topic"); assertThat((TestUtils.getPropertyValue(source, "consumerFactory.configs", Map.class) .get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG))).isEqualTo(1); try { new KafkaMessageSource((new DefaultKafkaConsumerFactory(Collections.emptyMap()) { - })); + }), "topic"); fail("Expected exception"); } catch (IllegalArgumentException e) {