From 598eb7f80eae57c039c6d99e5d421f3f3d357d05 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 11 Nov 2016 21:13:02 -0500 Subject: [PATCH] Upgrade to SI-5.0 & Port `Kafka` Java DSL * Upgrade all the dependencies * Upgrade to Gradle 3.1 (experimental) * Make Java 8 as minimal * Decrease test logging level to `warn` * Document Java DSL support and some other polishing in the `README.adoc` --- .../integration/kafka/dsl/Kafka.java | 251 ++++++++++ .../KafkaMessageDrivenChannelAdapterSpec.java | 450 ++++++++++++++++++ .../dsl/KafkaProducerMessageHandlerSpec.java | 251 ++++++++++ .../integration/kafka/dsl/package-info.java | 4 + .../integration/kafka/dsl/KafkaTests.java | 201 ++++++++ .../src/test/resources/log4j2-test.xml | 4 +- 6 files changed, 1159 insertions(+), 2 deletions(-) create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/package-info.java create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaTests.java 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 new file mode 100644 index 0000000000..fc2999beee --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java @@ -0,0 +1,251 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.dsl; + +import java.util.regex.Pattern; + +import org.apache.kafka.common.TopicPartition; + +import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; +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.config.ContainerProperties; +import org.springframework.kafka.support.TopicPartitionInitialOffset; + +/** + * Factory class for Apache Kafka components. + * + * @author Artem Bilan + * @author Nasko Vasilev + * + * @since 3.0 + */ +public final class Kafka { + + /** + * Create an initial {@link KafkaProducerMessageHandlerSpec}. + * @param kafkaTemplate the {@link KafkaTemplate} to use + * @param the Kafka message key type. + * @param the Kafka message value type. + * @return the Kafka09ProducerMessageHandlerSpec. + */ + public static KafkaProducerMessageHandlerSpec + outboundChannelAdapter(KafkaTemplate kafkaTemplate) { + return new KafkaProducerMessageHandlerSpec<>(kafkaTemplate); + } + + /** + * Create an initial {@link KafkaProducerMessageHandlerSpec} with ProducerFactory. + * @param producerFactory the {@link ProducerFactory} Java 8 Lambda. + * @param the Kafka message key type. + * @param the Kafka message value type. + * @return the KafkaProducerMessageHandlerSpec. + * @see Kafka Producer Configs + */ + public static KafkaProducerMessageHandlerSpec.KafkaProducerMessageHandlerTemplateSpec + outboundChannelAdapter(ProducerFactory producerFactory) { + return new KafkaProducerMessageHandlerSpec.KafkaProducerMessageHandlerTemplateSpec<>(producerFactory); + } + + /** + * Create an initial {@link KafkaMessageDrivenChannelAdapterSpec}. + * @param listenerContainer the {@link AbstractMessageListenerContainer}. + * @param the Kafka message key type. + * @param the Kafka message value type. + * @param the {@link KafkaMessageDrivenChannelAdapterSpec} extension type. + * @return the Kafka09MessageDrivenChannelAdapterSpec. + */ + public static > + KafkaMessageDrivenChannelAdapterSpec messageDrivenChannelAdapter( + AbstractMessageListenerContainer listenerContainer) { + return messageDrivenChannelAdapter(listenerContainer, KafkaMessageDrivenChannelAdapter.ListenerMode.record); + } + + /** + * Create an initial {@link KafkaMessageDrivenChannelAdapterSpec}. + * @param listenerContainer the {@link AbstractMessageListenerContainer}. + * @param listenerMode the {@link KafkaMessageDrivenChannelAdapter.ListenerMode}. + * @param the Kafka message key type. + * @param the Kafka message value type. + * @param the {@link KafkaMessageDrivenChannelAdapterSpec} extension type. + * @return the Kafka09MessageDrivenChannelAdapterSpec. + */ + public static > + KafkaMessageDrivenChannelAdapterSpec messageDrivenChannelAdapter( + AbstractMessageListenerContainer listenerContainer, + KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) { + return new KafkaMessageDrivenChannelAdapterSpec<>(listenerContainer, listenerMode); + } + + /** + * Create an initial + * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}. + * @param consumerFactory the {@link ConsumerFactory}. + * @param containerProperties the {@link ContainerProperties} to use. + * @param the Kafka message key type. + * @param the Kafka message value type. + * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec. + */ + public static + KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec + messageDrivenChannelAdapter(ConsumerFactory consumerFactory, ContainerProperties containerProperties) { + return messageDrivenChannelAdapter(consumerFactory, containerProperties, + KafkaMessageDrivenChannelAdapter.ListenerMode.record); + } + + /** + * Create an initial + * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}. + * @param consumerFactory the {@link ConsumerFactory}. + * @param containerProperties the {@link ContainerProperties} to use. + * @param listenerMode the {@link KafkaMessageDrivenChannelAdapter.ListenerMode}. + * @param the Kafka message key type. + * @param the Kafka message value type. + * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec. + */ + public static + KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec + messageDrivenChannelAdapter(ConsumerFactory consumerFactory, ContainerProperties containerProperties, + KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) { + return messageDrivenChannelAdapter( + new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory, + containerProperties), listenerMode); + } + + /** + * Create an initial + * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}. + * @param consumerFactory the {@link ConsumerFactory}. + * @param topicPartitions the {@link TopicPartition} vararg. + * @param the Kafka message key type. + * @param the Kafka message value type. + * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec. + */ + public static + KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec + messageDrivenChannelAdapter(ConsumerFactory consumerFactory, + TopicPartitionInitialOffset... topicPartitions) { + return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record, + topicPartitions); + } + + /** + * Create an initial + * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}. + * @param consumerFactory the {@link ConsumerFactory}. + * @param listenerMode the {@link KafkaMessageDrivenChannelAdapter.ListenerMode}. + * @param topicPartitions the {@link TopicPartition} vararg. + * @param the Kafka message key type. + * @param the Kafka message value type. + * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec. + */ + public static + KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec + messageDrivenChannelAdapter(ConsumerFactory consumerFactory, + KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode, + TopicPartitionInitialOffset... topicPartitions) { + return messageDrivenChannelAdapter( + new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory, + topicPartitions), listenerMode); + } + + /** + * Create an initial + * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}. + * @param consumerFactory the {@link ConsumerFactory}. + * @param topics the topics vararg. + * @param the Kafka message key type. + * @param the Kafka message value type. + * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec. + */ + public static + KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec + messageDrivenChannelAdapter(ConsumerFactory consumerFactory, String... topics) { + return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record, + topics); + } + + /** + * Create an initial + * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}. + * @param consumerFactory the {@link ConsumerFactory}. + * @param listenerMode the {@link KafkaMessageDrivenChannelAdapter.ListenerMode}. + * @param topics the topics vararg. + * @param the Kafka message key type. + * @param the Kafka message value type. + * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec. + */ + public static + KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec + messageDrivenChannelAdapter(ConsumerFactory consumerFactory, + KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode, String... topics) { + return messageDrivenChannelAdapter( + new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory, + topics), listenerMode); + } + + /** + * Create an initial + * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}. + * @param consumerFactory the {@link ConsumerFactory}. + * @param topicPattern the topicPattern vararg. + * @param the Kafka message key type. + * @param the Kafka message value type. + * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec. + */ + public static + KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec + messageDrivenChannelAdapter(ConsumerFactory consumerFactory, Pattern topicPattern) { + return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record, + topicPattern); + } + + /** + * Create an initial + * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}. + * @param consumerFactory the {@link ConsumerFactory}. + * @param listenerMode the {@link KafkaMessageDrivenChannelAdapter.ListenerMode}. + * @param topicPattern the topicPattern vararg. + * @param the Kafka message key type. + * @param the Kafka message value type. + * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec. + */ + public static + KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec + messageDrivenChannelAdapter(ConsumerFactory consumerFactory, + KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode, Pattern topicPattern) { + return messageDrivenChannelAdapter( + new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory, + topicPattern), + listenerMode); + } + + private static + KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec + messageDrivenChannelAdapter(KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec spec, + KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) { + return new KafkaMessageDrivenChannelAdapterSpec + .KafkaMessageDrivenChannelAdapterListenerContainerSpec<>(spec, listenerMode); + } + + private Kafka() { + super(); + } + +} 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 new file mode 100644 index 0000000000..92950845b5 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java @@ -0,0 +1,450 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.dsl; + +import java.util.Collection; +import java.util.Collections; +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.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.FilteringAcknowledgingMessageListenerAdapter; +import org.springframework.kafka.listener.adapter.RecordFilterStrategy; +import org.springframework.kafka.listener.adapter.RetryingAcknowledgingMessageListenerAdapter; +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; +import org.springframework.retry.RecoveryCallback; +import org.springframework.retry.support.RetryTemplate; +import org.springframework.util.Assert; + +/** + * A {@link MessageProducerSpec} implementation for the {@link KafkaMessageDrivenChannelAdapter}. + * + * @param the key type. + * @param the value type. + * @param the target {@link KafkaMessageDrivenChannelAdapterSpec} implementation type. + * + * @author Artem Bilan + * + * @since 3.0 + */ +public class KafkaMessageDrivenChannelAdapterSpec> + extends MessageProducerSpec> { + + KafkaMessageDrivenChannelAdapterSpec(AbstractMessageListenerContainer messageListenerContainer, + KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) { + super(new KafkaMessageDrivenChannelAdapter<>(messageListenerContainer, listenerMode)); + } + + /** + * Set the message converter; must be a {@link RecordMessageConverter} or + * {@link BatchMessageConverter} depending on mode. + * @param messageConverter the converter. + * @return the spec + */ + public S messageConverter(MessageConverter messageConverter) { + this.target.setMessageConverter(messageConverter); + return _this(); + } + + /** + * Set the message converter to use with a record-based consumer. + * @param messageConverter the converter. + * @return the spec + */ + public S recordMessageConverter(RecordMessageConverter messageConverter) { + this.target.setRecordMessageConverter(messageConverter); + return _this(); + } + + /** + * Set the message converter to use with a batch-based consumer. + * @param messageConverter the converter. + * @return the spec + */ + public S batchMessageConverter(BatchMessageConverter messageConverter) { + this.target.setBatchMessageConverter(messageConverter); + return _this(); + } + + /** + * Specify a {@link RecordFilterStrategy} to wrap + * {@code KafkaMessageDrivenChannelAdapter.IntegrationRecordMessageListener} into + * {@link FilteringAcknowledgingMessageListenerAdapter}. + * @param recordFilterStrategy the {@link RecordFilterStrategy} to use. + * @return the spec + */ + public S recordFilterStrategy(RecordFilterStrategy recordFilterStrategy) { + this.target.setRecordFilterStrategy(recordFilterStrategy); + return _this(); + } + + /** + * A {@code boolean} flag to indicate if {@link FilteringAcknowledgingMessageListenerAdapter} + * should acknowledge discarded records or not. + * Does not make sense if {@link #recordFilterStrategy(RecordFilterStrategy)} isn't specified. + * @param ackDiscarded true to ack (commit offset for) discarded messages. + * @return the spec + */ + public S ackDiscarded(boolean ackDiscarded) { + this.target.setAckDiscarded(ackDiscarded); + return _this(); + } + + /** + * Specify a {@link RetryTemplate} instance to wrap + * {@code KafkaMessageDrivenChannelAdapter.IntegrationRecordMessageListener} into + * {@link RetryingAcknowledgingMessageListenerAdapter}. + * @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(); + } + + /** + /** + * The {@code boolean} flag to specify the order how + * {@link RetryingAcknowledgingMessageListenerAdapter} and + * {@link FilteringAcknowledgingMessageListenerAdapter} are wrapped to each other, + * if both of them are present. + * Does not make sense if only one of {@link RetryTemplate} or + * {@link RecordFilterStrategy} is present, or any. + * @param filterInRetry the order for {@link RetryingAcknowledgingMessageListenerAdapter} and + * {@link FilteringAcknowledgingMessageListenerAdapter} wrapping. Defaults to {@code false}. + * @return the spec + */ + public S filterInRetry(boolean filterInRetry) { + this.target.setFilterInRetry(filterInRetry); + return _this(); + } + + /** + * A {@link ConcurrentMessageListenerContainer} configuration {@link KafkaMessageDrivenChannelAdapterSpec} + * extension. + * @param the key type. + * @param the value type. + */ + public static class KafkaMessageDrivenChannelAdapterListenerContainerSpec extends + KafkaMessageDrivenChannelAdapterSpec> + implements ComponentsRegistration { + + private KafkaMessageListenerContainerSpec spec; + + KafkaMessageDrivenChannelAdapterListenerContainerSpec(KafkaMessageListenerContainerSpec spec, + KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) { + super(spec.container, listenerMode); + this.spec = spec; + } + + /** + * 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 KafkaMessageDrivenChannelAdapterListenerContainerSpec configureListenerContainer( + Consumer> configurer) { + Assert.notNull(configurer); + configurer.accept(this.spec); + return _this(); + } + + @Override + public Collection getComponentsToRegister() { + return Collections.singleton(this.spec.container); + } + + } + + /** + * 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 { + + 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)); + } + + /** + * 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 executor for threads that invoke the listener. + * @param listenerTaskExecutor the executor + * @return the spec. + * @see ContainerProperties#setListenerTaskExecutor(AsyncListenableTaskExecutor) + */ + public KafkaMessageListenerContainerSpec listenerTaskExecutor( + AsyncListenableTaskExecutor listenerTaskExecutor) { + this.container.getContainerProperties().setListenerTaskExecutor(listenerTaskExecutor); + return this; + } + + /** + * When using Kafka group management and {@link #pauseEnabled(boolean)} is + * true, set the delay after which the consumer should be paused. Default 10000. + * @param pauseAfter the delay. + * @return the spec. + * @see ContainerProperties#setPauseAfter(long) + */ + public KafkaMessageListenerContainerSpec pauseAfter(long pauseAfter) { + this.container.getContainerProperties().setPauseAfter(pauseAfter); + return this; + } + + /** + * Set to true to avoid rebalancing when this consumer is slow or throws a + * qualifying exception - pause the consumer. Default: true. + * @param pauseEnabled true to pause. + * @return the spec. + * @see #pauseAfter(long) + * @see ContainerProperties#setPauseEnabled(boolean) + */ + public KafkaMessageListenerContainerSpec pauseEnabled(boolean pauseEnabled) { + this.container.getContainerProperties().setPauseEnabled(pauseEnabled); + return this; + } + + /** + * Set the queue depth for handoffs from the consumer thread to the listener + * thread. Default 1 (up to 2 in process). + * @param queueDepth the queue depth. + * @return the spec. + * @see ContainerProperties#setQueueDepth(int) + */ + public KafkaMessageListenerContainerSpec queueDepth(int queueDepth) { + this.container.getContainerProperties().setQueueDepth(queueDepth); + 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; + } + + } + +} 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 new file mode 100644 index 0000000000..74824d9754 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java @@ -0,0 +1,251 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.dsl; + +import java.util.Collection; +import java.util.Collections; +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.MessageHandlerSpec; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.expression.ValueExpression; +import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.support.ProducerListener; +import org.springframework.kafka.support.converter.RecordMessageConverter; +import org.springframework.messaging.Message; + +/** + * A {@link MessageHandlerSpec} implementation for the {@link KafkaProducerMessageHandler}. + * + * @param the key type. + * @param the value type. + * + * @author Artem Bilan + * + * @since 3.0 + */ +public class KafkaProducerMessageHandlerSpec + extends MessageHandlerSpec, KafkaProducerMessageHandler> { + + protected final KafkaTemplate kafkaTemplate; + + KafkaProducerMessageHandlerSpec(KafkaTemplate kafkaTemplate) { + this.target = new KafkaProducerMessageHandler(kafkaTemplate); + this.kafkaTemplate = kafkaTemplate; + } + + /** + * Configure the Kafka topic to send messages. + * @param topic the Kafka topic name. + * @return the spec. + */ + public KafkaProducerMessageHandlerSpec topic(String topic) { + return topicExpression(new LiteralExpression(topic)); + } + + + /** + * Configure a SpEL expression to determine the Kafka topic at runtime against + * request Message as a root object of evaluation context. + * @param topicExpression the topic SpEL expression. + * @return the spec. + */ + public KafkaProducerMessageHandlerSpec topicExpression(String topicExpression) { + return topicExpression(PARSER.parseExpression(topicExpression)); + } + + /** + * Configure an {@link Expression} to determine the Kafka topic at runtime against + * request Message as a root object of evaluation context. + * @param topicExpression the topic expression. + * @return the spec. + */ + public KafkaProducerMessageHandlerSpec topicExpression(Expression topicExpression) { + this.target.setTopicExpression(topicExpression); + return _this(); + } + + /** + * Configure a {@link Function} that will be invoked at run time to determine the topic to + * which a message will be sent. Typically used with a Java 8 Lambda expression: + * 
+	 * {@code
+	 * .topic(m -> m.getPayload().getTopic())
+	 * }
+	 *
+ * @param topicFunction the topic function. + * @param

the expected payload type. + * @return the current {@link KafkaProducerMessageHandlerSpec}. + * @see FunctionExpression + */ + public

KafkaProducerMessageHandlerSpec topic(Function, String> topicFunction) { + return topicExpression(new FunctionExpression<>(topicFunction)); + } + + /** + * Configure a SpEL expression to determine the Kafka message key to store at runtime against + * request Message as a root object of evaluation context. + * @param messageKeyExpression the message key SpEL expression. + * @return the spec. + */ + public KafkaProducerMessageHandlerSpec messageKeyExpression(String messageKeyExpression) { + return messageKeyExpression(PARSER.parseExpression(messageKeyExpression)); + } + + /** + * Configure the message key to store message in Kafka topic. + * @param messageKey the message key to use. + * @return the spec. + */ + public KafkaProducerMessageHandlerSpec messageKey(String messageKey) { + return messageKeyExpression(new LiteralExpression(messageKey)); + } + + /** + * Configure an {@link Expression} to determine the Kafka message key to store at runtime against + * request Message as a root object of evaluation context. + * @param messageKeyExpression the message key expression. + * @return the spec. + */ + public KafkaProducerMessageHandlerSpec messageKeyExpression(Expression messageKeyExpression) { + this.target.setMessageKeyExpression(messageKeyExpression); + return _this(); + } + + /** + * Configure a {@link Function} that will be invoked at run time to determine the message key under + * which a message will be stored in the topic. Typically used with a Java 8 Lambda expression: + * 

+	 * {@code
+	 * .messageKey(m -> m.getPayload().getKey())
+	 * }
+	 *
+ * @param messageKeyFunction the message key function. + * @param

the expected payload type. + * @return the current {@link KafkaProducerMessageHandlerSpec}. + * @see FunctionExpression + */ + public

KafkaProducerMessageHandlerSpec messageKey(Function, ?> messageKeyFunction) { + return messageKeyExpression(new FunctionExpression<>(messageKeyFunction)); + } + + /** + * Configure a partitionId of Kafka topic. + * @param partitionId the partitionId to use. + * @return the spec. + */ + public KafkaProducerMessageHandlerSpec partitionId(Integer partitionId) { + return partitionIdExpression(new ValueExpression(partitionId)); + } + + /** + * Configure a SpEL expression to determine the topic partitionId at runtime against + * request Message as a root object of evaluation context. + * @param partitionIdExpression the partitionId expression to use. + * @return the spec. + */ + public KafkaProducerMessageHandlerSpec partitionIdExpression(String partitionIdExpression) { + return partitionIdExpression(PARSER.parseExpression(partitionIdExpression)); + } + + /** + * Configure a {@link Function} that will be invoked at run time to determine the partition id under + * which a message will be stored in the topic. Typically used with a Java 8 Lambda expression: + * 

+	 * {@code
+	 * .partitionId(m -> m.getHeaders().get("partitionId", Integer.class))
+	 * }
+	 *
+ * @param partitionIdFunction the partitionId function. + * @param

the expected payload type. + * @return the spec. + */ + public

KafkaProducerMessageHandlerSpec partitionId(Function, Integer> partitionIdFunction) { + return partitionIdExpression(new FunctionExpression<>(partitionIdFunction)); + } + + /** + * Configure an {@link Expression} to determine the topic partitionId at runtime against + * request Message as a root object of evaluation context. + * @param partitionIdExpression the partitionId expression to use. + * @return the spec. + */ + public KafkaProducerMessageHandlerSpec partitionIdExpression(Expression partitionIdExpression) { + this.target.setPartitionIdExpression(partitionIdExpression); + return _this(); + } + + /** + * A {@code boolean} indicating if the {@link KafkaProducerMessageHandler} + * should wait for the send operation results or not. Defaults to {@code false}. + * In {@code sync} mode a downstream send operation exception will be re-thrown. + * @param sync the send mode; async by default. + * @return the spec. + */ + public KafkaProducerMessageHandlerSpec sync(boolean sync) { + this.target.setSync(sync); + return this; + } + + /** + * Specify a timeout in milliseconds how long {@link KafkaProducerMessageHandler} + * should wait wait for send operation results. Defaults to 10 seconds. + * @param sendTimeout the timeout to wait for result fo send operation. + * @return the spec. + */ + public KafkaProducerMessageHandlerSpec sendTimeout(long sendTimeout) { + this.target.setSendTimeout(sendTimeout); + return this; + } + + /** + * A {@link KafkaTemplate}-based {@link KafkaProducerMessageHandlerSpec} extension. + * + * @param the key type. + * @param the value type. + */ + public static class KafkaProducerMessageHandlerTemplateSpec extends KafkaProducerMessageHandlerSpec + implements ComponentsRegistration { + + KafkaProducerMessageHandlerTemplateSpec(ProducerFactory producerFactory) { + super(new KafkaTemplate<>(producerFactory)); + } + + public KafkaProducerMessageHandlerTemplateSpec producerListener(ProducerListener producerListener) { + this.kafkaTemplate.setProducerListener(producerListener); + return this; + } + + public KafkaProducerMessageHandlerTemplateSpec messageConverter(RecordMessageConverter messageConverter) { + this.kafkaTemplate.setMessageConverter(messageConverter); + return this; + } + + @Override + public Collection getComponentsToRegister() { + return Collections.singleton(this.kafkaTemplate); + } + + } + +} + diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/package-info.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/package-info.java new file mode 100644 index 0000000000..adf6553c27 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides Kafka Components support for Spring Integration Java DSL. + */ +package org.springframework.integration.kafka.dsl; diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaTests.java new file mode 100644 index 0000000000..ece137980d --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaTests.java @@ -0,0 +1,201 @@ +/* + * Copyright 2015-2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.dsl; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.util.Map; +import java.util.stream.Stream; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.MessageRejectedException; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.expression.ValueExpression; +import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; +import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; +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.ProducerFactory; +import org.springframework.kafka.listener.AbstractMessageListenerContainer; +import org.springframework.kafka.support.Acknowledgment; +import org.springframework.kafka.support.KafkaHeaders; +import org.springframework.kafka.test.rule.KafkaEmbedded; +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.PollableChannel; +import org.springframework.messaging.support.ErrorMessage; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.retry.support.RetryTemplate; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * @author Artem Bilan + * @author Nasko Vasilev + * + * @since 3.0 + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class KafkaTests { + + private static final String TEST_TOPIC = "test-topic"; + + private static final String TEST_TOPIC2 = "test-topic2"; + + private static final String TEST_TOPIC3 = "test-topic3"; + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, TEST_TOPIC, TEST_TOPIC2, TEST_TOPIC3); + + @Autowired + @Qualifier("sendToKafkaFlow.input") + private MessageChannel sendToKafkaFlowInput; + + @Autowired + private PollableChannel listeningFromKafkaResults; + + @Autowired + @Qualifier("kafkaProducer.handler") + private KafkaProducerMessageHandler kafkaProducer; + + @Autowired + @Qualifier("kafkaProducer3.handler") + private KafkaProducerMessageHandler kafkaProducer3; + + @Autowired + private PollableChannel errorChannel; + + @Test + public void testKafkaAdapters() { + + assertThatThrownBy(() -> this.sendToKafkaFlowInput.send(new GenericMessage<>("foo"))) + .hasMessageContaining("10 is not in the range"); + + this.kafkaProducer.setPartitionIdExpression(new ValueExpression<>(0)); + this.kafkaProducer3.setPartitionIdExpression(new ValueExpression<>(0)); + this.sendToKafkaFlowInput.send(new GenericMessage<>("foo")); + + for (int i = 0; i < 100; i++) { + Message receive = this.listeningFromKafkaResults.receive(10000); + assertThat(receive).isNotNull(); + assertThat(receive.getPayload()).isEqualTo("FOO"); + MessageHeaders headers = receive.getHeaders(); + assertThat(headers.containsKey(KafkaHeaders.ACKNOWLEDGMENT)).isTrue(); + Acknowledgment acknowledgment = headers.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class); + acknowledgment.acknowledge(); + assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(TEST_TOPIC); + assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo(i + 1); + assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0); + assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo((long) i); + } + + Message message = MessageBuilder.withPayload("BAR").setHeader(KafkaHeaders.TOPIC, TEST_TOPIC2).build(); + + this.sendToKafkaFlowInput.send(message); + + assertThat(this.listeningFromKafkaResults.receive(10)).isNull(); + + Message error = this.errorChannel.receive(10000); + assertThat(error).isNotNull(); + assertThat(error).isInstanceOf(ErrorMessage.class); + assertThat(error.getPayload()).isInstanceOf(MessageRejectedException.class); + } + + @Configuration + @EnableIntegration + public static class ContextConfiguration { + + + @Bean + public ConsumerFactory consumerFactory() { + Map props = KafkaTestUtils.consumerProps("test1", "false", embeddedKafka); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + return new DefaultKafkaConsumerFactory<>(props); + } + + @Bean + public PollableChannel errorChannel() { + return new QueueChannel(); + } + + @Bean + public IntegrationFlow listeningFromKafkaFlow() { + return IntegrationFlows + .from(Kafka.messageDrivenChannelAdapter(consumerFactory(), + KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC) + .configureListenerContainer(c -> + c.ackMode(AbstractMessageListenerContainer.AckMode.MANUAL)) + .errorChannel("errorChannel") + .retryTemplate(new RetryTemplate()) + .filterInRetry(true)) + .filter(Message.class, m -> + m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer.class) < 101, + f -> f.throwExceptionOnRejection(true)) + .transform(String::toUpperCase) + .channel(c -> c.queue("listeningFromKafkaResults")) + .get(); + } + + @Bean + public ProducerFactory producerFactory() { + return new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(embeddedKafka)); + } + + @Bean + public IntegrationFlow sendToKafkaFlow() { + return f -> f + .split(p -> Stream.generate(() -> p).limit(101).iterator(), null) + .publishSubscribeChannel(c -> c + .subscribe(sf -> sf.handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC), + e -> e.id("kafkaProducer"))) + .subscribe(sf -> sf.handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC3), + e -> e.id("kafkaProducer3"))) + ); + } + + private KafkaProducerMessageHandlerSpec + kafkaMessageHandler(ProducerFactory producerFactory, String topic) { + return Kafka + .outboundChannelAdapter(producerFactory) + .messageKey(m -> m + .getHeaders() + .get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)) + .partitionId(m -> 10) + .topicExpression("headers[kafka_topic] ?: '" + topic + "'"); + } + + } + +} diff --git a/spring-integration-kafka/src/test/resources/log4j2-test.xml b/spring-integration-kafka/src/test/resources/log4j2-test.xml index a2b7417bca..b2735637cf 100644 --- a/spring-integration-kafka/src/test/resources/log4j2-test.xml +++ b/spring-integration-kafka/src/test/resources/log4j2-test.xml @@ -8,8 +8,8 @@ - - + +