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 fc7a17fb9a..47a6ee3bad 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 @@ -26,7 +26,9 @@ import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.AbstractMessageListenerContainer; +import org.springframework.kafka.listener.GenericMessageListenerContainer; import org.springframework.kafka.listener.config.ContainerProperties; +import org.springframework.kafka.requestreply.ReplyingKafkaTemplate; import org.springframework.kafka.support.TopicPartitionInitialOffset; /** @@ -34,6 +36,7 @@ import org.springframework.kafka.support.TopicPartitionInitialOffset; * * @author Artem Bilan * @author Nasko Vasilev + * @author Gary Russell * * @since 3.0 */ @@ -47,8 +50,8 @@ public final class Kafka { * @param the {@link KafkaProducerMessageHandlerSpec} extension type. * @return the KafkaProducerMessageHandlerSpec. */ - public static > KafkaProducerMessageHandlerSpec - outboundChannelAdapter(KafkaTemplate kafkaTemplate) { + public static > KafkaProducerMessageHandlerSpec outboundChannelAdapter( + KafkaTemplate kafkaTemplate) { return new KafkaProducerMessageHandlerSpec<>(kafkaTemplate); } @@ -61,8 +64,8 @@ public final class Kafka { * @return the KafkaProducerMessageHandlerSpec. * @see Kafka Producer Configs */ - public static KafkaProducerMessageHandlerSpec.KafkaProducerMessageHandlerTemplateSpec - outboundChannelAdapter(ProducerFactory producerFactory) { + public static KafkaProducerMessageHandlerSpec.KafkaProducerMessageHandlerTemplateSpec outboundChannelAdapter( + ProducerFactory producerFactory) { return new KafkaProducerMessageHandlerSpec.KafkaProducerMessageHandlerTemplateSpec<>(producerFactory); } @@ -77,8 +80,8 @@ public final class Kafka { * @return the spec. * @since 3.0.1 */ - public static KafkaInboundChannelAdapterSpec - inboundChannelAdapter(ConsumerFactory consumerFactory, String... topics) { + public static KafkaInboundChannelAdapterSpec inboundChannelAdapter( + ConsumerFactory consumerFactory, String... topics) { return new KafkaInboundChannelAdapterSpec<>(consumerFactory, topics); } @@ -94,9 +97,9 @@ public final class Kafka { * @return the spec. * @since 3.0.1 */ - public static KafkaInboundChannelAdapterSpec - inboundChannelAdapter(ConsumerFactory consumerFactory, - KafkaAckCallbackFactory ackCallbackFactory, String... topics) { + public static KafkaInboundChannelAdapterSpec inboundChannelAdapter( + ConsumerFactory consumerFactory, + KafkaAckCallbackFactory ackCallbackFactory, String... topics) { return new KafkaInboundChannelAdapterSpec<>(consumerFactory, ackCallbackFactory, topics); } @@ -109,8 +112,7 @@ public final class Kafka { * @param the {@link KafkaMessageDrivenChannelAdapterSpec} extension type. * @return the KafkaMessageDrivenChannelAdapterSpec. */ - public static > - KafkaMessageDrivenChannelAdapterSpec messageDrivenChannelAdapter( + public static > KafkaMessageDrivenChannelAdapterSpec messageDrivenChannelAdapter( AbstractMessageListenerContainer listenerContainer) { return messageDrivenChannelAdapter(listenerContainer, KafkaMessageDrivenChannelAdapter.ListenerMode.record); @@ -125,8 +127,7 @@ public final class Kafka { * @param the {@link KafkaMessageDrivenChannelAdapterSpec} extension type. * @return the KafkaMessageDrivenChannelAdapterSpec. */ - public static > - KafkaMessageDrivenChannelAdapterSpec messageDrivenChannelAdapter( + public static > KafkaMessageDrivenChannelAdapterSpec messageDrivenChannelAdapter( AbstractMessageListenerContainer listenerContainer, KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) { @@ -143,8 +144,8 @@ public final class Kafka { * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec. */ public static - KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec - messageDrivenChannelAdapter(ConsumerFactory consumerFactory, ContainerProperties containerProperties) { + KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec messageDrivenChannelAdapter( + ConsumerFactory consumerFactory, ContainerProperties containerProperties) { return messageDrivenChannelAdapter(consumerFactory, containerProperties, KafkaMessageDrivenChannelAdapter.ListenerMode.record); @@ -161,12 +162,12 @@ public final class Kafka { * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec. */ public static - KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec - messageDrivenChannelAdapter(ConsumerFactory consumerFactory, ContainerProperties containerProperties, + KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec messageDrivenChannelAdapter( + ConsumerFactory consumerFactory, ContainerProperties containerProperties, KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) { return messageDrivenChannelAdapter( - new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory, + new KafkaMessageListenerContainerSpec<>(consumerFactory, containerProperties), listenerMode); } @@ -180,8 +181,8 @@ public final class Kafka { * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec. */ public static - KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec - messageDrivenChannelAdapter(ConsumerFactory consumerFactory, + KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec messageDrivenChannelAdapter( + ConsumerFactory consumerFactory, TopicPartitionInitialOffset... topicPartitions) { return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record, @@ -199,13 +200,13 @@ public final class Kafka { * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec. */ public static - KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec - messageDrivenChannelAdapter(ConsumerFactory consumerFactory, + KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec messageDrivenChannelAdapter( + ConsumerFactory consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode, TopicPartitionInitialOffset... topicPartitions) { return messageDrivenChannelAdapter( - new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory, + new KafkaMessageListenerContainerSpec<>(consumerFactory, topicPartitions), listenerMode); } @@ -219,8 +220,8 @@ public final class Kafka { * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec. */ public static - KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec - messageDrivenChannelAdapter(ConsumerFactory consumerFactory, String... topics) { + KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec messageDrivenChannelAdapter( + ConsumerFactory consumerFactory, String... topics) { return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record, topics); @@ -237,12 +238,12 @@ public final class Kafka { * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec. */ public static - KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec - messageDrivenChannelAdapter(ConsumerFactory consumerFactory, - KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode, String... topics) { + KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec messageDrivenChannelAdapter( + ConsumerFactory consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode, + String... topics) { return messageDrivenChannelAdapter( - new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory, + new KafkaMessageListenerContainerSpec<>(consumerFactory, topics), listenerMode); } @@ -256,8 +257,8 @@ public final class Kafka { * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec. */ public static - KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec - messageDrivenChannelAdapter(ConsumerFactory consumerFactory, Pattern topicPattern) { + KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec messageDrivenChannelAdapter( + ConsumerFactory consumerFactory, Pattern topicPattern) { return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record, topicPattern); @@ -274,20 +275,108 @@ public final class Kafka { * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec. */ public static - KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec - messageDrivenChannelAdapter(ConsumerFactory consumerFactory, + KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec messageDrivenChannelAdapter( + ConsumerFactory consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode, Pattern topicPattern) { return messageDrivenChannelAdapter( - new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory, + new KafkaMessageListenerContainerSpec<>(consumerFactory, topicPattern), listenerMode); } + /** + * Create an initial {@link KafkaProducerMessageHandlerSpec}. + * @param kafkaTemplate the {@link ReplyingKafkaTemplate} to use + * @param the Kafka message key type. + * @param the Kafka message value type (request). + * @param the Kafka message value type (reply). + * @param the {@link KafkaOutboundGatewaySpec} extension type. + * @return the KafkaGatewayMessageHandlerSpec. + * @since 3.0.2 + */ + public static > KafkaOutboundGatewaySpec outboundGateway( + ReplyingKafkaTemplate kafkaTemplate) { + + return new KafkaOutboundGatewaySpec<>(kafkaTemplate); + } + + /** + * Create an initial {@link KafkaProducerMessageHandlerSpec} with ProducerFactory. + * @param producerFactory the {@link ProducerFactory} Java 8 Lambda. + * @param replyContainer a listener container for replies. + * @param the Kafka message key type. + * @param the Kafka message value type (request). + * @param the Kafka message value type (reply). + * @return the KafkaGatewayMessageHandlerSpec. + * @since 3.0.2 + */ + public static KafkaOutboundGatewaySpec.KafkaGatewayMessageHandlerTemplateSpec outboundGateway( + ProducerFactory producerFactory, GenericMessageListenerContainer replyContainer) { + + return new KafkaOutboundGatewaySpec.KafkaGatewayMessageHandlerTemplateSpec<>(producerFactory, + replyContainer); + } + + /** + * Create an initial {@link KafkaInboundGatewaySpec} with the provided container and + * template. + * @param container the container. + * @param template the template. + * @param the Kafka message key type. + * @param the Kafka message value type (request). + * @param the Kafka message value type (reply). + * @param the {@link KafkaInboundGatewaySpec} extension type. + * @return the spec. + * @since 3.0.2 + */ + public static > KafkaInboundGatewaySpec inboundGateway( + AbstractMessageListenerContainer container, KafkaTemplate template) { + + return new KafkaInboundGatewaySpec<>(container, template); + } + + /** + * Create an initial {@link KafkaInboundGatewaySpec} with the provided consumer factory, + * container properties and producer factory. + * @param consumerFactory the consumer factory. + * @param containerProperties the container properties. + * @param producerFactory the producer factory. + * @param the Kafka message key type. + * @param the Kafka message value type (request). + * @param the Kafka message value type (reply). + * @return the spec. + * @since 3.0.2 + */ + public static KafkaInboundGatewaySpec.KafkaInboundGatewayListenerContainerSpec inboundGateway( + ConsumerFactory consumerFactory, ContainerProperties containerProperties, + ProducerFactory producerFactory) { + + return inboundGateway( + new KafkaMessageListenerContainerSpec<>(consumerFactory, containerProperties), + new KafkaTemplateSpec<>(producerFactory)); + } + + /** + * Create an initial {@link KafkaInboundGatewaySpec} with the provided container and + * template specs. + * @param containerSpec the container spec. + * @param templateSpec the template spec. + * @param the Kafka message key type. + * @param the Kafka message value type (request). + * @param the Kafka message value type (reply). + * @return the spec. + * @since 3.0.2 + */ + public static KafkaInboundGatewaySpec.KafkaInboundGatewayListenerContainerSpec inboundGateway( + KafkaMessageListenerContainerSpec containerSpec, KafkaTemplateSpec templateSpec) { + + return new KafkaInboundGatewaySpec.KafkaInboundGatewayListenerContainerSpec<>(containerSpec, templateSpec); + } + private static - KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec - messageDrivenChannelAdapter(KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec spec, - KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) { + KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec messageDrivenChannelAdapter( + KafkaMessageListenerContainerSpec spec, KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) { return new KafkaMessageDrivenChannelAdapterSpec .KafkaMessageDrivenChannelAdapterListenerContainerSpec<>(spec, listenerMode); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java new file mode 100644 index 0000000000..bc4a5e04bd --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java @@ -0,0 +1,150 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.dsl; + +import java.util.Map; +import java.util.function.Consumer; + +import org.springframework.integration.dsl.ComponentsRegistration; +import org.springframework.integration.dsl.MessagingGatewaySpec; +import org.springframework.integration.kafka.inbound.KafkaInboundGateway; +import org.springframework.integration.support.ObjectStringMapBuilder; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.listener.AbstractMessageListenerContainer; +import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; +import org.springframework.kafka.support.converter.RecordMessageConverter; +import org.springframework.retry.RecoveryCallback; +import org.springframework.retry.support.RetryTemplate; +import org.springframework.util.Assert; + +/** + * A {@link MessagingGatewaySpec} implementation for the {@link KafkaInboundGateway}. + * + * @param the key type. + * @param the request value type. + * @param the reply value type. + * @param the target {@link KafkaInboundGatewaySpec} implementation type. + * + * @author Gary Russell + * + * @since 3.0.2 + */ +public class KafkaInboundGatewaySpec> + extends MessagingGatewaySpec> { + + KafkaInboundGatewaySpec(AbstractMessageListenerContainer messageListenerContainer, + KafkaTemplate kafkaTemplate) { + + super(new KafkaInboundGateway<>(messageListenerContainer, kafkaTemplate)); + } + + /** + * Set the message converter to use with a record-based consumer. + * @param messageConverter the converter. + * @return the spec + */ + public S messageConverter(RecordMessageConverter messageConverter) { + this.target.setMessageConverter(messageConverter); + return _this(); + } + + /** + * Specify a {@link RetryTemplate} instance to wrap + * {@code KafkaInboundGateway.IntegrationRecordMessageListener} into + * {@link org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter}. + * @param retryTemplate the {@link RetryTemplate} to use. + * @return the spec + */ + public S retryTemplate(RetryTemplate retryTemplate) { + this.target.setRetryTemplate(retryTemplate); + return _this(); + } + + /** + * A {@link RecoveryCallback} instance for retry operation; + * if null, the exception will be thrown to the container after retries are exhausted. + * Does not make sense if {@link #retryTemplate(RetryTemplate)} isn't specified. + * @param recoveryCallback the recovery callback. + * @return the spec + */ + public S recoveryCallback(RecoveryCallback recoveryCallback) { + this.target.setRecoveryCallback(recoveryCallback); + return _this(); + } + + /** + * A {@link ConcurrentMessageListenerContainer} configuration {@link KafkaInboundGatewaySpec} + * extension. + * @param the key type. + * @param the request value type. + * @param the reply value type. + */ + public static class KafkaInboundGatewayListenerContainerSpec extends + KafkaInboundGatewaySpec> + implements ComponentsRegistration { + + private final KafkaMessageListenerContainerSpec containerSpec; + + private final KafkaTemplateSpec templateSpec; + + KafkaInboundGatewayListenerContainerSpec(KafkaMessageListenerContainerSpec containerSpec, + KafkaTemplateSpec templateSpec) { + + super(containerSpec.get(), templateSpec.getTemplate()); + this.containerSpec = containerSpec; + this.templateSpec = templateSpec; + } + + /** + * Configure a listener container by invoking the {@link Consumer} callback, with a + * {@link KafkaMessageListenerContainerSpec} argument. + * @param configurer the configurer Java 8 Lambda. + * @return the spec. + */ + public KafkaInboundGatewayListenerContainerSpec configureListenerContainer( + Consumer> configurer) { + + Assert.notNull(configurer, "The 'configurer' cannot be null"); + configurer.accept(this.containerSpec); + return _this(); + } + + /** + * Configure a template by invoking the {@link Consumer} callback, with a + * {@link KafkaTemplateSpec} argument. + * @param configurer the configurer Java 8 Lambda. + * @return the spec. + */ + public KafkaInboundGatewayListenerContainerSpec configureTemplate( + Consumer> configurer) { + + Assert.notNull(configurer, "The 'configurer' cannot be null"); + configurer.accept(this.templateSpec); + return _this(); + } + + @Override + public Map getComponentsToRegister() { + return new ObjectStringMapBuilder() + .put(this.containerSpec.get(), this.containerSpec.getId()) + .put(this.templateSpec.get(), this.templateSpec.getId()) + .get(); + } + + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java index a022441622..c24f54136a 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java @@ -19,24 +19,13 @@ package org.springframework.integration.kafka.dsl; import java.util.Collections; import java.util.Map; import java.util.function.Consumer; -import java.util.regex.Pattern; -import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; -import org.apache.kafka.clients.consumer.OffsetCommitCallback; - -import org.springframework.core.task.AsyncListenableTaskExecutor; import org.springframework.integration.dsl.ComponentsRegistration; -import org.springframework.integration.dsl.IntegrationComponentSpec; import org.springframework.integration.dsl.MessageProducerSpec; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; -import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.listener.AbstractMessageListenerContainer; -import org.springframework.kafka.listener.AcknowledgingMessageListener; import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; -import org.springframework.kafka.listener.ErrorHandler; import org.springframework.kafka.listener.adapter.RecordFilterStrategy; -import org.springframework.kafka.listener.config.ContainerProperties; -import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.kafka.support.converter.BatchMessageConverter; import org.springframework.kafka.support.converter.MessageConverter; import org.springframework.kafka.support.converter.RecordMessageConverter; @@ -177,7 +166,7 @@ public class KafkaMessageDrivenChannelAdapterSpec spec, KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) { - super(spec.container, listenerMode); + super(spec.get(), listenerMode); this.spec = spec; } @@ -196,225 +185,7 @@ public class KafkaMessageDrivenChannelAdapterSpec getComponentsToRegister() { - return Collections.singletonMap(this.spec.container, this.spec.getId()); - } - - } - - /** - * A helper class in the Builder pattern style to delegate options to the - * {@link ConcurrentMessageListenerContainer}. - * - * @param the key type. - * @param the value type. - */ - public static class KafkaMessageListenerContainerSpec - extends IntegrationComponentSpec, ConcurrentMessageListenerContainer> { - - private final ConcurrentMessageListenerContainer container; - - - KafkaMessageListenerContainerSpec(ConsumerFactory consumerFactory, - ContainerProperties containerProperties) { - this.container = new ConcurrentMessageListenerContainer<>(consumerFactory, containerProperties); - } - - KafkaMessageListenerContainerSpec(ConsumerFactory consumerFactory, - TopicPartitionInitialOffset... topicPartitions) { - this(consumerFactory, new ContainerProperties(topicPartitions)); - } - - KafkaMessageListenerContainerSpec(ConsumerFactory consumerFactory, String... topics) { - this(consumerFactory, new ContainerProperties(topics)); - } - - KafkaMessageListenerContainerSpec(ConsumerFactory consumerFactory, Pattern topicPattern) { - this(consumerFactory, new ContainerProperties(topicPattern)); - } - - @Override - public KafkaMessageListenerContainerSpec id(String id) { - return super.id(id); - } - - /** - * Specify a concurrency maximum number for the {@link AbstractMessageListenerContainer}. - * @param concurrency the concurrency maximum number. - * @return the spec. - * @see ConcurrentMessageListenerContainer#setConcurrency(int) - */ - public KafkaMessageListenerContainerSpec concurrency(int concurrency) { - this.container.setConcurrency(concurrency); - return this; - } - - /** - * Specify an {@link ErrorHandler} for the {@link AbstractMessageListenerContainer}. - * @param errorHandler the {@link ErrorHandler}. - * @return the spec. - * @see ErrorHandler - */ - public KafkaMessageListenerContainerSpec errorHandler(ErrorHandler errorHandler) { - this.container.getContainerProperties().setErrorHandler(errorHandler); - return this; - } - - /** - * Set the ack mode to use when auto ack (in the configuration properties) is false. - *
    - *
  • RECORD: Ack after each record has been passed to the listener.
    • - *
  • BATCH: Ack after each batch of records received from the consumer has been - * passed to the listener
    • - *
  • TIME: Ack after this number of milliseconds; (should be greater than - * {@code #setPollTimeout(long) pollTimeout}.
    • - *
  • COUNT: Ack after at least this number of records have been received
    • - *
  • MANUAL: Listener is responsible for acking - use a - * {@link AcknowledgingMessageListener}. - *
- * @param ackMode the {@link AbstractMessageListenerContainer.AckMode}; default BATCH. - * @return the spec. - * @see AbstractMessageListenerContainer.AckMode - */ - public KafkaMessageListenerContainerSpec ackMode(AbstractMessageListenerContainer.AckMode ackMode) { - this.container.getContainerProperties().setAckMode(ackMode); - return this; - } - - /** - * Set the max time to block in the consumer waiting for records. - * @param pollTimeout the timeout in ms; default 1000. - * @return the spec. - * @see ContainerProperties#setPollTimeout(long) - */ - public KafkaMessageListenerContainerSpec pollTimeout(long pollTimeout) { - this.container.getContainerProperties().setPollTimeout(pollTimeout); - return this; - } - - /** - * Set the number of outstanding record count after which offsets should be - * committed when {@link AbstractMessageListenerContainer.AckMode#COUNT} - * or {@link AbstractMessageListenerContainer.AckMode#COUNT_TIME} is being used. - * @param count the count - * @return the spec. - * @see ContainerProperties#setAckCount(int) - */ - public KafkaMessageListenerContainerSpec ackCount(int count) { - this.container.getContainerProperties().setAckCount(count); - return this; - } - - /** - * Set the time (ms) after which outstanding offsets should be committed when - * {@link AbstractMessageListenerContainer.AckMode#TIME} or - * {@link AbstractMessageListenerContainer.AckMode#COUNT_TIME} is being used. - * Should be larger than zero. - * @param millis the time - * @return the spec. - * @see ContainerProperties#setAckTime(long) - */ - public KafkaMessageListenerContainerSpec ackTime(long millis) { - this.container.getContainerProperties().setAckTime(millis); - return this; - } - - /** - * Set the executor for threads that poll the consumer. - * @param consumerTaskExecutor the executor - * @return the spec. - * @see ContainerProperties#setConsumerTaskExecutor(AsyncListenableTaskExecutor) - */ - public KafkaMessageListenerContainerSpec consumerTaskExecutor( - AsyncListenableTaskExecutor consumerTaskExecutor) { - this.container.getContainerProperties().setConsumerTaskExecutor(consumerTaskExecutor); - return this; - } - - /** - * Set the timeout for shutting down the container. This is the maximum amount of - * time that the invocation to {@code #stop(Runnable)} will block for, before - * returning. - * @param shutdownTimeout the shutdown timeout. - * @return the spec. - * @see ContainerProperties#setShutdownTimeout(long) - */ - public KafkaMessageListenerContainerSpec shutdownTimeout(long shutdownTimeout) { - this.container.getContainerProperties().setShutdownTimeout(shutdownTimeout); - return this; - } - - /** - * Set the user defined {@link ConsumerRebalanceListener} implementation. - * @param consumerRebalanceListener the {@link ConsumerRebalanceListener} instance - * @return the spec. - * @see ContainerProperties#setConsumerRebalanceListener(ConsumerRebalanceListener) - */ - public KafkaMessageListenerContainerSpec consumerRebalanceListener( - ConsumerRebalanceListener consumerRebalanceListener) { - this.container.getContainerProperties().setConsumerRebalanceListener(consumerRebalanceListener); - return this; - } - - /** - * Set the commit callback; by default a simple logging callback is used to log - * success at DEBUG level and failures at ERROR level. - * @param commitCallback the callback. - * @return the spec. - * @see ContainerProperties#setCommitCallback(OffsetCommitCallback) - */ - public KafkaMessageListenerContainerSpec commitCallback(OffsetCommitCallback commitCallback) { - this.container.getContainerProperties().setCommitCallback(commitCallback); - return this; - } - - /** - * Set whether or not to call consumer.commitSync() or commitAsync() when the - * container is responsible for commits. Default true. See - * https://github.com/spring-projects/spring-kafka/issues/62 At the time of - * writing, async commits are not entirely reliable. - * @param syncCommits true to use commitSync(). - * @return the spec. - * @see ContainerProperties#setSyncCommits(boolean) - */ - public KafkaMessageListenerContainerSpec syncCommits(boolean syncCommits) { - this.container.getContainerProperties().setSyncCommits(syncCommits); - return this; - } - - /** - * Set the idle event interval; when set, an event is emitted if a poll returns - * no records and this interval has elapsed since a record was returned. - * @param idleEventInterval the interval. - * @return the spec. - * @see ContainerProperties#setIdleEventInterval(Long) - */ - public KafkaMessageListenerContainerSpec idleEventInterval(Long idleEventInterval) { - this.container.getContainerProperties().setIdleEventInterval(idleEventInterval); - return this; - } - - /** - * Set whether the container should ack messages that throw exceptions or not. - * @param ackOnError whether the container should acknowledge messages that throw - * exceptions. - * @return the spec. - * @see ContainerProperties#setAckOnError(boolean) - */ - public KafkaMessageListenerContainerSpec ackOnError(boolean ackOnError) { - this.container.getContainerProperties().setAckOnError(ackOnError); - return this; - } - - /** - * Set the group id for this container. Overrides any {@code group.id} property - * provided by the consumer factory configuration. - * @param groupId the group id. - * @return the spec. - * @see ContainerProperties#setAckOnError(boolean) - */ - public KafkaMessageListenerContainerSpec groupId(String groupId) { - this.container.getContainerProperties().setGroupId(groupId); - return this; + return Collections.singletonMap(this.spec.get(), this.spec.getId()); } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java new file mode 100644 index 0000000000..e3929be86a --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java @@ -0,0 +1,256 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.dsl; + +import java.util.regex.Pattern; + +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.clients.consumer.OffsetCommitCallback; + +import org.springframework.core.task.AsyncListenableTaskExecutor; +import org.springframework.integration.dsl.IntegrationComponentSpec; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.listener.AbstractMessageListenerContainer; +import org.springframework.kafka.listener.AcknowledgingMessageListener; +import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; +import org.springframework.kafka.listener.ErrorHandler; +import org.springframework.kafka.listener.config.ContainerProperties; +import org.springframework.kafka.support.TopicPartitionInitialOffset; + +/** + * A helper class in the Builder pattern style to delegate options to the + * {@link ConcurrentMessageListenerContainer}. + * + * @param the key type. + * @param the value type. + * + * @author Artem Bilan + * @author Gary Russell + * + * @since 3.0 + */ +public class KafkaMessageListenerContainerSpec + extends IntegrationComponentSpec, ConcurrentMessageListenerContainer> { + + KafkaMessageListenerContainerSpec(ConsumerFactory consumerFactory, + ContainerProperties containerProperties) { + + this.target = new ConcurrentMessageListenerContainer<>(consumerFactory, containerProperties); + } + + KafkaMessageListenerContainerSpec(ConsumerFactory consumerFactory, + TopicPartitionInitialOffset... topicPartitions) { + + this(consumerFactory, new ContainerProperties(topicPartitions)); + } + + KafkaMessageListenerContainerSpec(ConsumerFactory consumerFactory, String... topics) { + this(consumerFactory, new ContainerProperties(topics)); + } + + KafkaMessageListenerContainerSpec(ConsumerFactory consumerFactory, Pattern topicPattern) { + this(consumerFactory, new ContainerProperties(topicPattern)); + } + + @Override + public KafkaMessageListenerContainerSpec id(String id) { + return super.id(id); + } + + /** + * Specify a concurrency maximum number for the {@link AbstractMessageListenerContainer}. + * @param concurrency the concurrency maximum number. + * @return the spec. + * @see ConcurrentMessageListenerContainer#setConcurrency(int) + */ + public KafkaMessageListenerContainerSpec concurrency(int concurrency) { + this.target.setConcurrency(concurrency); + return this; + } + + /** + * Specify an {@link ErrorHandler} for the {@link AbstractMessageListenerContainer}. + * @param errorHandler the {@link ErrorHandler}. + * @return the spec. + * @see ErrorHandler + */ + public KafkaMessageListenerContainerSpec errorHandler(ErrorHandler errorHandler) { + this.target.getContainerProperties().setErrorHandler(errorHandler); + return this; + } + + /** + * Set the ack mode to use when auto ack (in the configuration properties) is false. + *
    + *
  • RECORD: Ack after each record has been passed to the listener.
    • + *
  • BATCH: Ack after each batch of records received from the consumer has been + * passed to the listener
    • + *
  • TIME: Ack after this number of milliseconds; (should be greater than + * {@code #setPollTimeout(long) pollTimeout}.
    • + *
  • COUNT: Ack after at least this number of records have been received
    • + *
  • MANUAL: Listener is responsible for acking - use a + * {@link AcknowledgingMessageListener}. + *
+ * @param ackMode the {@link AbstractMessageListenerContainer.AckMode}; default BATCH. + * @return the spec. + * @see AbstractMessageListenerContainer.AckMode + */ + public KafkaMessageListenerContainerSpec ackMode(AbstractMessageListenerContainer.AckMode ackMode) { + this.target.getContainerProperties().setAckMode(ackMode); + return this; + } + + /** + * Set the max time to block in the consumer waiting for records. + * @param pollTimeout the timeout in ms; default 1000. + * @return the spec. + * @see ContainerProperties#setPollTimeout(long) + */ + public KafkaMessageListenerContainerSpec pollTimeout(long pollTimeout) { + this.target.getContainerProperties().setPollTimeout(pollTimeout); + return this; + } + + /** + * Set the number of outstanding record count after which offsets should be + * committed when {@link AbstractMessageListenerContainer.AckMode#COUNT} + * or {@link AbstractMessageListenerContainer.AckMode#COUNT_TIME} is being used. + * @param count the count + * @return the spec. + * @see ContainerProperties#setAckCount(int) + */ + public KafkaMessageListenerContainerSpec ackCount(int count) { + this.target.getContainerProperties().setAckCount(count); + return this; + } + + /** + * Set the time (ms) after which outstanding offsets should be committed when + * {@link AbstractMessageListenerContainer.AckMode#TIME} or + * {@link AbstractMessageListenerContainer.AckMode#COUNT_TIME} is being used. + * Should be larger than zero. + * @param millis the time + * @return the spec. + * @see ContainerProperties#setAckTime(long) + */ + public KafkaMessageListenerContainerSpec ackTime(long millis) { + this.target.getContainerProperties().setAckTime(millis); + return this; + } + + /** + * Set the executor for threads that poll the consumer. + * @param consumerTaskExecutor the executor + * @return the spec. + * @see ContainerProperties#setConsumerTaskExecutor(AsyncListenableTaskExecutor) + */ + public KafkaMessageListenerContainerSpec consumerTaskExecutor( + AsyncListenableTaskExecutor consumerTaskExecutor) { + + this.target.getContainerProperties().setConsumerTaskExecutor(consumerTaskExecutor); + return this; + } + + /** + * Set the timeout for shutting down the container. This is the maximum amount of + * time that the invocation to {@code #stop(Runnable)} will block for, before + * returning. + * @param shutdownTimeout the shutdown timeout. + * @return the spec. + * @see ContainerProperties#setShutdownTimeout(long) + */ + public KafkaMessageListenerContainerSpec shutdownTimeout(long shutdownTimeout) { + this.target.getContainerProperties().setShutdownTimeout(shutdownTimeout); + return this; + } + + /** + * Set the user defined {@link ConsumerRebalanceListener} implementation. + * @param consumerRebalanceListener the {@link ConsumerRebalanceListener} instance + * @return the spec. + * @see ContainerProperties#setConsumerRebalanceListener(ConsumerRebalanceListener) + */ + public KafkaMessageListenerContainerSpec consumerRebalanceListener( + ConsumerRebalanceListener consumerRebalanceListener) { + + this.target.getContainerProperties().setConsumerRebalanceListener(consumerRebalanceListener); + return this; + } + + /** + * Set the commit callback; by default a simple logging callback is used to log + * success at DEBUG level and failures at ERROR level. + * @param commitCallback the callback. + * @return the spec. + * @see ContainerProperties#setCommitCallback(OffsetCommitCallback) + */ + public KafkaMessageListenerContainerSpec commitCallback(OffsetCommitCallback commitCallback) { + this.target.getContainerProperties().setCommitCallback(commitCallback); + return this; + } + + /** + * Set whether or not to call consumer.commitSync() or commitAsync() when the + * container is responsible for commits. Default true. See + * https://github.com/spring-projects/spring-kafka/issues/62 At the time of + * writing, async commits are not entirely reliable. + * @param syncCommits true to use commitSync(). + * @return the spec. + * @see ContainerProperties#setSyncCommits(boolean) + */ + public KafkaMessageListenerContainerSpec syncCommits(boolean syncCommits) { + this.target.getContainerProperties().setSyncCommits(syncCommits); + return this; + } + + /** + * Set the idle event interval; when set, an event is emitted if a poll returns + * no records and this interval has elapsed since a record was returned. + * @param idleEventInterval the interval. + * @return the spec. + * @see ContainerProperties#setIdleEventInterval(Long) + */ + public KafkaMessageListenerContainerSpec idleEventInterval(Long idleEventInterval) { + this.target.getContainerProperties().setIdleEventInterval(idleEventInterval); + return this; + } + + /** + * Set whether the container should ack messages that throw exceptions or not. + * @param ackOnError whether the container should acknowledge messages that throw + * exceptions. + * @return the spec. + * @see ContainerProperties#setAckOnError(boolean) + */ + public KafkaMessageListenerContainerSpec ackOnError(boolean ackOnError) { + this.target.getContainerProperties().setAckOnError(ackOnError); + return this; + } + + /** + * Set the group id for this container. Overrides any {@code group.id} property + * provided by the consumer factory configuration. + * @param groupId the group id. + * @return the spec. + * @see ContainerProperties#setAckOnError(boolean) + */ + public KafkaMessageListenerContainerSpec groupId(String groupId) { + this.target.getContainerProperties().setGroupId(groupId); + return this; + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaOutboundGatewaySpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaOutboundGatewaySpec.java new file mode 100644 index 0000000000..04449f13dd --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaOutboundGatewaySpec.java @@ -0,0 +1,135 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.dsl; + +import java.util.Collections; +import java.util.Map; +import java.util.function.Consumer; + +import org.springframework.integration.dsl.ComponentsRegistration; +import org.springframework.integration.dsl.IntegrationComponentSpec; +import org.springframework.integration.dsl.MessageHandlerSpec; +import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.listener.GenericMessageListenerContainer; +import org.springframework.kafka.requestreply.ReplyingKafkaTemplate; +import org.springframework.kafka.support.converter.RecordMessageConverter; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.util.Assert; + +/** + * A {@link MessageHandlerSpec} implementation for the {@link KafkaProducerMessageHandler} + * as a gateway. + * @param the key type. + * @param the outbound value type. + * @param the reply value type. + * @param the {@link KafkaProducerMessageHandlerSpec} extension type. + * + * @author Gary Russell + * + * @since 3.0.2 + * + */ +public class KafkaOutboundGatewaySpec> + extends KafkaProducerMessageHandlerSpec { + + KafkaOutboundGatewaySpec(ReplyingKafkaTemplate kafkaTemplate) { + super(kafkaTemplate); + } + + /** + * Set a message converter for replies (when a gateway). + * @param messageConverter the converter. + * @return the spec. + */ + public S replyMessageConverter(RecordMessageConverter messageConverter) { + this.target.setReplyMessageConverter(messageConverter); + return _this(); + } + + /** + * A {@link KafkaTemplate}-based {@link KafkaProducerMessageHandlerSpec} extension. + * + * @param the key type. + * @param the outbound value type. + * @param the reply value type. + */ + public static class KafkaGatewayMessageHandlerTemplateSpec + extends KafkaOutboundGatewaySpec> + implements ComponentsRegistration { + + private final ReplyingKafkaTemplateSpec kafkaTemplateSpec; + + @SuppressWarnings("unchecked") + KafkaGatewayMessageHandlerTemplateSpec(ProducerFactory producerFactory, + GenericMessageListenerContainer replyContainer) { + + super(new ReplyingKafkaTemplate<>(producerFactory, replyContainer)); + this.kafkaTemplateSpec = + new ReplyingKafkaTemplateSpec<>((ReplyingKafkaTemplate) this.target.getKafkaTemplate()); + } + + /** + * Configure a Kafka Template by invoking the {@link Consumer} callback, with a + * {@link KafkaTemplateSpec} argument. + * @param configurer the configurer Java 8 Lambda. + * @return the spec. + */ + public KafkaGatewayMessageHandlerTemplateSpec configureKafkaTemplate( + Consumer> configurer) { + + Assert.notNull(configurer, "The 'configurer' cannot be null"); + configurer.accept(this.kafkaTemplateSpec); + return _this(); + } + + @Override + public Map getComponentsToRegister() { + return Collections.singletonMap(this.kafkaTemplateSpec.get(), this.kafkaTemplateSpec.getId()); + } + + } + + /** + * An {@link IntegrationComponentSpec} implementation for the {@link KafkaTemplate}. + * + * @param the key type. + * @param the request value type. + * @param the reply value type. + */ + public static class ReplyingKafkaTemplateSpec extends KafkaTemplateSpec { + + ReplyingKafkaTemplateSpec(ReplyingKafkaTemplate kafkaTemplate) { + super(kafkaTemplate); + } + + @SuppressWarnings("unchecked") + ReplyingKafkaTemplateSpec taskScheduler(TaskScheduler scheduler) { + ((ReplyingKafkaTemplate) this.target).setTaskScheduler(scheduler); + return this; + } + + @SuppressWarnings("unchecked") + ReplyingKafkaTemplateSpec replyTimeout(long replyTimeout) { + ((ReplyingKafkaTemplate) this.target).setReplyTimeout(replyTimeout); + return this; + } + + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java index d12c805e14..ce74da2de7 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-2018 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. @@ -24,7 +24,6 @@ import java.util.function.Function; import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.dsl.ComponentsRegistration; -import org.springframework.integration.dsl.IntegrationComponentSpec; import org.springframework.integration.dsl.MessageHandlerSpec; import org.springframework.integration.expression.FunctionExpression; import org.springframework.integration.expression.ValueExpression; @@ -32,10 +31,8 @@ import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandle import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.support.KafkaHeaderMapper; -import org.springframework.kafka.support.LoggingProducerListener; -import org.springframework.kafka.support.ProducerListener; -import org.springframework.kafka.support.converter.RecordMessageConverter; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; /** @@ -47,6 +44,7 @@ import org.springframework.util.Assert; * * @author Artem Bilan * @author Biju Kunjummen + * @author Gary Russell * * @since 3.0 */ @@ -268,6 +266,50 @@ public class KafkaProducerMessageHandlerSpec the key type. - * @param the value type. - */ - public static class KafkaTemplateSpec - extends IntegrationComponentSpec, KafkaTemplate> { - - KafkaTemplateSpec(KafkaTemplate kafkaTemplate) { - this.target = kafkaTemplate; - } - - @Override - public KafkaTemplateSpec id(String id) { - return super.id(id); - } - - /** - /** - * Set the default topic for send methods where a topic is not - * providing. - * @param defaultTopic the topic. - * @return the spec - */ - public KafkaTemplateSpec defaultTopic(String defaultTopic) { - this.target.setDefaultTopic(defaultTopic); - return this; - } - - /** - * Set a {@link ProducerListener} which will be invoked when Kafka acknowledges - * a send operation. By default a {@link LoggingProducerListener} is configured - * which logs errors only. - * @param producerListener the listener; may be {@code null}. - * @return the spec - */ - public KafkaTemplateSpec producerListener(ProducerListener producerListener) { - this.target.setProducerListener(producerListener); - return this; - } - - /** - * Set the message converter to use. - * @param messageConverter the message converter. - * @return the spec - */ - public KafkaTemplateSpec messageConverter(RecordMessageConverter messageConverter) { - this.target.setMessageConverter(messageConverter); - return this; - } - - } - } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaTemplateSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaTemplateSpec.java new file mode 100644 index 0000000000..ec8b574bd5 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaTemplateSpec.java @@ -0,0 +1,91 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.dsl; + +import org.springframework.integration.dsl.IntegrationComponentSpec; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.support.LoggingProducerListener; +import org.springframework.kafka.support.ProducerListener; +import org.springframework.kafka.support.converter.RecordMessageConverter; + +/** + * An {@link IntegrationComponentSpec} implementation for the {@link KafkaTemplate}. + * + * @param the key type. + * @param the value type. + * + * @author Artem Bilan + * @author Gary Russell + * + * @since 3.0 + */ +public class KafkaTemplateSpec + extends IntegrationComponentSpec, KafkaTemplate> { + + KafkaTemplateSpec(KafkaTemplate kafkaTemplate) { + this.target = kafkaTemplate; + } + + KafkaTemplateSpec(ProducerFactory producerFactory) { + this.target = new KafkaTemplate<>(producerFactory); + } + + public KafkaTemplate getTemplate() { + return this.target; + } + + @Override + public KafkaTemplateSpec id(String id) { + return super.id(id); + } + + /** + /** + * Set the default topic for send methods where a topic is not + * providing. + * @param defaultTopic the topic. + * @return the spec + */ + public KafkaTemplateSpec defaultTopic(String defaultTopic) { + this.target.setDefaultTopic(defaultTopic); + return this; + } + + /** + * Set a {@link ProducerListener} which will be invoked when Kafka acknowledges + * a send operation. By default a {@link LoggingProducerListener} is configured + * which logs errors only. + * @param producerListener the listener; may be {@code null}. + * @return the spec + */ + public KafkaTemplateSpec producerListener(ProducerListener producerListener) { + this.target.setProducerListener(producerListener); + return this; + } + + /** + * Set the message converter to use. + * @param messageConverter the message converter. + * @return the spec + */ + public KafkaTemplateSpec messageConverter(RecordMessageConverter messageConverter) { + this.target.setMessageConverter(messageConverter); + return this; + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java new file mode 100644 index 0000000000..66975f4ba5 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java @@ -0,0 +1,318 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.inbound; + +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; + +import org.springframework.core.AttributeAccessor; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.context.OrderlyShutdownCapable; +import org.springframework.integration.gateway.MessagingGatewaySupport; +import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; +import org.springframework.integration.support.ErrorMessageStrategy; +import org.springframework.integration.support.ErrorMessageUtils; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.listener.AbstractMessageListenerContainer; +import org.springframework.kafka.listener.MessageListener; +import org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter; +import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter; +import org.springframework.kafka.support.Acknowledgment; +import org.springframework.kafka.support.KafkaHeaders; +import org.springframework.kafka.support.converter.BatchMessageConverter; +import org.springframework.kafka.support.converter.ConversionException; +import org.springframework.kafka.support.converter.KafkaMessageHeaders; +import org.springframework.kafka.support.converter.RecordMessageConverter; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.retry.RecoveryCallback; +import org.springframework.retry.RetryCallback; +import org.springframework.retry.RetryContext; +import org.springframework.retry.RetryListener; +import org.springframework.retry.support.RetryTemplate; +import org.springframework.util.Assert; + +/** + * Inbound gateway. + * + * @param the key type. + * @param the request value type. + * @param the reply value type. + * + * @author Gary Russell + * + * @since 3.0.2 + * + */ +public class KafkaInboundGateway extends MessagingGatewaySupport implements OrderlyShutdownCapable { + + private static final ThreadLocal attributesHolder = new ThreadLocal<>(); + + private final IntegrationRecordMessageListener listener = new IntegrationRecordMessageListener(); + + private final AbstractMessageListenerContainer messageListenerContainer; + + private final KafkaTemplate kafkaTemplate; + + private RetryTemplate retryTemplate; + + private RecoveryCallback recoveryCallback; + + /** + * Construct an instance with the provided container. + * @param messageListenerContainer the container. + * @param kafkaTemplate the kafka template. + */ + public KafkaInboundGateway(AbstractMessageListenerContainer messageListenerContainer, + KafkaTemplate kafkaTemplate) { + + Assert.notNull(messageListenerContainer, "messageListenerContainer is required"); + Assert.notNull(kafkaTemplate, "kafkaTemplate is required"); + Assert.isNull(messageListenerContainer.getContainerProperties().getMessageListener(), + "Container must not already have a listener"); + this.messageListenerContainer = messageListenerContainer; + this.messageListenerContainer.setAutoStartup(false); + this.kafkaTemplate = kafkaTemplate; + setErrorMessageStrategy(new RawRecordHeaderErrorMessageStrategy()); + } + + /** + * Set the message converter; must be a {@link RecordMessageConverter} or + * {@link BatchMessageConverter} depending on mode. + * @param messageConverter the converter. + */ + public void setMessageConverter(RecordMessageConverter messageConverter) { + this.listener.setMessageConverter(messageConverter); + } + + /** + * When using a type-aware message converter (such as {@code StringJsonMessageConverter}, + * set the payload type the converter should create. Defaults to {@link Object}. + * @param payloadType the type. + */ + public void setPayloadType(Class payloadType) { + this.listener.setFallbackType(payloadType); + } + + /** + * Specify a {@link RetryTemplate} instance to wrap + * {@link KafkaInboundGateway.IntegrationRecordMessageListener} into + * {@link RetryingMessageListenerAdapter}. + * @param retryTemplate the {@link RetryTemplate} to use. + */ + public void setRetryTemplate(RetryTemplate retryTemplate) { + this.retryTemplate = retryTemplate; + } + + /** + * A {@link RecoveryCallback} instance for retry operation; + * if null, the exception will be thrown to the container after retries are exhausted + * (unless an error channel is configured). + * Does not make sense if {@link #setRetryTemplate(RetryTemplate)} isn't specified. + * @param recoveryCallback the recovery callback. + */ + public void setRecoveryCallback(RecoveryCallback recoveryCallback) { + this.recoveryCallback = recoveryCallback; + } + + @Override + protected void onInit() throws Exception { + super.onInit(); + MessageListener listener = this.listener; + if (this.retryTemplate != null) { + listener = new RetryingMessageListenerAdapter<>(listener, this.retryTemplate, + this.recoveryCallback); + this.retryTemplate.registerListener(this.listener); + } + this.messageListenerContainer.getContainerProperties().setMessageListener(listener); + } + + @Override + protected void doStart() { + this.messageListenerContainer.start(); + } + + @Override + protected void doStop() { + this.messageListenerContainer.stop(); + } + + @Override + public String getComponentType() { + return "kafka:inbound-gateway"; + } + + @Override + public int beforeShutdown() { + this.messageListenerContainer.stop(); + return getPhase(); + } + + @Override + public int afterShutdown() { + return getPhase(); + } + + /** + * If there's a retry template, it will set the attributes holder via the listener. If + * there's no retry template, but there's an error channel, we create a new attributes + * holder here. If an attributes holder exists (by either method), we set the + * attributes for use by the {@link ErrorMessageStrategy}. + * @param record the record. + * @param message the message. + */ + private void setAttributesIfNecessary(Object record, Message message) { + boolean needHolder = getErrorChannel() != null && this.retryTemplate == null; + boolean needAttributes = needHolder | this.retryTemplate != null; + if (needHolder) { + attributesHolder.set(ErrorMessageUtils.getAttributeAccessor(null, null)); + } + if (needAttributes) { + AttributeAccessor attributes = attributesHolder.get(); + if (attributes != null) { + attributes.setAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY, message); + attributes.setAttribute(KafkaHeaders.RAW_DATA, record); + } + } + } + + @Override + protected AttributeAccessor getErrorMessageAttributes(Message message) { + AttributeAccessor attributes = attributesHolder.get(); + if (attributes == null) { + return super.getErrorMessageAttributes(message); + } + else { + return attributes; + } + } + + private class IntegrationRecordMessageListener extends RecordMessagingMessageListenerAdapter + implements RetryListener { + + IntegrationRecordMessageListener() { + super(null, null); + } + + @Override + public void onMessage(ConsumerRecord record, Acknowledgment acknowledgment, Consumer consumer) { + Message message = null; + try { + message = toMessagingMessage(record, acknowledgment, consumer); + if (KafkaInboundGateway.this.retryTemplate != null) { + message = addDeliveryAttemptHeader(message); + } + setAttributesIfNecessary(record, message); + } + catch (RuntimeException e) { + if (getErrorChannel() != null) { + KafkaInboundGateway.this.messagingTemplate.send(getErrorChannel(), buildErrorMessage(null, + new ConversionException("Failed to convert to message for: " + record, e))); + } + } + if (message != null) { + try { + Message reply = sendAndReceiveMessage(message); + if (reply != null) { + reply = enhanceReply(message, reply); + KafkaInboundGateway.this.kafkaTemplate.send(reply); + } + } + finally { + if (KafkaInboundGateway.this.retryTemplate == null) { + attributesHolder.remove(); + } + } + } + else { + KafkaInboundGateway.this.logger.debug("Converter returned a null message for: " + + record); + } + } + + private Message addDeliveryAttemptHeader(Message message) { + Message messageToReturn = message; + AtomicInteger deliveryAttempt = + new AtomicInteger(((RetryContext) attributesHolder.get()).getRetryCount() + 1); + if (message.getHeaders() instanceof KafkaMessageHeaders) { + ((KafkaMessageHeaders) message.getHeaders()).getRawHeaders() + .put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, deliveryAttempt); + } + else { + messageToReturn = MessageBuilder.fromMessage(message) + .setHeader(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, deliveryAttempt) + .build(); + } + return messageToReturn; + } + + private Message enhanceReply(Message message, Message reply) { + AbstractIntegrationMessageBuilder builder = null; + MessageHeaders replyHeaders = reply.getHeaders(); + MessageHeaders requestHeaders = message.getHeaders(); + if (replyHeaders.get(KafkaHeaders.CORRELATION_ID) == null && + requestHeaders.get(KafkaHeaders.CORRELATION_ID) != null) { + builder = getMessageBuilderFactory().fromMessage(reply) + .setHeader(KafkaHeaders.CORRELATION_ID, requestHeaders.get(KafkaHeaders.CORRELATION_ID)); + } + if (replyHeaders.get(KafkaHeaders.TOPIC) == null && + requestHeaders.get(KafkaHeaders.REPLY_TOPIC) != null) { + if (builder == null) { + builder = getMessageBuilderFactory().fromMessage(reply); + } + builder.setHeader(KafkaHeaders.TOPIC, requestHeaders.get(KafkaHeaders.REPLY_TOPIC)); + } + if (replyHeaders.get(KafkaHeaders.PARTITION_ID) == null && + requestHeaders.get(KafkaHeaders.REPLY_PARTITION) != null) { + if (builder == null) { + builder = getMessageBuilderFactory().fromMessage(reply); + } + builder.setHeader(KafkaHeaders.PARTITION_ID, requestHeaders.get(KafkaHeaders.REPLY_PARTITION)); + } + if (builder != null) { + return builder.build(); + } + return reply; + } + + @Override + public boolean open(RetryContext context, RetryCallback callback) { + if (KafkaInboundGateway.this.recoveryCallback != null) { + attributesHolder.set(context); + } + return true; + } + + @Override + public void close(RetryContext context, RetryCallback callback, + Throwable throwable) { + attributesHolder.remove(); + } + + @Override + public void onError(RetryContext context, RetryCallback callback, + Throwable throwable) { + // Empty + } + + } + +} 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 6f13b7945c..40fd7c4601 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 @@ -305,9 +305,8 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo * @since 2.1.1 */ private void setAttributesIfNecessary(Object record, Message message) { - boolean needHolder = getErrorChannel() != null - && KafkaMessageDrivenChannelAdapter.this.retryTemplate == null; - boolean needAttributes = needHolder | KafkaMessageDrivenChannelAdapter.this.retryTemplate != null; + boolean needHolder = getErrorChannel() != null && this.retryTemplate == null; + boolean needAttributes = needHolder | this.retryTemplate != null; if (needHolder) { attributesHolder.set(ErrorMessageUtils.getAttributeAccessor(null, null)); } 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 7321a7ead6..f3051f592a 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 @@ -1,5 +1,5 @@ /* - * Copyright 2013-2017 the original author or authors. + * Copyright 2013-2018 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. @@ -16,11 +16,23 @@ package org.springframework.integration.kafka.outbound; +import java.lang.reflect.Type; +import java.nio.charset.StandardCharsets; +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import java.util.TreeSet; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeader; import org.apache.kafka.common.header.internals.RecordHeaders; import org.springframework.expression.EvaluationContext; @@ -28,27 +40,38 @@ import org.springframework.expression.Expression; import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.expression.ValueExpression; -import org.springframework.integration.handler.AbstractMessageProducingHandler; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.kafka.support.KafkaSendFailureException; import org.springframework.integration.support.DefaultErrorMessageStrategy; import org.springframework.integration.support.ErrorMessageStrategy; import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.requestreply.ReplyingKafkaTemplate; +import org.springframework.kafka.requestreply.RequestReplyFuture; import org.springframework.kafka.support.DefaultKafkaHeaderMapper; import org.springframework.kafka.support.JacksonPresent; import org.springframework.kafka.support.KafkaHeaderMapper; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.KafkaNull; import org.springframework.kafka.support.SendResult; +import org.springframework.kafka.support.SimpleKafkaHeaderMapper; +import org.springframework.kafka.support.converter.KafkaMessageHeaders; +import org.springframework.kafka.support.converter.MessagingMessageConverter; +import org.springframework.kafka.support.converter.RecordMessageConverter; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.Assert; import org.springframework.util.StringUtils; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.ListenableFutureCallback; +import org.springframework.util.concurrent.SettableListenableFuture; /** - * Kafka Message Handler. + * Kafka Message Handler; when supplied with a {@link ReplyingKafkaTemplate} + * it is used as the handler in an outbound gateway. When supplied with a simple + * {@link KafkaTemplate} it used as the handler in an outbound channel adapter. * * @param the key type. * @param the value type. @@ -61,12 +84,16 @@ import org.springframework.util.concurrent.ListenableFutureCallback; * * @since 0.5 */ -public class KafkaProducerMessageHandler extends AbstractMessageProducingHandler { +public class KafkaProducerMessageHandler extends AbstractReplyProducingMessageHandler { private static final long DEFAULT_SEND_TIMEOUT = 10000; + private final Map> replyTopicsAndPartitions = new HashMap<>(); + private final KafkaTemplate kafkaTemplate; + private final boolean isGateway; + private EvaluationContext evaluationContext; private volatile Expression topicExpression; @@ -83,18 +110,37 @@ public class KafkaProducerMessageHandler extends AbstractMessageProducingH private KafkaHeaderMapper headerMapper; + private RecordMessageConverter replyMessageConverter = new MessagingMessageConverter(); + private MessageChannel sendFailureChannel; private String sendFailureChannelName; + private MessageChannel sendSuccessChannel; + + private String sendSuccessChannelName; + private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy(); + private Type replyPayloadType = Object.class; + + private volatile boolean noOutputChannel; + public KafkaProducerMessageHandler(final KafkaTemplate kafkaTemplate) { Assert.notNull(kafkaTemplate, "kafkaTemplate cannot be null"); this.kafkaTemplate = kafkaTemplate; + this.isGateway = kafkaTemplate instanceof ReplyingKafkaTemplate; + if (this.isGateway) { + setAsync(true); + updateNotPropagatedHeaders( + new String[] { KafkaHeaders.TOPIC, KafkaHeaders.PARTITION_ID, KafkaHeaders.MESSAGE_KEY }, false); + } if (JacksonPresent.isJackson2Present()) { this.headerMapper = new DefaultKafkaHeaderMapper(); } + else { + this.headerMapper = new SimpleKafkaHeaderMapper(); + } } public void setTopicExpression(Expression topicExpression) { @@ -193,6 +239,24 @@ public class KafkaProducerMessageHandler extends AbstractMessageProducingH this.sendFailureChannelName = sendFailureChannelName; } + /** + * Set the success channel. + * @param sendSuccessChannel the Success channel. + * @since 3.0.2 + */ + public void setSendSuccessChannel(MessageChannel sendSuccessChannel) { + this.sendSuccessChannel = sendSuccessChannel; + } + + /** + * Set the Success channel name. + * @param sendSuccessChannelName the Success channel name. + * @since 3.0.2 + */ + public void setSendSuccessChannelName(String sendSuccessChannelName) { + this.sendSuccessChannelName = sendSuccessChannelName; + } + /** * Set the error message strategy implementation to use when sending error messages after * send failures. Cannot be null. @@ -204,6 +268,34 @@ public class KafkaProducerMessageHandler extends AbstractMessageProducingH this.errorMessageStrategy = errorMessageStrategy; } + /** + * Set a message converter for gateway replies. + * @param messageConverter the converter. + * @since 3.0.2 + * @see #setReplyPayloadType(Type) + */ + public void setReplyMessageConverter(RecordMessageConverter messageConverter) { + Assert.notNull(messageConverter, "'messageConverter' cannot be null"); + this.replyMessageConverter = messageConverter; + } + + /** + * When using a type-aware message converter (such as {@code StringJsonMessageConverter}, + * set the payload type the converter should create. Defaults to {@link Object}. + * @param payloadType the type. + * @since 3.0.2 + * @see #setReplyMessageConverter(RecordMessageConverter) + */ + public void setReplyPayloadType(Type payloadType) { + Assert.notNull(payloadType, "'payloadType' cannot be null"); + this.replyPayloadType = payloadType; + } + + @Override + public String getComponentType() { + return this.isGateway ? "kafka:outbound-gateway" : "kafka:outbound-channel-adapter"; + } + protected MessageChannel getSendFailureChannel() { if (this.sendFailureChannel != null) { return this.sendFailureChannel; @@ -215,32 +307,43 @@ public class KafkaProducerMessageHandler extends AbstractMessageProducingH return null; } + protected MessageChannel getSendSuccessChannel() { + if (this.sendSuccessChannel != null) { + return this.sendSuccessChannel; + } + else if (this.sendSuccessChannelName != null) { + this.sendSuccessChannel = getChannelResolver().resolveDestination(this.sendSuccessChannelName); + return this.sendSuccessChannel; + } + return null; + } + @Override - protected void onInit() throws Exception { - super.onInit(); + protected void doInit() { this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); } @SuppressWarnings("unchecked") @Override - protected void handleMessageInternal(final Message message) throws Exception { + protected Object handleRequestMessage(final Message message) { + MessageHeaders messageHeaders = message.getHeaders(); String topic = this.topicExpression != null ? this.topicExpression.getValue(this.evaluationContext, message, String.class) - : message.getHeaders().get(KafkaHeaders.TOPIC, String.class); + : messageHeaders.get(KafkaHeaders.TOPIC, String.class); Assert.state(StringUtils.hasText(topic), "The 'topic' can not be empty or null"); Integer partitionId = this.partitionIdExpression != null ? this.partitionIdExpression.getValue(this.evaluationContext, message, Integer.class) - : message.getHeaders().get(KafkaHeaders.PARTITION_ID, Integer.class); + : messageHeaders.get(KafkaHeaders.PARTITION_ID, Integer.class); Object messageKey = this.messageKeyExpression != null ? this.messageKeyExpression.getValue(this.evaluationContext, message) - : message.getHeaders().get(KafkaHeaders.MESSAGE_KEY); + : messageHeaders.get(KafkaHeaders.MESSAGE_KEY); Long timestamp = this.timestampExpression != null ? this.timestampExpression.getValue(this.evaluationContext, message, Long.class) - : message.getHeaders().get(KafkaHeaders.TIMESTAMP, Long.class); + : messageHeaders.get(KafkaHeaders.TIMESTAMP, Long.class); V payload = (V) message.getPayload(); if (payload instanceof KafkaNull) { @@ -250,21 +353,120 @@ public class KafkaProducerMessageHandler extends AbstractMessageProducingH Headers headers = null; if (this.headerMapper != null) { headers = new RecordHeaders(); - this.headerMapper.fromHeaders(message.getHeaders(), headers); + this.headerMapper.fromHeaders(messageHeaders, headers); } - final ProducerRecord producerRecord = new ProducerRecord(topic, partitionId, timestamp, + final ProducerRecord producerRecord = new ProducerRecord<>(topic, partitionId, timestamp, (K) messageKey, payload, headers); - ListenableFuture> future = this.kafkaTemplate.send(producerRecord); - if (getSendFailureChannel() != null || getOutputChannel() != null) { + ListenableFuture> sendFuture; + RequestReplyFuture gatewayFuture = null; + MessageChannel metadataChannel = null; + if (this.isGateway) { + metadataChannel = getSendSuccessChannel(); + producerRecord.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, getReplyTopic(message))); + gatewayFuture = ((ReplyingKafkaTemplate) this.kafkaTemplate).sendAndReceive(producerRecord); + sendFuture = gatewayFuture.getSendFuture(); + } + else { + sendFuture = this.kafkaTemplate.send(producerRecord); + // TODO: In 3.1, always use the success channel. + if (!this.noOutputChannel) { + metadataChannel = getOutputChannel(); + if (metadataChannel == null) { + this.noOutputChannel = true; + } + } + if (metadataChannel == null) { + metadataChannel = getSendSuccessChannel(); + } + } + try { + processSendResult(message, producerRecord, sendFuture, metadataChannel); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MessageHandlingException(message, e); + } + catch (ExecutionException e) { + // TODO: in 3.1 change this to e.getCause() + throw new MessageHandlingException(message, e); + } + return processReplyFuture(gatewayFuture); + } + + private byte[] getReplyTopic(final Message message) { + if (this.replyTopicsAndPartitions.isEmpty()) { + determineValidReplyTopicsAndPartitions(); + } + Object replyHeader = message.getHeaders().get(KafkaHeaders.REPLY_TOPIC); + byte[] replyTopic = null; + String topicToCheck = null; + if (replyHeader instanceof String) { + replyTopic = ((String) replyHeader).getBytes(StandardCharsets.UTF_8); + topicToCheck = (String) replyHeader; + } + else if (replyHeader instanceof byte[]) { + replyTopic = (byte[]) replyHeader; + } + else if (replyHeader != null) { + throw new IllegalStateException(KafkaHeaders.REPLY_TOPIC + " must be String or byte[]"); + } + if (replyTopic == null) { + if (this.replyTopicsAndPartitions.size() == 1) { + replyTopic = this.replyTopicsAndPartitions.keySet().iterator().next().getBytes(StandardCharsets.UTF_8); + } + else { + throw new IllegalStateException("No reply topic header and no default reply topic is can be determined"); + } + } + else { + if (topicToCheck == null) { + topicToCheck = new String(replyTopic, StandardCharsets.UTF_8); + } + if (!this.replyTopicsAndPartitions.keySet().contains(topicToCheck)) { + throw new IllegalStateException("The reply topic header [" + + topicToCheck + + "] does not match any reply container topic: " + this.replyTopicsAndPartitions.keySet()); + } + } + Integer replyPartition = message.getHeaders().get(KafkaHeaders.REPLY_PARTITION, Integer.class); + if (replyPartition != null) { + if (topicToCheck == null) { + topicToCheck = new String(replyTopic, StandardCharsets.UTF_8); + } + if (!this.replyTopicsAndPartitions.get(topicToCheck).contains(replyPartition)) { + throw new IllegalStateException("The reply partition header [" + + replyPartition + "] does not match any reply container partition for topic [" + + topicToCheck + "]: " + this.replyTopicsAndPartitions.get(topicToCheck)); + } + } + return replyTopic; + } + + private void determineValidReplyTopicsAndPartitions() { + ReplyingKafkaTemplate rkt = (ReplyingKafkaTemplate) kafkaTemplate; + Collection replyTopics = rkt.getAssignedReplyTopicPartitions(); + Map> topicsAndPartitions = new HashMap<>(); + if (replyTopics != null) { + replyTopics.forEach(tp -> { + topicsAndPartitions.computeIfAbsent(tp.topic(), (k) -> new TreeSet<>()); + topicsAndPartitions.get(tp.topic()).add(tp.partition()); + }); + this.replyTopicsAndPartitions.putAll(topicsAndPartitions); + } + } + + public void processSendResult(final Message message, final ProducerRecord producerRecord, + ListenableFuture> future, MessageChannel metadataChannel) + throws InterruptedException, ExecutionException { + if (getSendFailureChannel() != null || metadataChannel != null) { future.addCallback(new ListenableFutureCallback>() { @Override public void onSuccess(SendResult result) { - if (getOutputChannel() != null) { - KafkaProducerMessageHandler.this.messagingTemplate.send(getOutputChannel(), + if (metadataChannel != null) { + KafkaProducerMessageHandler.this.messagingTemplate.send(metadataChannel, getMessageBuilderFactory().fromMessage(message) - // TODO: change to constant when available - .setHeader("kafka_recordMetadata", result.getRecordMetadata()).build()); + .setHeader(KafkaHeaders.RECORD_METADATA, result.getRecordMetadata()).build()); } } @@ -296,9 +498,58 @@ public class KafkaProducerMessageHandler extends AbstractMessageProducingH } } - @Override - public String getComponentType() { - return "kafka:outbound-channel-adapter"; + private Future processReplyFuture(RequestReplyFuture future) { + if (future == null) { + return null; + } + return new ConvertingReplyFuture(future); + } + + private final class ConvertingReplyFuture extends SettableListenableFuture { + + ConvertingReplyFuture(RequestReplyFuture future) { + addCallback(future); + } + + private void addCallback(final RequestReplyFuture future) { + future.addCallback(new ListenableFutureCallback>() { + + @Override + public void onSuccess(ConsumerRecord result) { + try { + set(dontLeakHeaders(KafkaProducerMessageHandler.this.replyMessageConverter.toMessage(result, + null, null, KafkaProducerMessageHandler.this.replyPayloadType))); + } + catch (Exception e) { + setException(e); + } + } + + private Message dontLeakHeaders(Message message) { + if (message.getHeaders() instanceof KafkaMessageHeaders) { + Map headers = ((KafkaMessageHeaders) message.getHeaders()).getRawHeaders(); + headers.remove(KafkaHeaders.CORRELATION_ID); + headers.remove(KafkaHeaders.REPLY_TOPIC); + headers.remove(KafkaHeaders.REPLY_PARTITION); + return message; + } + else { + return getMessageBuilderFactory().fromMessage(message) + .removeHeader(KafkaHeaders.CORRELATION_ID) + .removeHeader(KafkaHeaders.REPLY_TOPIC) + .removeHeader(KafkaHeaders.REPLY_PARTITION) + .build(); + } + } + + @Override + public void onFailure(Throwable ex) { + setException(ex); + } + + }); + } + } } 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 632e3772ae..6509849a13 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 @@ -19,6 +19,7 @@ package org.springframework.integration.kafka.dsl; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import java.util.Collection; import java.util.Collections; import java.util.Map; import java.util.concurrent.CountDownLatch; @@ -26,6 +27,8 @@ import java.util.concurrent.TimeUnit; import java.util.stream.Stream; import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.common.TopicPartition; import org.junit.ClassRule; import org.junit.Test; import org.junit.runner.RunWith; @@ -48,13 +51,17 @@ import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandle import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; +import org.springframework.kafka.annotation.EnableKafka; 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.AbstractMessageListenerContainer; +import org.springframework.kafka.listener.GenericMessageListenerContainer; +import org.springframework.kafka.listener.KafkaMessageListenerContainer; import org.springframework.kafka.listener.MessageListenerContainer; +import org.springframework.kafka.listener.config.ContainerProperties; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.DefaultKafkaHeaderMapper; import org.springframework.kafka.support.KafkaHeaders; @@ -64,6 +71,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; import org.springframework.retry.support.RetryTemplate; @@ -88,8 +96,13 @@ public class KafkaDslTests { private static final String TEST_TOPIC3 = "test-topic3"; + private static final String TEST_TOPIC4 = "test-topic4"; + + private static final String TEST_TOPIC5 = "test-topic5"; + @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, TEST_TOPIC1, TEST_TOPIC2, TEST_TOPIC3); + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, TEST_TOPIC1, TEST_TOPIC2, TEST_TOPIC3, + TEST_TOPIC4, TEST_TOPIC5); @Autowired @Qualifier("sendToKafkaFlow.input") @@ -130,6 +143,9 @@ public class KafkaDslTests { @Autowired private ContextConfiguration config; + @Autowired + private Gate gate; + @Test public void testKafkaAdapters() throws Exception { @@ -191,15 +207,24 @@ public class KafkaDslTests { assertThat(this.kafkaTemplateTopic2).isNotNull(); this.kafkaTemplateTopic1.send(TEST_TOPIC3, "foo"); - assertThat(this.config.latch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.sourceFlowLatch.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(this.config.fromSource).isEqualTo("foo"); } + @Test + public void testGateways() throws Exception { + assertThat(this.config.replyContainerLatch.await(30, TimeUnit.SECONDS)); + assertThat(this.gate.exchange(TEST_TOPIC4, "foo")).isEqualTo("FOO"); + } + @Configuration @EnableIntegration + @EnableKafka public static class ContextConfiguration { - private final CountDownLatch latch = new CountDownLatch(1); + private final CountDownLatch sourceFlowLatch = new CountDownLatch(1); + + private final CountDownLatch replyContainerLatch = new CountDownLatch(1); private Object fromSource; @@ -301,12 +326,59 @@ public class KafkaDslTests { e -> e.poller(Pollers.fixedDelay(100))) .handle(p -> { this.fromSource = p.getPayload(); - this.latch.countDown(); + this.sourceFlowLatch.countDown(); }) .get(); } + @Bean + public IntegrationFlow outboundGateFlow() { + return IntegrationFlows.from(Gate.class) + .handle(Kafka.outboundGateway(producerFactory(), replyContainer()) + .configureKafkaTemplate(t -> t.replyTimeout(30_000))) + .get(); + } + + private GenericMessageListenerContainer replyContainer() { + ContainerProperties containerProperties = new ContainerProperties(TEST_TOPIC5); + containerProperties.setGroupId("outGate"); + containerProperties.setConsumerRebalanceListener(new ConsumerRebalanceListener() { + + @Override + public void onPartitionsRevoked(Collection partitions) { + // empty + } + + @Override + public void onPartitionsAssigned(Collection partitions) { + ContextConfiguration.this.replyContainerLatch.countDown(); + } + + }); + return new KafkaMessageListenerContainer<>(consumerFactory(), containerProperties); + } + + @Bean + public IntegrationFlow serverGateway() { + return IntegrationFlows + .from(Kafka.inboundGateway(consumerFactory(), containerProperties(), + producerFactory())) + .transform(String::toUpperCase) + .get(); + } + + private ContainerProperties containerProperties() { + ContainerProperties containerProperties = new ContainerProperties(TEST_TOPIC4); + containerProperties.setGroupId("inGateGroup"); + return containerProperties; + } + } + public interface Gate { + + String exchange(@Header(KafkaHeaders.TOPIC) String topic, String out); + + } } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java new file mode 100644 index 0000000000..c26368e4a0 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java @@ -0,0 +1,306 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.inbound; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.springframework.kafka.test.assertj.KafkaConditions.partition; +import static org.springframework.kafka.test.assertj.KafkaConditions.value; + +import java.lang.reflect.Type; +import java.util.Map; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.junit.ClassRule; +import org.junit.Test; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer; +import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy; +import org.springframework.integration.support.MessageBuilder; +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.KafkaMessageListenerContainer; +import org.springframework.kafka.listener.config.ContainerProperties; +import org.springframework.kafka.support.Acknowledgment; +import org.springframework.kafka.support.KafkaHeaders; +import org.springframework.kafka.support.converter.MessagingMessageConverter; +import org.springframework.kafka.test.rule.KafkaEmbedded; +import org.springframework.kafka.test.utils.ContainerTestUtils; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.support.ErrorMessage; +import org.springframework.retry.backoff.NoBackOffPolicy; +import org.springframework.retry.policy.SimpleRetryPolicy; +import org.springframework.retry.support.RetryTemplate; + +/** + * + * @author Gary Russell + * + * @since 3.0.2 + * + */ +public class InboundGatewayTests { + + private static String topic1 = "testTopic1"; + + private static String topic2 = "testTopic2"; + + private static String topic3 = "testTopic3"; + + private static String topic4 = "testTopic4"; + + private static String topic5 = "testTopic5"; + + private static String topic6 = "testTopic6"; + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5, + topic6); + + @Test + public void testInbound() throws Exception { + Map props = KafkaTestUtils.consumerProps("test1", "false", embeddedKafka); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + ContainerProperties containerProps = new ContainerProperties(topic1); + KafkaMessageListenerContainer container = + new KafkaMessageListenerContainer<>(cf, containerProps); + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf); + template.setDefaultTopic(topic1); + KafkaInboundGateway gateway = new KafkaInboundGateway<>(container, template); + QueueChannel out = new QueueChannel(); + DirectChannel reply = new DirectChannel(); + gateway.setRequestChannel(out); + gateway.setReplyChannel(reply); + gateway.setBeanFactory(mock(BeanFactory.class)); + gateway.setReplyTimeout(30_000); + gateway.afterPropertiesSet(); + gateway.setMessageConverter(new MessagingMessageConverter() { + + @Override + public Message toMessage(ConsumerRecord record, Acknowledgment acknowledgment, + Consumer consumer, Type type) { + Message message = super.toMessage(record, acknowledgment, consumer, type); + return MessageBuilder.fromMessage(message) + .setHeader("testHeader", "testValue") + .setHeader(KafkaHeaders.REPLY_TOPIC, topic2) + .setHeader(KafkaHeaders.REPLY_PARTITION, 1) + .build(); + } + + }); + gateway.start(); + ContainerTestUtils.waitForAssignment(container, 2); + + template.sendDefault(0, 1487694048607L, 1, "foo"); + Message received = out.receive(30_000); + assertThat(received).isNotNull(); + + MessageHeaders headers = received.getHeaders(); + assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo(1); + assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(topic1); + assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0); + assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo(0L); + assertThat(headers.get(KafkaHeaders.RECEIVED_TIMESTAMP)).isEqualTo(1487694048607L); + assertThat(headers.get(KafkaHeaders.TIMESTAMP_TYPE)).isEqualTo("CREATE_TIME"); + assertThat(headers.get(KafkaHeaders.REPLY_TOPIC)).isEqualTo(topic2); + assertThat(headers.get("testHeader")).isEqualTo("testValue"); + reply.send(MessageBuilder.withPayload("FOO").copyHeaders(headers).build()); + + Map consumerProps = KafkaTestUtils.consumerProps("replyHandler1", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + ConsumerFactory cf2 = new DefaultKafkaConsumerFactory<>(consumerProps); + Consumer consumer = cf2.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic2); + ConsumerRecord record = KafkaTestUtils.getSingleRecord(consumer, topic2); + assertThat(record).has(partition(1)); + assertThat(record).has(value("FOO")); + + gateway.stop(); + } + + @Test + public void testInboundErrorRecover() throws Exception { + Map props = KafkaTestUtils.consumerProps("test2", "false", embeddedKafka); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + ContainerProperties containerProps = new ContainerProperties(topic3); + KafkaMessageListenerContainer container = + new KafkaMessageListenerContainer<>(cf, containerProps); + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf); + template.setDefaultTopic(topic3); + KafkaInboundGateway gateway = new KafkaInboundGateway<>(container, template); + MessageChannel out = new DirectChannel() { + + @Override + protected boolean doSend(Message message, long timeout) { + throw new RuntimeException("intended"); + } + + }; + QueueChannel errors = new QueueChannel(); + gateway.setRequestChannel(out); + gateway.setErrorChannel(errors); + gateway.setBeanFactory(mock(BeanFactory.class)); + gateway.setMessageConverter(new MessagingMessageConverter() { + + @Override + public Message toMessage(ConsumerRecord record, Acknowledgment acknowledgment, + Consumer consumer, Type type) { + Message message = super.toMessage(record, acknowledgment, consumer, type); + return MessageBuilder.fromMessage(message) + .setHeader("testHeader", "testValue") + .setHeader(KafkaHeaders.REPLY_TOPIC, topic4) + .setHeader(KafkaHeaders.REPLY_PARTITION, 1) + .build(); + } + + }); + gateway.setReplyTimeout(30_000); + gateway.afterPropertiesSet(); + gateway.start(); + ContainerTestUtils.waitForAssignment(container, 2); + + template.sendDefault(0, 1487694048607L, 1, "foo"); + ErrorMessage em = (ErrorMessage) errors.receive(30_000); + assertThat(em).isNotNull(); + Message failed = ((MessagingException) em.getPayload()).getFailedMessage(); + assertThat(failed).isNotNull(); + MessageChannel reply = (MessageChannel) em.getHeaders().getReplyChannel(); + MessageHeaders headers = failed.getHeaders(); + reply.send(MessageBuilder.withPayload("ERROR").copyHeaders(headers).build()); + + assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo(1); + assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(topic3); + assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0); + assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo(0L); + assertThat(headers.get(KafkaHeaders.RECEIVED_TIMESTAMP)).isEqualTo(1487694048607L); + assertThat(headers.get(KafkaHeaders.TIMESTAMP_TYPE)).isEqualTo("CREATE_TIME"); + assertThat(headers.get(KafkaHeaders.REPLY_TOPIC)).isEqualTo(topic4); + assertThat(headers.get("testHeader")).isEqualTo("testValue"); + + Map consumerProps = KafkaTestUtils.consumerProps("replyHandler2", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + ConsumerFactory cf2 = new DefaultKafkaConsumerFactory<>(consumerProps); + Consumer consumer = cf2.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic4); + ConsumerRecord record = KafkaTestUtils.getSingleRecord(consumer, topic4); + assertThat(record).has(partition(1)); + assertThat(record).has(value("ERROR")); + + gateway.stop(); + } + + @Test + public void testInboundRetryErrorRecover() throws Exception { + Map props = KafkaTestUtils.consumerProps("test3", "false", embeddedKafka); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + ContainerProperties containerProps = new ContainerProperties(topic5); + KafkaMessageListenerContainer container = + new KafkaMessageListenerContainer<>(cf, containerProps); + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf); + template.setDefaultTopic(topic5); + KafkaInboundGateway gateway = new KafkaInboundGateway<>(container, template); + MessageChannel out = new DirectChannel() { + + @Override + protected boolean doSend(Message message, long timeout) { + throw new RuntimeException("intended"); + } + + }; + QueueChannel errors = new QueueChannel(); + gateway.setRequestChannel(out); + gateway.setErrorChannel(errors); + gateway.setBeanFactory(mock(BeanFactory.class)); + gateway.setMessageConverter(new MessagingMessageConverter() { + + @Override + public Message toMessage(ConsumerRecord record, Acknowledgment acknowledgment, + Consumer consumer, Type type) { + Message message = super.toMessage(record, acknowledgment, consumer, type); + return MessageBuilder.fromMessage(message) + .setHeader("testHeader", "testValue") + .setHeader(KafkaHeaders.REPLY_TOPIC, topic6) + .setHeader(KafkaHeaders.REPLY_PARTITION, 1) + .build(); + } + + }); + gateway.setReplyTimeout(30_000); + RetryTemplate retryTemplate = new RetryTemplate(); + SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy(); + retryPolicy.setMaxAttempts(2); + retryTemplate.setRetryPolicy(retryPolicy); + retryTemplate.setBackOffPolicy(new NoBackOffPolicy()); + gateway.setRetryTemplate(retryTemplate); + gateway.setRecoveryCallback( + new ErrorMessageSendingRecoverer(errors, new RawRecordHeaderErrorMessageStrategy())); + gateway.afterPropertiesSet(); + gateway.start(); + ContainerTestUtils.waitForAssignment(container, 2); + + template.sendDefault(0, 1487694048607L, 1, "foo"); + ErrorMessage em = (ErrorMessage) errors.receive(30_000); + assertThat(em).isNotNull(); + Message failed = ((MessagingException) em.getPayload()).getFailedMessage(); + assertThat(failed).isNotNull(); + MessageChannel reply = (MessageChannel) em.getHeaders().getReplyChannel(); + MessageHeaders headers = failed.getHeaders(); + reply.send(MessageBuilder.withPayload("ERROR").copyHeaders(headers).build()); + + assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo(1); + assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(topic5); + assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0); + assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo(0L); + assertThat(headers.get(KafkaHeaders.RECEIVED_TIMESTAMP)).isEqualTo(1487694048607L); + assertThat(headers.get(KafkaHeaders.TIMESTAMP_TYPE)).isEqualTo("CREATE_TIME"); + assertThat(headers.get(KafkaHeaders.REPLY_TOPIC)).isEqualTo(topic6); + assertThat(headers.get("testHeader")).isEqualTo("testValue"); + + Map consumerProps = KafkaTestUtils.consumerProps("replyHandler3", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + ConsumerFactory cf2 = new DefaultKafkaConsumerFactory<>(consumerProps); + Consumer consumer = cf2.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic6); + ConsumerRecord record = KafkaTestUtils.getSingleRecord(consumer, topic6); + assertThat(record).has(partition(1)); + assertThat(record).has(value("ERROR")); + + gateway.stop(); + } + +} diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java index 39a87c31ad..4a67a5ecbd 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-2018 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. @@ -17,19 +17,26 @@ package org.springframework.integration.kafka.outbound; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; import static org.mockito.Mockito.mock; import static org.springframework.kafka.test.assertj.KafkaConditions.key; import static org.springframework.kafka.test.assertj.KafkaConditions.partition; import static org.springframework.kafka.test.assertj.KafkaConditions.timestamp; import static org.springframework.kafka.test.assertj.KafkaConditions.value; +import java.util.Collection; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.TopicPartition; +import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.ClassRule; import org.junit.Test; @@ -46,6 +53,9 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.listener.KafkaMessageListenerContainer; +import org.springframework.kafka.listener.config.ContainerProperties; +import org.springframework.kafka.requestreply.ReplyingKafkaTemplate; import org.springframework.kafka.support.DefaultKafkaHeaderMapper; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.KafkaNull; @@ -53,6 +63,7 @@ import org.springframework.kafka.support.SendResult; import org.springframework.kafka.test.rule.KafkaEmbedded; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.ErrorMessage; @@ -71,9 +82,17 @@ public class KafkaProducerMessageHandlerTests { private static String topic2 = "testTopic2out"; - @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2); + private static String topic3 = "testTopic3out"; + private static String topic4 = "testTopic4out"; + + private static String topic5 = "testTopic5out"; + + private static String topic6 = "testTopic6in"; + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5, + topic6); private static Consumer consumer; @@ -85,6 +104,11 @@ public class KafkaProducerMessageHandlerTests { embeddedKafka.consumeFromAllEmbeddedTopics(consumer); } + @AfterClass + public static void tearDown() { + consumer.close(); + } + @Test public void testOutbound() { ProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( @@ -149,7 +173,7 @@ public class KafkaProducerMessageHandlerTests { handler.afterPropertiesSet(); Message message = MessageBuilder.withPayload("foo") - .setHeader(KafkaHeaders.TOPIC, topic1) + .setHeader(KafkaHeaders.TOPIC, topic2) .setHeader(KafkaHeaders.MESSAGE_KEY, 2) .setHeader(KafkaHeaders.PARTITION_ID, 1) .setHeader(KafkaHeaders.TIMESTAMP, 1487694048607L) @@ -157,7 +181,7 @@ public class KafkaProducerMessageHandlerTests { .build(); handler.handleMessage(message); - ConsumerRecord record = KafkaTestUtils.getSingleRecord(consumer, topic1); + ConsumerRecord record = KafkaTestUtils.getSingleRecord(consumer, topic2); assertThat(record).has(key(2)); assertThat(record).has(partition(1)); assertThat(record).has(value("foo")); @@ -178,7 +202,7 @@ public class KafkaProducerMessageHandlerTests { handler.afterPropertiesSet(); Message message = MessageBuilder.withPayload("foo") - .setHeader(KafkaHeaders.TOPIC, topic1) + .setHeader(KafkaHeaders.TOPIC, topic3) .setHeader(KafkaHeaders.MESSAGE_KEY, 2) .setHeader(KafkaHeaders.PARTITION_ID, 1) .build(); @@ -187,7 +211,7 @@ public class KafkaProducerMessageHandlerTests { handler.handleMessage(message); - ConsumerRecord record1 = KafkaTestUtils.getSingleRecord(consumer, topic1); + ConsumerRecord record1 = KafkaTestUtils.getSingleRecord(consumer, topic3); assertThat(record1).has(key(2)); assertThat(record1).has(partition(1)); assertThat(record1).has(value("foo")); @@ -198,7 +222,7 @@ public class KafkaProducerMessageHandlerTests { handler.handleMessage(message); - ConsumerRecord record2 = KafkaTestUtils.getSingleRecord(consumer, topic1); + ConsumerRecord record2 = KafkaTestUtils.getSingleRecord(consumer, topic3); assertThat(record2).has(key(2)); assertThat(record2).has(partition(1)); assertThat(record2).has(value("foo")); @@ -217,21 +241,20 @@ public class KafkaProducerMessageHandlerTests { handler.afterPropertiesSet(); Message message = MessageBuilder.withPayload("foo") - .setHeader(KafkaHeaders.TOPIC, topic1) + .setHeader(KafkaHeaders.TOPIC, topic4) .setHeader(KafkaHeaders.MESSAGE_KEY, 2) .setHeader(KafkaHeaders.PARTITION_ID, 1) .build(); handler.handleMessage(message); - ConsumerRecord record = KafkaTestUtils.getSingleRecord(consumer, topic1); + ConsumerRecord record = KafkaTestUtils.getSingleRecord(consumer, topic4); assertThat(record).has(key(2)); assertThat(record).has(partition(1)); assertThat(record).has(value("foo")); Message received = successes.receive(10000); assertThat(received).isNotNull(); assertThat(received.getPayload()).isEqualTo("foo"); - // TODO: Change to constant when available - assertThat(received.getHeaders().get("kafka_recordMetadata")).isInstanceOf(RecordMetadata.class); + assertThat(received.getHeaders().get(KafkaHeaders.RECORD_METADATA)).isInstanceOf(RecordMetadata.class); final RuntimeException fooException = new RuntimeException("foo"); @@ -264,4 +287,94 @@ public class KafkaProducerMessageHandlerTests { assertThat(((KafkaSendFailureException) received.getPayload()).getRecord()).isNotNull(); } + @Test + public void testOutboundGateway() throws Exception { + ConsumerFactory consumerFactory = new DefaultKafkaConsumerFactory<>( + KafkaTestUtils.consumerProps(topic5, "false", embeddedKafka)); + ContainerProperties containerProperties = new ContainerProperties(topic6); + final CountDownLatch assigned = new CountDownLatch(1); + containerProperties.setConsumerRebalanceListener(new ConsumerRebalanceListener() { + + @Override + public void onPartitionsRevoked(Collection partitions) { + // empty + } + + @Override + public void onPartitionsAssigned(Collection partitions) { + assigned.countDown(); + } + + }); + KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(consumerFactory, containerProperties); + ProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( + KafkaTestUtils.producerProps(embeddedKafka)); + ReplyingKafkaTemplate template = new ReplyingKafkaTemplate<>(producerFactory, container); + template.start(); + assertThat(assigned.await(30, TimeUnit.SECONDS)).isTrue(); + KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler<>(template); + handler.setBeanFactory(mock(BeanFactory.class)); + QueueChannel replies = new QueueChannel(); + handler.setOutputChannel(replies); + handler.afterPropertiesSet(); + + Message message = MessageBuilder.withPayload("foo") + .setHeader(KafkaHeaders.TOPIC, topic5) + .setHeader(KafkaHeaders.MESSAGE_KEY, 2) + .setHeader(KafkaHeaders.PARTITION_ID, 1) + .build(); + handler.handleMessage(message); + + ConsumerRecord record = KafkaTestUtils.getSingleRecord(consumer, topic5); + assertThat(record).has(key(2)); + assertThat(record).has(partition(1)); + assertThat(record).has(value("foo")); + Map headers = new HashMap<>(); + new DefaultKafkaHeaderMapper().toHeaders(record.headers(), headers); + assertThat(headers.get(KafkaHeaders.REPLY_TOPIC)).isEqualTo(topic6.getBytes()); + ProducerRecord pr = new ProducerRecord<>(topic6, 0, 1, "FOO", record.headers()); + template.send(pr); + Message reply = replies.receive(30_000); + assertThat(reply).isNotNull(); + assertThat(reply.getPayload()).isEqualTo("FOO"); + assertThat(reply.getHeaders().get(KafkaHeaders.TOPIC)).isNull(); + assertThat(reply.getHeaders().get(KafkaHeaders.CORRELATION_ID)).isNull(); + + message = MessageBuilder.withPayload("foo") + .setHeader(KafkaHeaders.TOPIC, topic5) + .setHeader(KafkaHeaders.MESSAGE_KEY, 2) + .setHeader(KafkaHeaders.PARTITION_ID, 1) + .setHeader(KafkaHeaders.REPLY_TOPIC, "bad") + .build(); + try { + handler.handleMessage(message); + fail("Expected exception"); + } + catch (MessageHandlingException e) { + assertThat(e.getCause().getMessage()) + .isEqualTo("The reply topic header [bad] does not match any reply container topic: " + + "[" + topic6 + "]"); + } + message = MessageBuilder.withPayload("foo") + .setHeader(KafkaHeaders.TOPIC, topic5) + .setHeader(KafkaHeaders.MESSAGE_KEY, 2) + .setHeader(KafkaHeaders.PARTITION_ID, 1) + .setHeader(KafkaHeaders.REPLY_PARTITION, 999) + .build(); + try { + handler.handleMessage(message); + fail("Expected exception"); + } + catch (MessageHandlingException e) { + assertThat(e.getCause().getMessage()) + .isEqualTo( + "The reply partition header [999] does not match any reply container partition for topic [" + + topic6 + "]: [0, 1]"); + } + + template.stop(); + // discard from the test consumer + KafkaTestUtils.getSingleRecord(consumer, topic6); + } + }