diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java new file mode 100644 index 0000000000..503fe5afe5 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java @@ -0,0 +1,96 @@ +/* + * Copyright 2020 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.channel; + +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import org.springframework.core.log.LogAccessor; +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.kafka.core.KafkaOperations; +import org.springframework.kafka.support.KafkaHeaders; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.Assert; + +/** + * Abstract MessageChannel backed by a Kafka topic. + * + * @author Gary Russell + * @since 3.3 + * + */ +public abstract class AbstractKafkaChannel extends AbstractMessageChannel { + + protected final LogAccessor logger = new LogAccessor(super.logger); // NOSONAR final + + private final KafkaOperations template; + + protected final String topic; // NOSONAR final + + private String groupId; + + /** + * Construct an instance with the provided paramters. + * @param template the template. + * @param topic the topic. + */ + public AbstractKafkaChannel(KafkaOperations template, String topic) { + Assert.notNull(template, "'template' cannot be null"); + Assert.notNull(topic, "'topic' cannot be null"); + this.template = template; + this.topic = topic; + } + + /** + * Set the group id for the consumer; if not set, the bean name will be used. + * @param groupId the group id. + */ + public void setGroupId(String groupId) { + this.groupId = groupId; + } + + protected String getGroupId() { + return this.groupId; + } + + @Override + protected boolean doSend(Message message, long timeout) { + try { + this.template.send(MessageBuilder.fromMessage(message) + .setHeader(KafkaHeaders.TOPIC, this.topic) + .build()) + .get(timeout, TimeUnit.MILLISECONDS); + } + catch (@SuppressWarnings("unused") InterruptedException e) { + Thread.currentThread().interrupt(); + this.logger.debug(() -> "Interrupted while waiting for send result for: " + message); + return false; + } + catch (ExecutionException e) { + this.logger.error(e.getCause(), () -> "Interrupted while waiting for send result for: " + message); + return false; + } + catch (TimeoutException e) { + this.logger.debug(e, () -> "Timed out while waiting for send result for: " + message); + return false; + } + return true; + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PollableKafkaChannel.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PollableKafkaChannel.java new file mode 100644 index 0000000000..5a24ebf7e1 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PollableKafkaChannel.java @@ -0,0 +1,220 @@ +/* + * Copyright 2020 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.channel; + +import java.util.ArrayDeque; +import java.util.Deque; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.springframework.integration.channel.ExecutorChannelInterceptorAware; +import org.springframework.integration.kafka.inbound.KafkaMessageSource; +import org.springframework.integration.support.management.PollableChannelManagement; +import org.springframework.integration.support.management.metrics.CounterFacade; +import org.springframework.integration.support.management.metrics.MetricsCaptor; +import org.springframework.kafka.core.KafkaOperations; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.ChannelInterceptor; +import org.springframework.messaging.support.ExecutorChannelInterceptor; +import org.springframework.util.Assert; + +/** + * Pollable channel backed by a Kafka topic. + * + * @author Gary Russell + * @since 3.3 + * + */ +public class PollableKafkaChannel extends AbstractKafkaChannel + implements PollableChannel, PollableChannelManagement, ExecutorChannelInterceptorAware { + + private final KafkaMessageSource source; + + private CounterFacade receiveCounter; + + private volatile int executorInterceptorsSize; + + /** + * Construct an instance with the provided parameters. + * @param template the template for sending. + * @param source the source for receiving. + */ + public PollableKafkaChannel(KafkaOperations template, KafkaMessageSource source) { + super(template, topic(source)); + this.source = source; + if (source.getConsumerProperties().getGroupId() == null) { + String groupId = getGroupId(); + source.getConsumerProperties().setGroupId(groupId != null ? groupId : getBeanName()); + } + } + + private static String topic(KafkaMessageSource source) { + Assert.notNull(source, "'source' cannot be null"); + Assert.isTrue(source.getConsumerProperties().getTopics().length == 1, "Only one topic is allowed"); + return source.getConsumerProperties().getTopics()[0]; + } + + @Override + public int getReceiveCount() { + return getMetrics().getReceiveCount(); + } + + @Override + public long getReceiveCountLong() { + return getMetrics().getReceiveCountLong(); + } + + @Override + public int getReceiveErrorCount() { + return getMetrics().getReceiveErrorCount(); + } + + @Override + public long getReceiveErrorCountLong() { + return getMetrics().getReceiveErrorCountLong(); + } + + @Override + @Nullable + public Message receive() { + return doReceive(); + } + + @Override + @Nullable + public Message receive(long timeout) { + return doReceive(); + } + + @Nullable + protected Message doReceive() { + ChannelInterceptorList interceptorList = getIChannelInterceptorList(); + Deque interceptorStack = null; + AtomicBoolean counted = new AtomicBoolean(); + boolean countsEnabled = isCountsEnabled(); + boolean traceEnabled = isLoggingEnabled() && logger.isTraceEnabled(); + try { + if (traceEnabled) { + logger.trace("preReceive on channel '" + this + "'"); + } + if (interceptorList.getInterceptors().size() > 0) { + interceptorStack = new ArrayDeque<>(); + if (!interceptorList.preReceive(this, interceptorStack)) { + return null; + } + } + Message message = this.source.receive(); + if (message != null) { + incrementReceiveCounter(); + message = interceptorList.postReceive(message, this); + } + interceptorList.afterReceiveCompletion(message, this, null, interceptorStack); + return message; + } + catch (RuntimeException ex) { + if (countsEnabled && !counted.get()) { + incrementReceiveErrorCounter(ex); + } + interceptorList.afterReceiveCompletion(null, this, ex, interceptorStack); + throw ex; + } + } + + private void incrementReceiveCounter() { + MetricsCaptor metricsCaptor = getMetricsCaptor(); + if (metricsCaptor != null) { + if (this.receiveCounter == null) { + this.receiveCounter = buildReceiveCounter(metricsCaptor, null); + } + this.receiveCounter.increment(); + } + } + + private void incrementReceiveErrorCounter(Exception ex) { + MetricsCaptor metricsCaptor = getMetricsCaptor(); + if (metricsCaptor != null) { + buildReceiveCounter(metricsCaptor, ex).increment(); + } + getMetrics().afterError(); + } + + private CounterFacade buildReceiveCounter(MetricsCaptor metricsCaptor, @Nullable Exception ex) { + CounterFacade counterFacade = metricsCaptor + .counterBuilder(RECEIVE_COUNTER_NAME) + .tag("name", getComponentName() == null ? "unknown" : getComponentName()) + .tag("type", "channel") + .tag("result", ex == null ? "success" : "failure") + .tag("exception", ex == null ? "none" : ex.getClass().getSimpleName()) + .description("Messages received") + .build(); + this.meters.add(counterFacade); + return counterFacade; + } + + @Override + public void setInterceptors(List interceptors) { + super.setInterceptors(interceptors); + for (ChannelInterceptor interceptor : interceptors) { + if (interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize++; + } + } + } + + @Override + public void addInterceptor(ChannelInterceptor interceptor) { + super.addInterceptor(interceptor); + if (interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize++; + } + } + + @Override + public void addInterceptor(int index, ChannelInterceptor interceptor) { + super.addInterceptor(index, interceptor); + if (interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize++; + } + } + + @Override + public boolean removeInterceptor(ChannelInterceptor interceptor) { + boolean removed = super.removeInterceptor(interceptor); + if (removed && interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize--; + } + return removed; + } + + @Override + @Nullable + public ChannelInterceptor removeInterceptor(int index) { + ChannelInterceptor interceptor = super.removeInterceptor(index); + if (interceptor instanceof ExecutorChannelInterceptor) { + this.executorInterceptorsSize--; + } + return interceptor; + } + + @Override + public boolean hasExecutorInterceptors() { + return this.executorInterceptorsSize > 0; + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java new file mode 100644 index 0000000000..2603caf26c --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java @@ -0,0 +1,182 @@ +/* + * Copyright 2020 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.channel; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; + +import org.springframework.context.Phased; +import org.springframework.context.SmartLifecycle; +import org.springframework.integration.dispatcher.BroadcastingDispatcher; +import org.springframework.integration.dispatcher.MessageDispatcher; +import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; +import org.springframework.integration.dispatcher.UnicastingDispatcher; +import org.springframework.kafka.config.KafkaListenerContainerFactory; +import org.springframework.kafka.core.KafkaOperations; +import org.springframework.kafka.listener.MessageListenerContainer; +import org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter; +import org.springframework.kafka.support.Acknowledgment; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.SubscribableChannel; +import org.springframework.util.Assert; + +/** + * Subscribable channel backed by a Kafka topic. + * + * @author Gary Russell + * @since 3.3 + * + */ +public class SubscribableKafkaChannel extends AbstractKafkaChannel implements SubscribableChannel, SmartLifecycle { + + private static final int DEFAULT_PHASE = Integer.MAX_VALUE / 2; // same as MessageProducerSupport + + private final KafkaListenerContainerFactory factory; + + private final boolean pubSub; + + private MessageDispatcher dispatcher; + + private MessageListenerContainer container; + + private boolean autoStartup = true; + + private int phase = DEFAULT_PHASE; + + private volatile boolean running; + + /** + * Construct an instance with the provided parameters. + * @param template template for sending. + * @param factory factory for creating a container for receiving. + * @param channelTopic the topic. + */ + public SubscribableKafkaChannel(KafkaOperations template, KafkaListenerContainerFactory factory, + String channelTopic) { + + this(template, factory, channelTopic, false); + } + + /** + * Construct an instance with the provided parameters. + * @param template template for sending. + * @param factory factory for creating a container for receiving. + * @param channelTopic the topic. + * @param pubSub true for a publish/subscribe channel. + */ + public SubscribableKafkaChannel(KafkaOperations template, KafkaListenerContainerFactory factory, + String channelTopic, boolean pubSub) { + + super(template, channelTopic); + Assert.notNull(factory, "'factory' cannot be null"); + this.factory = factory; + this.pubSub = pubSub; + } + + @Override + public int getPhase() { + return this.phase; + } + + /** + * Set the phase. + * @param phase the phase. + * @see Phased + */ + public void setPhase(int phase) { + this.phase = phase; + } + + @Override + public boolean isRunning() { + return this.running; + } + + /** + * Set the auto startup. + * @param autoStartup true to automatically start. + * @see SmartLifecycle + */ + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + + @Override + public boolean isAutoStartup() { + return this.autoStartup; + } + + @Override + protected void onInit() { + if (this.pubSub) { + BroadcastingDispatcher broadcastingDispatcher = new BroadcastingDispatcher(true); + broadcastingDispatcher.setBeanFactory(this.getBeanFactory()); + this.dispatcher = broadcastingDispatcher; + } + else { + UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher(); + unicastingDispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy()); + this.dispatcher = unicastingDispatcher; + } + this.container = this.factory.createContainer(this.topic); + String groupId = getGroupId(); + this.container.getContainerProperties().setGroupId(groupId != null ? groupId : getBeanName()); + this.container.getContainerProperties().setMessageListener( + new RecordMessagingMessageListenerAdapter(null, null) { + + @Override + public void onMessage(ConsumerRecord record, Acknowledgment acknowledgment, + Consumer consumer) { + + SubscribableKafkaChannel.this.dispatcher + .dispatch(toMessagingMessage(record, acknowledgment, consumer)); + } + + }); + } + + @Override + public void start() { + this.container.start(); + this.running = true; + } + + @Override + public void stop() { + this.container.stop(); + this.running = false; + } + + @Override + public void stop(Runnable callback) { + this.container.stop(() -> { + callback.run(); + this.running = false; + }); + } + + @Override + public boolean subscribe(MessageHandler handler) { + return this.dispatcher.addHandler(handler); + } + + @Override + public boolean unsubscribe(MessageHandler handler) { + return this.dispatcher.removeHandler(handler); + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/package-info.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/package-info.java new file mode 100644 index 0000000000..b42feb5e6e --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes related to message channels. + */ +package org.springframework.integration.kafka.channel; diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaChannelParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaChannelParser.java new file mode 100644 index 0000000000..cdf8eddeea --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaChannelParser.java @@ -0,0 +1,77 @@ +/* + * Copyright 2020 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.AbstractChannelParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.kafka.channel.PollableKafkaChannel; +import org.springframework.integration.kafka.channel.SubscribableKafkaChannel; +import org.springframework.util.StringUtils; + +/** + * Parser for a channel backed by a Kafka topic. + * + * @author Gary Russell + * @since 3.3 + * + */ +public class KafkaChannelParser extends AbstractChannelParser { + + @Override + protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder; + String factory = element.getAttribute("container-factory"); + boolean hasFactory = StringUtils.hasText(factory); + String source = element.getAttribute("message-source"); + boolean hasSource = StringUtils.hasText(source); + String template = element.getAttribute("kafka-template"); + String topic = element.getAttribute("topic"); + boolean pubSub = "publish-subscribe-channel".equals(element.getLocalName()); + if (hasFactory) { + builder = BeanDefinitionBuilder.genericBeanDefinition(SubscribableKafkaChannel.class); + builder.addConstructorArgReference(template); + builder.addConstructorArgReference(factory); + builder.addConstructorArgValue(topic); + builder.addConstructorArgValue(pubSub); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "role"); + } + else if (hasSource) { + builder = BeanDefinitionBuilder.genericBeanDefinition(PollableKafkaChannel.class); + builder.addConstructorArgReference(template); + builder.addConstructorArgReference(source); + } + else { + if (pubSub) { + parserContext.getReaderContext().error("A 'container-factory' is required", element); + } + else { + parserContext.getReaderContext().error("Either a 'container-factory' or 'message-source' is required", + element); + } + return null; + } + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "group-id"); + return builder; + } + +} 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 19513fb910..d3fc6b01f1 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 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. @@ -35,6 +35,10 @@ public class KafkaNamespaceHandler extends AbstractIntegrationNamespaceHandler { registerBeanDefinitionParser("outbound-gateway", new KafkaOutboundGatewayParser()); registerBeanDefinitionParser("inbound-gateway", new KafkaInboundGatewayParser()); registerBeanDefinitionParser("inbound-channel-adapter", new KafkaInboundChannelAdapterParser()); + KafkaChannelParser channelParser = new KafkaChannelParser(); + registerBeanDefinitionParser("channel", channelParser); + registerBeanDefinitionParser("pollable-channel", channelParser); + registerBeanDefinitionParser("publish-subscribe-channel", channelParser); } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/AbstractKafkaChannelSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/AbstractKafkaChannelSpec.java new file mode 100644 index 0000000000..2a5fe2a240 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/AbstractKafkaChannelSpec.java @@ -0,0 +1,62 @@ +/* + * Copyright 2020 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.dsl; + +import org.springframework.integration.dsl.MessageChannelSpec; +import org.springframework.integration.kafka.channel.AbstractKafkaChannel; +import org.springframework.kafka.core.KafkaTemplate; + +/** + * + * Spec for a message channel backed by a Kafka topic. + * + * @param the spec type. + * + * @author Gary Russell + * @since 3.3 + * + */ +public abstract class AbstractKafkaChannelSpec> + extends MessageChannelSpec { + + protected final KafkaTemplate template; // NOSONAR final + + protected final String topic; // NOSONAR final + + protected String groupId; // NOSONAR + + protected AbstractKafkaChannelSpec(KafkaTemplate template, String topic) { + this.template = template; + this.topic = topic; + } + + @Override + public S id(String idToSet) { // NOSONAR - increase visibility + return super.id(idToSet); + } + + /** + * Set the group id to use on the consumer side. + * @param group the group id. + * @return the spec. + */ + public S groupId(String group) { + this.groupId = group; + return _this(); + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java index 114a9fd33a..5d8680b6e6 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2020 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. @@ -22,7 +22,9 @@ import java.util.regex.Pattern; import org.apache.kafka.common.TopicPartition; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; +import org.springframework.integration.kafka.inbound.KafkaMessageSource; import org.springframework.integration.kafka.inbound.KafkaMessageSource.KafkaAckCallbackFactory; +import org.springframework.kafka.config.KafkaListenerContainerFactory; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; @@ -535,6 +537,47 @@ public final class Kafka { return new KafkaInboundGatewaySpec.KafkaInboundGatewayListenerContainerSpec<>(containerSpec, templateSpec); } + /** + * Create a spec for a subscribable channel with the provided parameters. + * @param template the template. + * @param containerFactory the container factory. + * @param topic the topic. + * @return the spec. + * @since 3.3 + */ + public static KafkaSubscribableChannelSpec channel(KafkaTemplate template, + KafkaListenerContainerFactory containerFactory, String topic) { + + return new KafkaSubscribableChannelSpec(template, containerFactory, topic, false); + } + + /** + * Create a spec for a publish/subscribe channel with the provided parameters. + * @param template the template. + * @param containerFactory the container factory. + * @param topic the topic. + * @return the spec. + * @since 3.3 + */ + public static KafkaSubscribableChannelSpec publishSubscribeChannel(KafkaTemplate template, + KafkaListenerContainerFactory containerFactory, String topic) { + + return new KafkaSubscribableChannelSpec(template, containerFactory, topic, true); + } + + /** + * Create a spec for a pollable channel with the provided parameters. + * @param template the template. + * @param source the source. + * @return the spec. + * @since 3.3 + */ + public static KafkaPollableChannelSpec pollableChannel(KafkaTemplate template, + KafkaMessageSource source) { + + return new KafkaPollableChannelSpec(template, source); + } + private static KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec messageDrivenChannelAdapter( KafkaMessageListenerContainerSpec spec, KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPollableChannelSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPollableChannelSpec.java new file mode 100644 index 0000000000..09f8f6b500 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPollableChannelSpec.java @@ -0,0 +1,37 @@ +/* + * Copyright 2020 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.dsl; + +import org.springframework.integration.kafka.channel.PollableKafkaChannel; +import org.springframework.integration.kafka.inbound.KafkaMessageSource; +import org.springframework.kafka.core.KafkaTemplate; + +/** + * Spec for a pollable channel. + * + * @author Gary Russell + * @since 3.3 + * + */ +public class KafkaPollableChannelSpec extends AbstractKafkaChannelSpec { + + protected KafkaPollableChannelSpec(KafkaTemplate template, KafkaMessageSource source) { + super(null, null); + this.channel = new PollableKafkaChannel(template, source); + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaSubscribableChannelSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaSubscribableChannelSpec.java new file mode 100644 index 0000000000..a33bc2ac92 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaSubscribableChannelSpec.java @@ -0,0 +1,39 @@ +/* + * Copyright 2020 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.dsl; + +import org.springframework.integration.kafka.channel.SubscribableKafkaChannel; +import org.springframework.kafka.config.KafkaListenerContainerFactory; +import org.springframework.kafka.core.KafkaTemplate; + +/** + * Spec for a subscribable channel. + * + * @author Gary Russell + * @since 3.3 + * + */ +public class KafkaSubscribableChannelSpec extends AbstractKafkaChannelSpec { + + protected KafkaSubscribableChannelSpec(KafkaTemplate template, KafkaListenerContainerFactory factory, + String topic, boolean pubSub) { + + super(template, topic); + this.channel = new SubscribableKafkaChannel(template, factory, topic, pubSub); + } + +} 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 5eaae94780..70c7224a24 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 @@ -18,6 +18,63 @@ ]]> + + + + Creates a subscribable channel that is backed by a Kafka topic. + + + + + + + + + + + + + + Creates a pollable channel that is backed by a Kafka topic. + + + + + + + + + An 'org.springframework.integration.kafka.inbound.KafkaMessageSource' bean reference for + a pollable channel. + Mutually exclusive with 'listener-container'. + + + + + + + + + + + + + + + + + Creates a subscribable pub/sub channel that is backed by a Kafka topic. + + + + + + + + + + @@ -396,6 +453,56 @@ + + + + + + + + + + + + Unique ID for this Message Channel. + + + + + + + + + + + + + The topic name. + + + + + + + An 'org.springframework.kafka.config.KafkaListenerContainerFactory' bean reference for a + message-driven (subscribable) channel. + Mutually exclusive with 'message-source'. + + + + + + + + + + + + + @@ -641,6 +748,13 @@ + + + + Set the 'group.id' Kafka consumer property. + + + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java new file mode 100644 index 0000000000..2d3cabe777 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java @@ -0,0 +1,166 @@ +/* + * Copyright 2020 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.channnel; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.kafka.channel.PollableKafkaChannel; +import org.springframework.integration.kafka.channel.SubscribableKafkaChannel; +import org.springframework.integration.kafka.inbound.KafkaMessageSource; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.config.KafkaListenerContainerFactory; +import org.springframework.kafka.core.ConsumerFactory; +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.ConsumerProperties; +import org.springframework.kafka.support.KafkaHeaders; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.messaging.Message; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +/** + * @author Gary Russell + * @since 3.3 + * + */ +@SpringJUnitConfig +@EmbeddedKafka(topics = { "channel.1", "channel.2", "channel.3" }, partitions = 1) +public class ChannelTests { + + @Test + void subscribablePtp(@Autowired SubscribableChannel ptp) throws InterruptedException { + CountDownLatch latch = new CountDownLatch(1); + AtomicReference> message = new AtomicReference<>(); + ptp.subscribe(msg -> { + message.set(msg); + latch.countDown(); + }); + Message msg = new GenericMessage<>("foo"); + ptp.send(msg, 10_000L); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(message.get().getPayload()).isEqualTo("foo"); + assertThat(message.get().getHeaders().get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo("channel.1"); + } + + @Test + void pubSub(@Autowired SubscribableChannel pubSub) throws InterruptedException { + CountDownLatch latch = new CountDownLatch(2); + AtomicReference> message = new AtomicReference<>(); + pubSub.subscribe(msg -> { + message.set(msg); + latch.countDown(); + }); + pubSub.subscribe(msg -> { + message.set(msg); + latch.countDown(); + }); + Message msg = new GenericMessage<>("foo"); + pubSub.send(msg, 10_000L); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(message.get().getPayload()).isEqualTo("foo"); + assertThat(message.get().getHeaders().get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo("channel.2"); + } + + @Test + void pollable(@Autowired PollableChannel pollable) { + Message msg = new GenericMessage<>("foo"); + pollable.send(msg, 10_000L); + Message message = pollable.receive(10_000L); + assertThat(message.getPayload()).isEqualTo("foo"); + assertThat(message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo("channel.3"); + } + + @Configuration + public static class Config { + + @Autowired + private EmbeddedKafkaBroker broker; + + @Bean + public ProducerFactory pf() { + return new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(this.broker)); + } + + @Bean + public ConsumerFactory cf() { + Map consumerProps = KafkaTestUtils.consumerProps("channelTests", "false", this.broker); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + return new DefaultKafkaConsumerFactory<>(consumerProps); + } + + @Bean + public KafkaTemplate template(ProducerFactory pf) { + return new KafkaTemplate<>(pf); + } + + @Bean + public ConcurrentKafkaListenerContainerFactory factory(ConsumerFactory cf) { + ConcurrentKafkaListenerContainerFactory factory = + new ConcurrentKafkaListenerContainerFactory<>(); + factory.setConsumerFactory(cf); + return factory; + } + + @Bean + public SubscribableKafkaChannel ptp(KafkaTemplate template, + KafkaListenerContainerFactory factory) { + + SubscribableKafkaChannel channel = new SubscribableKafkaChannel(template, factory, "channel.1"); + channel.setGroupId("channel.1"); + return channel; + } + + @Bean + public SubscribableKafkaChannel pubSub(KafkaTemplate template, + KafkaListenerContainerFactory factory) { + + SubscribableKafkaChannel channel = new SubscribableKafkaChannel(template, factory, "channel.2", true); + channel.setGroupId("channel.2"); + return channel; + } + + @Bean + public KafkaMessageSource source(ConsumerFactory cf) { + return new KafkaMessageSource<>(cf, new ConsumerProperties("channel.3")); + } + + @Bean + public PollableKafkaChannel pollable(KafkaTemplate template, KafkaMessageSource source) { + return new PollableKafkaChannel(template, source); + } + + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/ChannelParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/ChannelParserTests.java new file mode 100644 index 0000000000..45fe258413 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/ChannelParserTests.java @@ -0,0 +1,107 @@ +/* + * Copyright 2020 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 static org.mockito.Mockito.mock; + +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.ImportResource; +import org.springframework.integration.kafka.channel.PollableKafkaChannel; +import org.springframework.integration.kafka.channel.SubscribableKafkaChannel; +import org.springframework.integration.kafka.inbound.KafkaMessageSource; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.config.KafkaListenerContainerFactory; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.KafkaOperations; +import org.springframework.kafka.listener.ConsumerProperties; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +/** + * @author Gary Russell + * @since 3.3 + * + */ +@SpringJUnitConfig +public class ChannelParserTests { + + @Autowired + KafkaListenerContainerFactory containerFactory; + + @Autowired + KafkaOperations template; + + @Autowired + KafkaMessageSource source; + + @Autowired + SubscribableKafkaChannel ptp; + + @Autowired + PollableKafkaChannel pollable; + + @Autowired + SubscribableKafkaChannel pubSub; + + @Test + void testParser() { + assertThat(KafkaTestUtils.getPropertyValue(this.ptp, "topic")).isEqualTo("ptpTopic"); + assertThat(KafkaTestUtils.getPropertyValue(this.pubSub, "topic")).isEqualTo("pubSubTopic"); + assertThat(KafkaTestUtils.getPropertyValue(this.ptp, "container")).isNotNull(); + assertThat(KafkaTestUtils.getPropertyValue(this.pubSub, "container")).isNotNull(); + assertThat(KafkaTestUtils.getPropertyValue(this.ptp, "template")).isSameAs(this.template); + assertThat(KafkaTestUtils.getPropertyValue(this.pubSub, "template")).isSameAs(this.template); + assertThat(KafkaTestUtils.getPropertyValue(this.pollable, "template")).isSameAs(this.template); + assertThat(KafkaTestUtils.getPropertyValue(this.pollable, "source")).isSameAs(this.source); + assertThat(KafkaTestUtils.getPropertyValue(this.ptp, "groupId")).isEqualTo("ptpGroup"); + assertThat(KafkaTestUtils.getPropertyValue(this.pubSub, "groupId")).isEqualTo("pubSubGroup"); + assertThat(KafkaTestUtils.getPropertyValue(this.pollable, "groupId")).isEqualTo("pollableGroup"); + } + + @Configuration + @ImportResource("org/springframework/integration/kafka/config/xml/channels-context.xml") + public static class Config { + + @SuppressWarnings("unchecked") + @Bean + public KafkaListenerContainerFactory containerFactory() { + ConcurrentKafkaListenerContainerFactory factory = + new ConcurrentKafkaListenerContainerFactory<>(); + factory.setConsumerFactory(mock(ConsumerFactory.class)); + return factory; + } + + @Bean + public KafkaOperations template() { + return mock(KafkaOperations.class); + } + + @SuppressWarnings("unchecked") + @Bean + public KafkaMessageSource source() { + ConsumerProperties properties = new ConsumerProperties("test"); + return new KafkaMessageSource(mock(ConsumerFactory.class), properties, true); + } + + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/channels-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/channels-context.xml new file mode 100644 index 0000000000..8c215da713 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/channels-context.xml @@ -0,0 +1,19 @@ + + + + + + + + + + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java index ca16ef513e..b5f1a548b8 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2020 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. @@ -44,7 +44,9 @@ import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.Pollers; import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer; +import org.springframework.integration.kafka.channel.PollableKafkaChannel; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; +import org.springframework.integration.kafka.inbound.KafkaMessageSource; import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy; import org.springframework.integration.support.MessageBuilder; @@ -91,7 +93,8 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; @SpringJUnitConfig @DirtiesContext @EmbeddedKafka(topics = { KafkaDslTests.TEST_TOPIC1, KafkaDslTests.TEST_TOPIC2, KafkaDslTests.TEST_TOPIC3, - KafkaDslTests.TEST_TOPIC4, KafkaDslTests.TEST_TOPIC5 }) + KafkaDslTests.TEST_TOPIC4, KafkaDslTests.TEST_TOPIC5, KafkaDslTests.TEST_TOPIC6, KafkaDslTests.TEST_TOPIC7, + KafkaDslTests.TEST_TOPIC8 }) public class KafkaDslTests { static final String TEST_TOPIC1 = "test-topic1"; @@ -104,6 +107,12 @@ public class KafkaDslTests { static final String TEST_TOPIC5 = "test-topic5"; + static final String TEST_TOPIC6 = "test-topic6"; + + static final String TEST_TOPIC7 = "test-topic7"; + + static final String TEST_TOPIC8 = "test-topic8"; + @Autowired @Qualifier("sendToKafkaFlow.input") private MessageChannel sendToKafkaFlowInput; @@ -213,6 +222,16 @@ public class KafkaDslTests { assertThat(this.gate.exchange(TEST_TOPIC4, "foo")).isEqualTo("FOO"); } + @Test + void channels(@Autowired MessageChannel topic6Channel, @Autowired PollableKafkaChannel topic8Channel) { + topic6Channel.send(new GenericMessage<>("foo")); + Message received = topic8Channel.receive(); + assertThat(received) + .isNotNull() + .extracting("payload") + .isEqualTo("foo"); + } + @Configuration @EnableIntegration @EnableKafka @@ -359,6 +378,33 @@ public class KafkaDslTests { .get(); } + @Bean + public KafkaSubscribableChannelSpec topic6Channel(KafkaTemplate template, + ConcurrentKafkaListenerContainerFactory containerFactory) { + return Kafka.channel(template, containerFactory, TEST_TOPIC6); + } + + @Bean + public KafkaTemplate template(ProducerFactory pf) { + return new KafkaTemplate<>(pf); + } + + @Bean + public KafkaMessageSource channelSource(ConsumerFactory cf) { + return new KafkaMessageSource<>(cf, new ConsumerProperties(TEST_TOPIC8)); + } + + @Bean + public IntegrationFlow channels(KafkaTemplate template, + ConcurrentKafkaListenerContainerFactory containerFactory, + KafkaMessageSource channelSource) { + + return IntegrationFlows.from(topic6Channel(template, containerFactory)) + .channel(Kafka.publishSubscribeChannel(template, containerFactory, TEST_TOPIC7)) + .channel(Kafka.pollableChannel(template, channelSource).id("topic8Channel")) + .get(); + } + private GenericMessageListenerContainer replyContainer() { ContainerProperties containerProperties = new ContainerProperties(TEST_TOPIC5); containerProperties.setGroupId("outGate");