diff --git a/pom.xml b/pom.xml index 651155eb9..6c4c210e4 100644 --- a/pom.xml +++ b/pom.xml @@ -7,10 +7,11 @@ org.springframework.cloud spring-cloud-build - 1.1.1.RELEASE + 1.1.2.BUILD-SNAPSHOT + 1.7 1.4.0.BUILD-SNAPSHOT diff --git a/spring-cloud-starter-stream-kafka/src/main/resources/META-INF/spring.provides b/spring-cloud-starter-stream-kafka/src/main/resources/META-INF/spring.provides index 0aa482658..cc7cb9cc2 100644 --- a/spring-cloud-starter-stream-kafka/src/main/resources/META-INF/spring.provides +++ b/spring-cloud-starter-stream-kafka/src/main/resources/META-INF/spring.provides @@ -1 +1 @@ -provides: spring-cloud-stream-binder-kafka \ No newline at end of file +provides: spring-cloud-starter-stream-kafka \ No newline at end of file diff --git a/spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/KafkaTestSupport.java b/spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/KafkaTestSupport.java index 16d89e672..9184f2467 100644 --- a/spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/KafkaTestSupport.java +++ b/spring-cloud-stream-binder-kafka-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/kafka/KafkaTestSupport.java @@ -86,12 +86,12 @@ public class KafkaTestSupport extends AbstractExternalResourceTestSupport - 0.8.2.2 - 2.6.0 - 1.3.1.BUILD-SNAPSHOT + 0.9.0.1 + 1.0.3.BUILD-SNAPSHOT + 2.0.1.BUILD-SNAPSHOT 1.0.0 + + org.springframework.cloud + spring-cloud-stream-binder-kafka-common + ${project.version} + org.springframework.boot spring-boot-configuration-processor @@ -61,9 +66,20 @@ + + org.springframework.kafka + spring-kafka + ${spring-kafka.version} + + + org.springframework.kafka + spring-kafka-test + test + ${spring-kafka.version} + org.apache.kafka - kafka_2.10 + kafka_2.11 org.apache.kafka @@ -78,29 +94,19 @@ rxjava-math ${rxjava-math.version} - - org.apache.curator - curator-recipes - test - org.apache.kafka - kafka_2.10 + kafka_2.11 test test - - org.apache.curator - curator-test - test - org.apache.kafka - kafka_2.10 + kafka_2.11 ${kafka.version} @@ -115,7 +121,7 @@ org.apache.kafka - kafka_2.10 + kafka_2.11 test ${kafka.version} @@ -124,21 +130,6 @@ kafka-clients ${kafka.version} - - org.apache.curator - curator-framework - ${curator.version} - - - org.apache.curator - curator-recipes - ${curator.version} - - - org.apache.curator - curator-test - ${curator.version} - diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java deleted file mode 100644 index c523f0854..000000000 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java +++ /dev/null @@ -1,95 +0,0 @@ -/* - * 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.cloud.stream.binder.kafka; - -import java.util.Collection; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; - -import kafka.cluster.Broker; -import kafka.utils.ZKStringSerializer$; -import kafka.utils.ZkUtils$; -import org.I0Itec.zkclient.ZkClient; -import scala.collection.JavaConversions; -import scala.collection.Seq; - -import org.springframework.boot.actuate.health.Health; -import org.springframework.boot.actuate.health.HealthIndicator; -import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties; -import org.springframework.integration.kafka.core.BrokerAddress; -import org.springframework.integration.kafka.core.Partition; - -/** - * Health indicator for Kafka. - * - * @author Ilayaperumal Gopinathan - */ -public class KafkaBinderHealthIndicator implements HealthIndicator { - - private final KafkaMessageChannelBinder binder; - - private final KafkaBinderConfigurationProperties configurationProperties; - - public KafkaBinderHealthIndicator(KafkaMessageChannelBinder binder, - KafkaBinderConfigurationProperties configurationProperties) { - this.binder = binder; - this.configurationProperties = configurationProperties; - } - - @Override - public Health health() { - ZkClient zkClient = null; - try { - zkClient = new ZkClient(configurationProperties.getZkConnectionString(), - configurationProperties.getZkSessionTimeout(), - configurationProperties.getZkConnectionTimeout(), ZKStringSerializer$.MODULE$); - Set brokersInClusterSet = new HashSet<>(); - Seq allBrokersInCluster = ZkUtils$.MODULE$.getAllBrokersInCluster(zkClient); - Collection brokersInCluster = JavaConversions.asJavaCollection(allBrokersInCluster); - for (Broker broker : brokersInCluster) { - brokersInClusterSet.add(broker.connectionString()); - } - Set downMessages = new HashSet<>(); - for (Map.Entry> entry : binder.getTopicsInUse().entrySet()) { - for (Partition partition : entry.getValue()) { - BrokerAddress address = binder.getConnectionFactory().getLeader(partition); - if (!brokersInClusterSet.contains(address.toString())) { - downMessages.add(address.toString()); - } - } - } - if (downMessages.isEmpty()) { - return Health.up().build(); - } - return Health.down().withDetail("Following brokers are down: ", downMessages.toString()).build(); - } - catch (Exception e) { - return Health.down(e).build(); - } - finally { - if (zkClient != null) { - try { - zkClient.close(); - } - catch (Exception e) { - // ignore - } - } - } - } -} diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaConsumerProperties.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaConsumerProperties.java index c52a38221..33976bf83 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaConsumerProperties.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaConsumerProperties.java @@ -21,13 +21,15 @@ package org.springframework.cloud.stream.binder.kafka; */ public class KafkaConsumerProperties { + private boolean autoRebalanceEnabled = true; + private boolean autoCommitOffset = true; private Boolean autoCommitOnError; private boolean resetOffsets; - private KafkaMessageChannelBinder.StartOffset startOffset; + private StartOffset startOffset; private boolean enableDlq; @@ -49,11 +51,11 @@ public class KafkaConsumerProperties { this.resetOffsets = resetOffsets; } - public KafkaMessageChannelBinder.StartOffset getStartOffset() { + public StartOffset getStartOffset() { return startOffset; } - public void setStartOffset(KafkaMessageChannelBinder.StartOffset startOffset) { + public void setStartOffset(StartOffset startOffset) { this.startOffset = startOffset; } @@ -80,4 +82,26 @@ public class KafkaConsumerProperties { public void setRecoveryInterval(int recoveryInterval) { this.recoveryInterval = recoveryInterval; } + + public boolean isAutoRebalanceEnabled() { + return autoRebalanceEnabled; + } + + public void setAutoRebalanceEnabled(boolean autoRebalanceEnabled) { + this.autoRebalanceEnabled = autoRebalanceEnabled; + } + + public enum StartOffset { + earliest(-2L), latest(-1L); + + private final long referencePoint; + + StartOffset(long referencePoint) { + this.referencePoint = referencePoint; + } + + public long getReferencePoint() { + return referencePoint; + } + } } diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaExtendedBindingProperties.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaExtendedBindingProperties.java index 637d14a35..25be14982 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaExtendedBindingProperties.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaExtendedBindingProperties.java @@ -26,12 +26,13 @@ import org.springframework.cloud.stream.binder.ExtendedBindingProperties; * @author Marius Bogoevici */ @ConfigurationProperties("spring.cloud.stream.kafka") -public class KafkaExtendedBindingProperties implements ExtendedBindingProperties { +public class KafkaExtendedBindingProperties + implements ExtendedBindingProperties { private Map bindings = new HashMap<>(); public Map getBindings() { - return bindings; + return this.bindings; } public void setBindings(Map bindings) { @@ -40,8 +41,8 @@ public class KafkaExtendedBindingProperties implements ExtendedBindingProperties @Override public KafkaConsumerProperties getExtendedConsumerProperties(String channelName) { - if (bindings.containsKey(channelName) && bindings.get(channelName).getConsumer() != null) { - return bindings.get(channelName).getConsumer(); + if (this.bindings.containsKey(channelName) && this.bindings.get(channelName).getConsumer() != null) { + return this.bindings.get(channelName).getConsumer(); } else { return new KafkaConsumerProperties(); @@ -50,8 +51,8 @@ public class KafkaExtendedBindingProperties implements ExtendedBindingProperties @Override public KafkaProducerProperties getExtendedProducerProperties(String channelName) { - if (bindings.containsKey(channelName) && bindings.get(channelName).getProducer() != null) { - return bindings.get(channelName).getProducer(); + if (this.bindings.containsKey(channelName) && this.bindings.get(channelName).getProducer() != null) { + return this.bindings.get(channelName).getProducer(); } else { return new KafkaProducerProperties(); diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index e4b3b390e..ac1f5b59f 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -16,33 +16,32 @@ package org.springframework.cloud.stream.binder.kafka; -import java.io.UnsupportedEncodingException; +import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; -import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.Properties; import java.util.UUID; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.ThreadFactory; import kafka.admin.AdminUtils; -import kafka.api.OffsetRequest; import kafka.api.TopicMetadata; import kafka.common.ErrorMapping; -import kafka.serializer.DefaultDecoder; import kafka.utils.ZKStringSerializer$; import kafka.utils.ZkUtils; import org.I0Itec.zkclient.ZkClient; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.PartitionInfo; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; import org.apache.kafka.common.serialization.ByteArraySerializer; +import org.apache.kafka.common.serialization.Deserializer; import org.apache.kafka.common.utils.Utils; import scala.collection.Seq; @@ -56,37 +55,29 @@ import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties; import org.springframework.context.Lifecycle; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.core.MessageProducer; -import org.springframework.integration.kafka.core.ConnectionFactory; -import org.springframework.integration.kafka.core.DefaultConnectionFactory; -import org.springframework.integration.kafka.core.KafkaMessage; -import org.springframework.integration.kafka.core.Partition; -import org.springframework.integration.kafka.core.ZookeeperConfiguration; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; -import org.springframework.integration.kafka.listener.AcknowledgingMessageListener; -import org.springframework.integration.kafka.listener.Acknowledgment; -import org.springframework.integration.kafka.listener.ErrorHandler; -import org.springframework.integration.kafka.listener.KafkaMessageListenerContainer; -import org.springframework.integration.kafka.listener.KafkaNativeOffsetManager; -import org.springframework.integration.kafka.listener.MessageListener; -import org.springframework.integration.kafka.listener.OffsetManager; -import org.springframework.integration.kafka.support.KafkaProducerContext; -import org.springframework.integration.kafka.support.ProducerConfiguration; -import org.springframework.integration.kafka.support.ProducerFactoryBean; -import org.springframework.integration.kafka.support.ProducerListener; -import org.springframework.integration.kafka.support.ProducerMetadata; -import org.springframework.integration.kafka.support.ZookeeperConnect; -import org.springframework.messaging.Message; +import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; +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.ConcurrentMessageListenerContainer; +import org.springframework.kafka.listener.ErrorHandler; +import org.springframework.kafka.listener.config.ContainerProperties; +import org.springframework.kafka.support.ProducerListener; +import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.MessagingException; import org.springframework.retry.RetryCallback; import org.springframework.retry.RetryContext; import org.springframework.retry.RetryOperations; import org.springframework.retry.backoff.ExponentialBackOffPolicy; import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; -import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; @@ -104,31 +95,17 @@ import org.springframework.util.StringUtils; */ public class KafkaMessageChannelBinder extends AbstractMessageChannelBinder, - ExtendedProducerProperties, Collection> + ExtendedProducerProperties, Collection> implements ExtendedPropertiesBinder, DisposableBean { - private static final ByteArraySerializer BYTE_ARRAY_SERIALIZER = new ByteArraySerializer(); - - private static final ThreadFactory DAEMON_THREAD_FACTORY; - - static { - CustomizableThreadFactory threadFactory = new CustomizableThreadFactory("kafka-binder-"); - threadFactory.setDaemon(true); - DAEMON_THREAD_FACTORY = threadFactory; - } - private final KafkaBinderConfigurationProperties configurationProperties; private RetryOperations metadataRetryOperations; - private final Map> topicsInUse = new HashMap<>(); + private final Map> topicsInUse = new HashMap<>(); - // -------- Default values for properties ------- - - private ConnectionFactory connectionFactory; - - private ProducerListener producerListener; + private ProducerListener producerListener; private volatile Producer dlqProducer; @@ -155,14 +132,6 @@ public class KafkaMessageChannelBinder extends return headersToMap; } - ConnectionFactory getConnectionFactory() { - return this.connectionFactory; - } - - public void setProducerListener(ProducerListener producerListener) { - this.producerListener = producerListener; - } - /** * Retry configuration for operations such as validating topic creation * @param metadataRetryOperations the retry configuration @@ -177,13 +146,7 @@ public class KafkaMessageChannelBinder extends @Override public void onInit() throws Exception { - ZookeeperConfiguration configuration = new ZookeeperConfiguration( - new ZookeeperConnect(this.configurationProperties.getZkConnectionString())); - configuration.setBufferSize(this.configurationProperties.getSocketBufferSize()); - configuration.setMaxWait(this.configurationProperties.getMaxWait()); - DefaultConnectionFactory defaultConnectionFactory = new DefaultConnectionFactory(configuration); - defaultConnectionFactory.afterPropertiesSet(); - this.connectionFactory = defaultConnectionFactory; + if (this.metadataRetryOperations == null) { RetryTemplate retryTemplate = new RetryTemplate(); @@ -208,23 +171,12 @@ public class KafkaMessageChannelBinder extends } } - /** - * Allowed chars are ASCII alphanumerics, '.', '_' and '-'. - */ - static void validateTopicName(String topicName) { - try { - byte[] utf8 = topicName.getBytes("UTF-8"); - for (byte b : utf8) { - if (!((b >= 'a') && (b <= 'z') || (b >= 'A') && (b <= 'Z') || (b >= '0') && (b <= '9') || (b == '.') - || (b == '-') || (b == '_'))) { - throw new IllegalArgumentException( - "Topic name can only have ASCII alphanumerics, '.', '_' and '-'"); - } - } - } - catch (UnsupportedEncodingException e) { - throw new AssertionError(e); // Can't happen - } + public void setProducerListener(ProducerListener producerListener) { + this.producerListener = producerListener; + } + + Map> getTopicsInUse() { + return this.topicsInUse; } @Override @@ -237,30 +189,89 @@ public class KafkaMessageChannelBinder extends return this.extendedBindingProperties.getExtendedProducerProperties(channelName); } - Map> getTopicsInUse() { - return this.topicsInUse; + @Override + protected MessageHandler createProducerMessageHandler(final String name, + ExtendedProducerProperties producerProperties) throws Exception { + + KafkaTopicUtils.validateTopicName(name); + + Collection partitions = ensureTopicCreated(name, producerProperties.getPartitionCount()); + + if (producerProperties.getPartitionCount() < partitions.size()) { + if (this.logger.isInfoEnabled()) { + this.logger.info("The `partitionCount` of the producer for topic " + name + " is " + + producerProperties.getPartitionCount() + ", smaller than the actual partition count of " + + partitions.size() + " of the topic. The larger number will be used instead."); + } + } + + this.topicsInUse.put(name, partitions); + + ProducerFactory producerFB = getProducerFactory(producerProperties); + KafkaTemplate kafkaTemplate = new KafkaTemplate<>(producerFB); + if (this.producerListener != null) { + kafkaTemplate.setProducerListener(this.producerListener); + } + return new ProducerConfigurationMessageHandler(kafkaTemplate, name, producerProperties); } @Override - protected Collection createConsumerDestinationIfNecessary(String name, String group, + protected void createProducerDestinationIfNecessary(String name, + ExtendedProducerProperties properties) { + if (this.logger.isInfoEnabled()) { + this.logger.info("Using kafka topic for outbound: " + name); + } + KafkaTopicUtils.validateTopicName(name); + Collection partitions = ensureTopicCreated(name, properties.getPartitionCount()); + if (properties.getPartitionCount() < partitions.size()) { + if (this.logger.isInfoEnabled()) { + this.logger.info("The `partitionCount` of the producer for topic " + name + " is " + + properties.getPartitionCount() + ", smaller than the actual partition count of " + + partitions.size() + " of the topic. The larger number will be used instead."); + } + } + this.topicsInUse.put(name, partitions); + } + + private ProducerFactory getProducerFactory( + ExtendedProducerProperties producerProperties) { + Map props = new HashMap<>(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString()); + props.put(ProducerConfig.RETRIES_CONFIG, 0); + props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); + props.put(ProducerConfig.LINGER_MS_CONFIG, 1); + props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); + props.put(ProducerConfig.ACKS_CONFIG, String.valueOf(this.configurationProperties.getRequiredAcks())); + props.put(ProducerConfig.LINGER_MS_CONFIG, + String.valueOf(producerProperties.getExtension().getBatchTimeout())); + props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, + producerProperties.getExtension().getCompressionType().toString()); + + return new DefaultKafkaProducerFactory<>(props); + } + + @Override + protected Collection createConsumerDestinationIfNecessary(String name, String group, ExtendedConsumerProperties properties) { - validateTopicName(name); + KafkaTopicUtils.validateTopicName(name); if (properties.getInstanceCount() == 0) { throw new IllegalArgumentException("Instance count cannot be zero"); } - Collection allPartitions = ensureTopicCreated(name, + Collection allPartitions = ensureTopicCreated(name, properties.getInstanceCount() * properties.getConcurrency()); - Collection listenedPartitions; + Collection listenedPartitions; if (properties.getInstanceCount() == 1) { listenedPartitions = allPartitions; } else { listenedPartitions = new ArrayList<>(); - for (Partition partition : allPartitions) { + for (PartitionInfo partition : allPartitions) { // divide partitions across modules - if ((partition.getId() % properties.getInstanceCount()) == properties.getInstanceIndex()) { + if ((partition.partition() % properties.getInstanceCount()) == properties.getInstanceIndex()) { listenedPartitions.add(partition); } } @@ -269,135 +280,73 @@ public class KafkaMessageChannelBinder extends return listenedPartitions; } - @Override @SuppressWarnings("unchecked") - protected MessageProducer createConsumerEndpoint(String name, String group, Collection destination, + protected MessageProducer createConsumerEndpoint(String name, String group, Collection destination, ExtendedConsumerProperties properties) { - - Assert.isTrue(!CollectionUtils.isEmpty(destination), "A list of partitions must be provided"); - - int concurrency = Math.min(properties.getConcurrency(), destination.size()); - - final ExecutorService dispatcherTaskExecutor = - Executors.newFixedThreadPool(concurrency, DAEMON_THREAD_FACTORY); - final KafkaMessageListenerContainer messageListenerContainer = new KafkaMessageListenerContainer( - this.connectionFactory, destination.toArray(new Partition[destination.size()])) { - - @Override - public void stop(Runnable callback) { - super.stop(callback); - if (getOffsetManager() instanceof DisposableBean) { - try { - ((DisposableBean) getOffsetManager()).destroy(); - } - catch (Exception e) { - KafkaMessageChannelBinder.this.logger.error("Error while closing the offset manager", e); - } - } - dispatcherTaskExecutor.shutdown(); - } - }; - - if (this.logger.isDebugEnabled()) { - this.logger.debug( - "Listened partitions: " + StringUtils.collectionToCommaDelimitedString(destination)); - } - boolean anonymous = !StringUtils.hasText(group); Assert.isTrue(!anonymous || !properties.getExtension().isEnableDlq(), "DLQ support is not available for anonymous subscriptions"); String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString() : group; - long referencePoint = properties.getExtension().getStartOffset() != null - ? properties.getExtension().getStartOffset().getReferencePoint() - : (anonymous ? OffsetRequest.LatestTime() : OffsetRequest.EarliestTime()); - OffsetManager offsetManager = createOffsetManager(consumerGroup, referencePoint); - if (properties.getExtension().isResetOffsets()) { - offsetManager.resetOffsets(destination); - } - messageListenerContainer.setOffsetManager(offsetManager); - messageListenerContainer.setQueueSize(this.configurationProperties.getQueueSize()); - messageListenerContainer.setMaxFetch(this.configurationProperties.getFetchSize()); - boolean autoCommitOnError = properties.getExtension().getAutoCommitOnError() != null - ? properties.getExtension().getAutoCommitOnError() - : properties.getExtension().isAutoCommitOffset() && properties.getExtension().isEnableDlq(); - messageListenerContainer.setAutoCommitOnError(autoCommitOnError); - messageListenerContainer.setRecoveryInterval(properties.getExtension().getRecoveryInterval()); + Map props = getConsumerConfig(anonymous, consumerGroup); + Deserializer valueDecoder = new ByteArrayDeserializer(); + Deserializer keyDecoder = new ByteArrayDeserializer(); + + ConsumerFactory consumerFactory = new DefaultKafkaConsumerFactory<>(props, keyDecoder, + valueDecoder); + + Collection listenedPartitions = (Collection) destination; + Assert.isTrue(!CollectionUtils.isEmpty(listenedPartitions), "A list of partitions must be provided"); + final TopicPartitionInitialOffset[] topicPartitionInitialOffsets = getTopicPartitionInitialOffsets( + listenedPartitions); + + final ContainerProperties containerProperties = + anonymous || properties.getExtension().isAutoRebalanceEnabled() ? new ContainerProperties(name) + : new ContainerProperties(topicPartitionInitialOffsets); + + int concurrency = Math.min(properties.getConcurrency(), listenedPartitions.size()); + final ConcurrentMessageListenerContainer messageListenerContainer = + new ConcurrentMessageListenerContainer( + consumerFactory, containerProperties) { + + @Override + public void stop(Runnable callback) { + super.stop(callback); + } + }; messageListenerContainer.setConcurrency(concurrency); - messageListenerContainer.setDispatcherTaskExecutor(dispatcherTaskExecutor); - final KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter = new KafkaMessageDrivenChannelAdapter( - messageListenerContainer); - kafkaMessageDrivenChannelAdapter.setBeanFactory(this.getBeanFactory()); - kafkaMessageDrivenChannelAdapter.setKeyDecoder(new DefaultDecoder(null)); - kafkaMessageDrivenChannelAdapter.setPayloadDecoder(new DefaultDecoder(null)); - kafkaMessageDrivenChannelAdapter.setAutoCommitOffset(properties.getExtension().isAutoCommitOffset()); - kafkaMessageDrivenChannelAdapter.afterPropertiesSet(); - if (properties.getMaxAttempts() > 1) { - // we need to wrap the adapter listener into a retrying listener so that the retry - // logic is applied before the ErrorHandler is executed - final RetryTemplate retryTemplate = buildRetryTemplate(properties); - if (properties.getExtension().isAutoCommitOffset()) { - final MessageListener originalMessageListener = (MessageListener) messageListenerContainer - .getMessageListener(); - messageListenerContainer.setMessageListener(new MessageListener() { + messageListenerContainer.getContainerProperties().setAckOnError(isAutoCommitOnError(properties)); - @Override - public void onMessage(final KafkaMessage message) { - try { - retryTemplate.execute(new RetryCallback() { - - @Override - public Object doWithRetry(RetryContext context) { - originalMessageListener.onMessage(message); - return null; - } - }); - } - catch (Throwable throwable) { - if (throwable instanceof RuntimeException) { - throw (RuntimeException) throwable; - } - else { - throw new RuntimeException(throwable); - } - } - } - }); - } - else { - messageListenerContainer.setMessageListener(new AcknowledgingMessageListener() { - - final AcknowledgingMessageListener originalMessageListener = - (AcknowledgingMessageListener) messageListenerContainer - .getMessageListener(); - - @Override - public void onMessage(final KafkaMessage message, final Acknowledgment acknowledgment) { - retryTemplate.execute(new RetryCallback() { - - @Override - public Object doWithRetry(RetryContext context) { - originalMessageListener.onMessage(message, acknowledgment); - return null; - } - }); - } - }); - } + if (this.logger.isDebugEnabled()) { + this.logger.debug( + "Listened partitions: " + StringUtils.collectionToCommaDelimitedString(listenedPartitions)); } + if (this.logger.isDebugEnabled()) { + this.logger.debug( + "Listened partitions: " + StringUtils.collectionToCommaDelimitedString(listenedPartitions)); + } + + final KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter = + new KafkaMessageDrivenChannelAdapter<>( + messageListenerContainer); + + kafkaMessageDrivenChannelAdapter.setBeanFactory(this.getBeanFactory()); + final RetryTemplate retryTemplate = buildRetryTemplate(properties); + kafkaMessageDrivenChannelAdapter.setRetryTemplate(retryTemplate); + if (properties.getExtension().isEnableDlq()) { - final String dlqTopic = "error." + name + "." + consumerGroup; + final String dlqTopic = "error." + name + "." + group; initDlqProducer(); - messageListenerContainer.setErrorHandler(new ErrorHandler() { + messageListenerContainer.getContainerProperties().setErrorHandler(new ErrorHandler() { @Override - public void handle(Exception thrownException, final KafkaMessage message) { - final byte[] key = message.getMessage().key() != null ? Utils.toArray(message.getMessage().key()) + public void handle(Exception thrownException, final ConsumerRecord message) { + final byte[] key = message.key() != null ? Utils.toArray(ByteBuffer.wrap((byte[]) message.key())) : null; - final byte[] payload = message.getMessage().payload() != null - ? Utils.toArray(message.getMessage().payload()) : null; + final byte[] payload = message.value() != null + ? Utils.toArray(ByteBuffer.wrap((byte[]) message.value())) : null; KafkaMessageChannelBinder.this.dlqProducer.send(new ProducerRecord<>(dlqTopic, key, payload), new Callback() { @@ -408,7 +357,7 @@ public class KafkaMessageChannelBinder extends + toDisplayString(ObjectUtils.nullSafeToString(key), 50) + "'"); messageLog.append(" and payload='" + toDisplayString(ObjectUtils.nullSafeToString(payload), 50) + "'"); - messageLog.append(" received from " + message.getMetadata().getPartition()); + messageLog.append(" received from " + message.partition()); if (exception != null) { KafkaMessageChannelBinder.this.logger.error( "Error sending to DLQ" + messageLog.toString(), exception); @@ -427,74 +376,60 @@ public class KafkaMessageChannelBinder extends return kafkaMessageDrivenChannelAdapter; } - @Override - protected MessageHandler createProducerMessageHandler(final String destination, - ExtendedProducerProperties producerProperties) throws Exception { - ProducerMetadata producerMetadata = new ProducerMetadata<>(destination, byte[].class, - byte[].class, - BYTE_ARRAY_SERIALIZER, BYTE_ARRAY_SERIALIZER); - producerMetadata.setSync(producerProperties.getExtension().isSync()); - producerMetadata.setCompressionType(producerProperties.getExtension().getCompressionType()); - producerMetadata.setBatchBytes(producerProperties.getExtension().getBufferSize()); - Properties additional = new Properties(); - additional.put(ProducerConfig.ACKS_CONFIG, String.valueOf(this.configurationProperties.getRequiredAcks())); - additional.put(ProducerConfig.LINGER_MS_CONFIG, - String.valueOf(producerProperties.getExtension().getBatchTimeout())); - ProducerFactoryBean producerFB = new ProducerFactoryBean<>(producerMetadata, - this.configurationProperties.getKafkaConnectionString(), additional); - final ProducerConfiguration producerConfiguration = new ProducerConfiguration<>( - producerMetadata, producerFB.getObject()); - producerConfiguration.setProducerListener(this.producerListener); - KafkaProducerContext kafkaProducerContext = new KafkaProducerContext(); - kafkaProducerContext.setProducerConfigurations( - Collections.>singletonMap(destination, producerConfiguration)); - return new ProducerConfigurationMessageHandler(producerConfiguration, destination); + private Map getConsumerConfig(boolean anonymous, String consumerGroup) { + Map props = new HashMap<>(); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString()); + props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); + props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, + anonymous ? "latest" : "earliest"); + props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100); + return props; } - @Override - protected void createProducerDestinationIfNecessary(String name, - ExtendedProducerProperties properties) { - if (this.logger.isInfoEnabled()) { - this.logger.info("Using kafka topic for outbound: " + name); + private boolean isAutoCommitOnError(ExtendedConsumerProperties properties) { + return properties.getExtension().getAutoCommitOnError() != null + ? properties.getExtension().getAutoCommitOnError() + : properties.getExtension().isAutoCommitOffset() && properties.getExtension().isEnableDlq(); + } + + private TopicPartitionInitialOffset[] getTopicPartitionInitialOffsets( + Collection listenedPartitions) { + final TopicPartitionInitialOffset[] topicPartitionInitialOffsets = + new TopicPartitionInitialOffset[listenedPartitions.size()]; + int i = 0; + for (PartitionInfo partition : listenedPartitions) { + + topicPartitionInitialOffsets[i++] = new TopicPartitionInitialOffset(partition.topic(), + partition.partition()); } - validateTopicName(name); - Collection partitions = ensureTopicCreated(name, properties.getPartitionCount()); - // If the topic already exists, and it has a larger number of partitions than the one set in `partitionCount`, - // we will use the existing partition count of the topic instead of the user setting. - if (properties.getPartitionCount() < partitions.size()) { - if (this.logger.isInfoEnabled()) { - this.logger.info("The `partitionCount` setting of the producer for topic " + name + " is " - + properties.getPartitionCount() + ", smaller than the actual partition count of " - + partitions.size() + " of the topic. The larger number will be used instead."); - } - } - this.topicsInUse.put(name, partitions); + return topicPartitionInitialOffsets; } /** * Creates a Kafka topic if needed, or try to increase its partition count to the - * desired number. If a topic with a larger number of partitions already exists, - * the partition count remains unchanged. + * desired number. */ - private Collection ensureTopicCreated(final String topicName, final int partitionCount) { + private Collection ensureTopicCreated(final String topicName, final int partitionCount) { final ZkClient zkClient = new ZkClient(this.configurationProperties.getZkConnectionString(), this.configurationProperties.getZkSessionTimeout(), this.configurationProperties.getZkConnectionTimeout(), ZKStringSerializer$.MODULE$); + + final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); try { final Properties topicConfig = new Properties(); - TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topicName, zkClient); + TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topicName, zkUtils); if (topicMetadata.errorCode() == ErrorMapping.NoError()) { // only consider minPartitionCount for resizing if autoAddPartitions is // true int effectivePartitionCount = this.configurationProperties.isAutoAddPartitions() - ? Math.max(this.configurationProperties.getMinPartitionCount(), - partitionCount) : partitionCount; + ? Math.max(this.configurationProperties.getMinPartitionCount(), partitionCount) + : partitionCount; if (topicMetadata.partitionsMetadata().size() < effectivePartitionCount) { if (this.configurationProperties.isAutoAddPartitions()) { - AdminUtils.addPartitions(zkClient, topicName, effectivePartitionCount, null, false, - new Properties()); + AdminUtils.addPartitions(zkUtils, topicName, effectivePartitionCount, null, false); } else { int topicSize = topicMetadata.partitionsMetadata().size(); @@ -507,7 +442,7 @@ public class KafkaMessageChannelBinder extends } else if (topicMetadata.errorCode() == ErrorMapping.UnknownTopicOrPartitionCode()) { if (this.configurationProperties.isAutoCreateTopics()) { - Seq brokerList = ZkUtils.getSortedBrokerList(zkClient); + Seq brokerList = zkUtils.getSortedBrokerList(); // always consider minPartitionCount for topic creation int effectivePartitionCount = Math.max(this.configurationProperties.getMinPartitionCount(), partitionCount); @@ -518,7 +453,7 @@ public class KafkaMessageChannelBinder extends @Override public Object doWithRetry(RetryContext context) throws RuntimeException { - AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkClient, topicName, + AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkUtils, topicName, replicaAssignment, topicConfig, true); return null; } @@ -533,26 +468,25 @@ public class KafkaMessageChannelBinder extends ErrorMapping.exceptionFor(topicMetadata.errorCode())); } try { - Collection partitions = this.metadataRetryOperations - .execute(new RetryCallback, Exception>() { + return this.metadataRetryOperations + .execute(new RetryCallback, Exception>() { @Override - public Collection doWithRetry(RetryContext context) throws Exception { - KafkaMessageChannelBinder.this.connectionFactory.refreshMetadata( - Collections.singleton(topicName)); - Collection partitions = - KafkaMessageChannelBinder.this.connectionFactory.getPartitions(topicName); + public Collection doWithRetry(RetryContext context) throws Exception { + Collection partitions = + getProducerFactory( + new ExtendedProducerProperties<>(new KafkaProducerProperties())) + .createProducer().partitionsFor(topicName); + // do a sanity check on the partition set if (partitions.size() < partitionCount) { throw new IllegalStateException("The number of expected partitions was: " + partitionCount + ", but " + partitions.size() + (partitions.size() > 1 ? " have " : " has ") + "been found instead"); } - KafkaMessageChannelBinder.this.connectionFactory.getLeaders(partitions); return partitions; } }); - return partitions; } catch (Exception e) { this.logger.error("Cannot initialize Binder", e); @@ -572,19 +506,18 @@ public class KafkaMessageChannelBinder extends if (this.dlqProducer == null) { // we can use the producer defaults as we do not need to tune // performance - ProducerMetadata producerMetadata = new ProducerMetadata<>("dlqKafkaProducer", - byte[].class, byte[].class, BYTE_ARRAY_SERIALIZER, BYTE_ARRAY_SERIALIZER); - producerMetadata.setSync(false); - producerMetadata.setCompressionType(ProducerMetadata.CompressionType.none); - producerMetadata.setBatchBytes(16384); - Properties additionalProps = new Properties(); - additionalProps.put(ProducerConfig.ACKS_CONFIG, - String.valueOf(this.configurationProperties.getRequiredAcks())); - additionalProps.put(ProducerConfig.LINGER_MS_CONFIG, String.valueOf(0)); - ProducerFactoryBean producerFactoryBean = new ProducerFactoryBean<>( - producerMetadata, this.configurationProperties.getKafkaConnectionString(), - additionalProps); - this.dlqProducer = producerFactoryBean.getObject(); + Map props = new HashMap<>(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, + this.configurationProperties.getKafkaConnectionString()); + props.put(ProducerConfig.RETRIES_CONFIG, 0); + props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); + props.put(ProducerConfig.LINGER_MS_CONFIG, 1); + props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); + DefaultKafkaProducerFactory defaultKafkaProducerFactory = + new DefaultKafkaProducerFactory<>(props); + this.dlqProducer = defaultKafkaProducerFactory.createProducer(); } } } @@ -594,29 +527,6 @@ public class KafkaMessageChannelBinder extends } } - private OffsetManager createOffsetManager(String group, long referencePoint) { - try { - - KafkaNativeOffsetManager kafkaOffsetManager = new KafkaNativeOffsetManager(this.connectionFactory, - new ZookeeperConnect(this.configurationProperties.getZkConnectionString()), - Collections.emptyMap()); - kafkaOffsetManager.setConsumerId(group); - kafkaOffsetManager.setReferenceTimestamp(referencePoint); - kafkaOffsetManager.afterPropertiesSet(); - - WindowingOffsetManager windowingOffsetManager = new WindowingOffsetManager(kafkaOffsetManager); - windowingOffsetManager.setTimespan(this.configurationProperties.getOffsetUpdateTimeWindow()); - windowingOffsetManager.setCount(this.configurationProperties.getOffsetUpdateCount()); - windowingOffsetManager.setShutdownTimeout(this.configurationProperties.getOffsetUpdateShutdownTimeout()); - - windowingOffsetManager.afterPropertiesSet(); - return windowingOffsetManager; - } - catch (Exception e) { - throw new RuntimeException(e); - } - } - private String toDisplayString(String original, int maxCharacters) { if (original.length() <= maxCharacters) { return original; @@ -624,45 +534,38 @@ public class KafkaMessageChannelBinder extends return original.substring(0, maxCharacters) + "..."; } - public enum StartOffset { - earliest(OffsetRequest.EarliestTime()), latest(OffsetRequest.LatestTime()); + private final class ProducerConfigurationMessageHandler extends KafkaProducerMessageHandler + implements Lifecycle { - private final long referencePoint; - - StartOffset(long referencePoint) { - this.referencePoint = referencePoint; - } - - public long getReferencePoint() { - return this.referencePoint; - } - } - - private final static class ProducerConfigurationMessageHandler implements MessageHandler, Lifecycle { - - private ProducerConfiguration delegate; - - private String targetTopic; - - private boolean running; - - private ProducerConfigurationMessageHandler( - ProducerConfiguration delegate, String targetTopic) { - Assert.notNull(delegate, "Delegate cannot be null"); - Assert.hasText(targetTopic, "Target topic cannot be null"); - this.delegate = delegate; - this.targetTopic = targetTopic; + private boolean running = true; + private ProducerConfigurationMessageHandler(KafkaTemplate kafkaTemplate, String topic, + ExtendedProducerProperties producerProperties) { + super(kafkaTemplate); + setTopicExpression(new LiteralExpression(topic)); + setBeanFactory(KafkaMessageChannelBinder.this.getBeanFactory()); + if (producerProperties.isPartitioned()) { + SpelExpressionParser parser = new SpelExpressionParser(); + setPartitionIdExpression(parser.parseExpression("headers.partition")); + } + if (producerProperties.getExtension().isSync()) { + setSync(true); + } } @Override public void start() { - this.running = true; + try { + super.onInit(); + } + catch (Exception e) { + this.logger.error("Initialization errors: ", e); + throw new RuntimeException(e); + } } @Override public void stop() { - this.delegate.stop(); this.running = false; } @@ -670,12 +573,5 @@ public class KafkaMessageChannelBinder extends public boolean isRunning() { return this.running; } - - @Override - public void handleMessage(Message message) throws MessagingException { - this.delegate.send(this.targetTopic, - message.getHeaders().get(BinderHeaders.PARTITION_HEADER, Integer.class), null, - (byte[]) message.getPayload()); - } } } diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaProducerProperties.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaProducerProperties.java index 99f2486a1..8d03e10c3 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaProducerProperties.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaProducerProperties.java @@ -18,8 +18,6 @@ package org.springframework.cloud.stream.binder.kafka; import javax.validation.constraints.NotNull; -import org.springframework.integration.kafka.support.ProducerMetadata; - /** * @author Marius Bogoevici */ @@ -27,14 +25,14 @@ public class KafkaProducerProperties { private int bufferSize = 16384; - private ProducerMetadata.CompressionType compressionType = ProducerMetadata.CompressionType.none; + private CompressionType compressionType = CompressionType.none; private boolean sync; private int batchTimeout; public int getBufferSize() { - return bufferSize; + return this.bufferSize; } public void setBufferSize(int bufferSize) { @@ -42,16 +40,16 @@ public class KafkaProducerProperties { } @NotNull - public ProducerMetadata.CompressionType getCompressionType() { - return compressionType; + public CompressionType getCompressionType() { + return this.compressionType; } - public void setCompressionType(ProducerMetadata.CompressionType compressionType) { + public void setCompressionType(CompressionType compressionType) { this.compressionType = compressionType; } public boolean isSync() { - return sync; + return this.sync; } public void setSync(boolean sync) { @@ -59,10 +57,16 @@ public class KafkaProducerProperties { } public int getBatchTimeout() { - return batchTimeout; + return this.batchTimeout; } public void setBatchTimeout(int batchTimeout) { this.batchTimeout = batchTimeout; } + + public enum CompressionType { + none, + gzip, + snappy + } } diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaTopicUtils.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaTopicUtils.java new file mode 100644 index 000000000..b7434ddbe --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaTopicUtils.java @@ -0,0 +1,48 @@ +/* + * 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.cloud.stream.binder.kafka; + +import java.io.UnsupportedEncodingException; + +/** + * @author Soby Chacko + */ +public final class KafkaTopicUtils { + + private KafkaTopicUtils() { + + } + + /** + * Allowed chars are ASCII alphanumerics, '.', '_' and '-'. + */ + public static void validateTopicName(String topicName) { + try { + byte[] utf8 = topicName.getBytes("UTF-8"); + for (byte b : utf8) { + if (!((b >= 'a') && (b <= 'z') || (b >= 'A') && (b <= 'Z') || (b >= '0') && (b <= '9') || (b == '.') + || (b == '-') || (b == '_'))) { + throw new IllegalArgumentException( + "Topic name can only have ASCII alphanumerics, '.', '_' and '-'"); + } + } + } + catch (UnsupportedEncodingException e) { + throw new AssertionError(e); // Can't happen + } + } +} diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/WindowingOffsetManager.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/WindowingOffsetManager.java deleted file mode 100644 index 44fa9792d..000000000 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/WindowingOffsetManager.java +++ /dev/null @@ -1,262 +0,0 @@ -/* - * Copyright 2015 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.cloud.stream.binder.kafka; - -import java.io.IOException; -import java.util.Collection; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; - -import rx.Observable; -import rx.Subscription; -import rx.functions.Action0; -import rx.functions.Action1; -import rx.functions.Func1; -import rx.functions.Func2; -import rx.observables.GroupedObservable; -import rx.observables.MathObservable; -import rx.subjects.PublishSubject; -import rx.subjects.SerializedSubject; -import rx.subjects.Subject; - -import org.springframework.beans.factory.DisposableBean; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.integration.kafka.core.Partition; -import org.springframework.integration.kafka.listener.OffsetManager; -import org.springframework.util.Assert; - -/** - * An {@link OffsetManager} that aggregates writes over a time or count window, using an underlying delegate to - * do the actual operations. Its purpose is to reduce the performance impact of writing operations - * wherever this is desirable. - * - * Either a time window or a number of writes can be specified, but not both. - * - * @author Marius Bogoevici - */ -public class WindowingOffsetManager implements OffsetManager, InitializingBean, DisposableBean { - - private final CreatePartitionAndOffsetFunction createPartitionAndOffsetFunction = new CreatePartitionAndOffsetFunction(); - - private final GetOffsetFunction getOffsetFunction = new GetOffsetFunction(); - - private final ComputeMaximumOffsetByPartitionFunction findHighestOffsetInPartitionGroup = new ComputeMaximumOffsetByPartitionFunction(); - - private final GetPartitionFunction getPartition = new GetPartitionFunction(); - - private final FindHighestOffsetsByPartitionFunction findHighestOffsetsByPartition = new FindHighestOffsetsByPartitionFunction(); - - private final DelegateUpdateOffsetAction delegateUpdateOffsetAction = new DelegateUpdateOffsetAction(); - - private final NotifyObservableClosedAction notifyObservableClosed = new NotifyObservableClosedAction(); - - private final OffsetManager delegate; - - private long timespan = 10 * 1000; - - private int count; - - private Subject offsets; - - private Subscription subscription; - - private int shutdownTimeout = 2000; - - private CountDownLatch shutdownLatch; - - public WindowingOffsetManager(OffsetManager offsetManager) { - this.delegate = offsetManager; - } - - /** - * The timespan for aggregating write operations, before invoking the underlying {@link OffsetManager}. - * - * @param timespan duration in milliseconds - */ - public void setTimespan(long timespan) { - Assert.isTrue(timespan >= 0, "Timespan must be a positive value"); - this.timespan = timespan; - } - - /** - * How many writes should be aggregated, before invoking the underlying {@link OffsetManager}. Setting this value - * to 1 effectively disables windowing. - * - * @param count number of writes - */ - public void setCount(int count) { - Assert.isTrue(count >= 0, "Count must be a positive value"); - this.count = count; - } - - /** - * The timeout that {@link #close()} and {@link #destroy()} operations will wait for receving a confirmation that the - * underlying writes have been processed. - * - * @param shutdownTimeout duration in milliseconds - */ - public void setShutdownTimeout(int shutdownTimeout) { - this.shutdownTimeout = shutdownTimeout; - } - - @Override - public void afterPropertiesSet() throws Exception { - Assert.isTrue(timespan > 0 ^ count > 0, "Only one of the timespan or count must be set"); - // create the stream if windowing is set, and count is higher than 1 - if (timespan > 0 || count > 1) { - offsets = new SerializedSubject<>(PublishSubject.create()); - // window by either count or time - Observable> window = - timespan > 0 ? offsets.window(timespan, TimeUnit.MILLISECONDS) : offsets.window(count); - Observable maximumOffsetsByWindow = window - .flatMap(findHighestOffsetsByPartition) - .doOnCompleted(notifyObservableClosed); - subscription = maximumOffsetsByWindow.subscribe(delegateUpdateOffsetAction); - } - else { - offsets = null; - } - } - - @Override - public void destroy() throws Exception { - this.flush(); - this.close(); - if (delegate instanceof DisposableBean) { - ((DisposableBean) delegate).destroy(); - } - } - - @Override - public void updateOffset(Partition partition, long offset) { - if (offsets != null) { - offsets.onNext(new PartitionAndOffset(partition, offset)); - } - else { - delegate.updateOffset(partition, offset); - } - } - - @Override - public long getOffset(Partition partition) { - return delegate.getOffset(partition); - } - - @Override - public void deleteOffset(Partition partition) { - delegate.deleteOffset(partition); - } - - @Override - public void resetOffsets(Collection partition) { - delegate.resetOffsets(partition); - } - - @Override - public void close() throws IOException { - if (offsets != null) { - shutdownLatch = new CountDownLatch(1); - offsets.onCompleted(); - try { - shutdownLatch.await(shutdownTimeout, TimeUnit.MILLISECONDS); - } - catch (InterruptedException e) { - // ignore - } - subscription.unsubscribe(); - } - delegate.close(); - } - - @Override - public void flush() throws IOException { - delegate.flush(); - } - - private final class PartitionAndOffset { - - private final Partition partition; - - private final Long offset; - - private PartitionAndOffset(Partition partition, Long offset) { - this.partition = partition; - this.offset = offset; - } - - public Partition getPartition() { - return partition; - } - - public Long getOffset() { - return offset; - } - } - - private class DelegateUpdateOffsetAction implements Action1 { - @Override - public void call(PartitionAndOffset partitionAndOffset) { - delegate.updateOffset(partitionAndOffset.getPartition(), partitionAndOffset.getOffset()); - } - } - - private class NotifyObservableClosedAction implements Action0 { - @Override - public void call() { - if (shutdownLatch != null) { - shutdownLatch.countDown(); - } - } - } - - private class CreatePartitionAndOffsetFunction implements Func2 { - @Override - public PartitionAndOffset call(Partition partition, Long offset) { - return new PartitionAndOffset(partition, offset); - } - } - - private class GetOffsetFunction implements Func1 { - @Override - public Long call(PartitionAndOffset partitionAndOffset) { - return partitionAndOffset.getOffset(); - } - } - - private class ComputeMaximumOffsetByPartitionFunction implements Func1, Observable> { - @Override - public Observable call(GroupedObservable group) { - return Observable.zip(Observable.just(group.getKey()), - MathObservable.max(group.map(getOffsetFunction)), - createPartitionAndOffsetFunction); - } - } - - private class GetPartitionFunction implements Func1 { - @Override - public Partition call(PartitionAndOffset partitionAndOffset) { - return partitionAndOffset.getPartition(); - } - } - - private class FindHighestOffsetsByPartitionFunction implements Func1, Observable> { - @Override - public Observable call(Observable windowBuffer) { - return windowBuffer.groupBy(getPartition).flatMap(findHighestOffsetInPartitionGroup); - } - } -} diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java index 623bbf35d..82c523074 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java @@ -21,7 +21,6 @@ import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfigurati import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.binder.Binder; -import org.springframework.cloud.stream.binder.kafka.KafkaBinderHealthIndicator; import org.springframework.cloud.stream.binder.kafka.KafkaExtendedBindingProperties; import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder; import org.springframework.cloud.stream.config.codec.kryo.KryoCodecAutoConfiguration; @@ -29,8 +28,8 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; import org.springframework.integration.codec.Codec; -import org.springframework.integration.kafka.support.LoggingProducerListener; -import org.springframework.integration.kafka.support.ProducerListener; +import org.springframework.kafka.support.LoggingProducerListener; +import org.springframework.kafka.support.ProducerListener; /** * @author David Turanski @@ -41,8 +40,8 @@ import org.springframework.integration.kafka.support.ProducerListener; */ @Configuration @ConditionalOnMissingBean(Binder.class) -@Import({ KryoCodecAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class }) -@EnableConfigurationProperties({ KafkaBinderConfigurationProperties.class, KafkaExtendedBindingProperties.class }) +@Import({KryoCodecAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class}) +@EnableConfigurationProperties({KafkaBinderConfigurationProperties.class, KafkaExtendedBindingProperties.class}) public class KafkaBinderConfiguration { @Autowired @@ -59,10 +58,11 @@ public class KafkaBinderConfiguration { @Bean KafkaMessageChannelBinder kafkaMessageChannelBinder() { - KafkaMessageChannelBinder kafkaMessageChannelBinder = new KafkaMessageChannelBinder(configurationProperties); - kafkaMessageChannelBinder.setCodec(codec); - kafkaMessageChannelBinder.setProducerListener(producerListener); - kafkaMessageChannelBinder.setExtendedBindingProperties(kafkaExtendedBindingProperties); + KafkaMessageChannelBinder kafkaMessageChannelBinder = new KafkaMessageChannelBinder( + this.configurationProperties); + kafkaMessageChannelBinder.setCodec(this.codec); + //kafkaMessageChannelBinder.setProducerListener(producerListener); + kafkaMessageChannelBinder.setExtendedBindingProperties(this.kafkaExtendedBindingProperties); return kafkaMessageChannelBinder; } @@ -72,8 +72,8 @@ public class KafkaBinderConfiguration { return new LoggingProducerListener(); } - @Bean - KafkaBinderHealthIndicator healthIndicator(KafkaMessageChannelBinder kafkaMessageChannelBinder) { - return new KafkaBinderHealthIndicator(kafkaMessageChannelBinder, configurationProperties); - } +// @Bean +// KafkaBinderHealthIndicator healthIndicator(KafkaMessageChannelBinder kafkaMessageChannelBinder) { +// return new KafkaBinderHealthIndicator(kafkaMessageChannelBinder, configurationProperties); +// } } diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java index 98ee963f2..9a8135b03 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java @@ -23,15 +23,16 @@ import org.springframework.util.StringUtils; * @author David Turanski * @author Ilayaperumal Gopinathan * @author Marius Bogoevici + * @author Soby Chacko */ @ConfigurationProperties(prefix = "spring.cloud.stream.kafka.binder") public class KafkaBinderConfigurationProperties { - private String[] zkNodes = new String[] { "localhost" }; + private String[] zkNodes = new String[] {"localhost"}; private String defaultZkPort = "2181"; - private String[] brokers = new String[] { "localhost" }; + private String[] brokers = new String[] {"localhost"}; private String defaultBrokerPort = "9092"; @@ -71,6 +72,12 @@ public class KafkaBinderConfigurationProperties { private int queueSize = 8192; + private String consumerGroup; + + public String getConsumerGroup() { + return this.consumerGroup; + } + public String getZkConnectionString() { return toConnectionString(this.zkNodes, this.defaultZkPort); } @@ -80,7 +87,7 @@ public class KafkaBinderConfigurationProperties { } public String[] getHeaders() { - return headers; + return this.headers; } public int getOffsetUpdateTimeWindow() { @@ -96,7 +103,7 @@ public class KafkaBinderConfigurationProperties { } public String[] getZkNodes() { - return zkNodes; + return this.zkNodes; } public void setZkNodes(String... zkNodes) { @@ -108,7 +115,7 @@ public class KafkaBinderConfigurationProperties { } public String[] getBrokers() { - return brokers; + return this.brokers; } public void setBrokers(String... brokers) { @@ -170,7 +177,7 @@ public class KafkaBinderConfigurationProperties { } public int getMaxWait() { - return maxWait; + return this.maxWait; } public void setMaxWait(int maxWait) { @@ -178,7 +185,7 @@ public class KafkaBinderConfigurationProperties { } public int getRequiredAcks() { - return requiredAcks; + return this.requiredAcks; } public void setRequiredAcks(int requiredAcks) { @@ -186,7 +193,7 @@ public class KafkaBinderConfigurationProperties { } public int getReplicationFactor() { - return replicationFactor; + return this.replicationFactor; } public void setReplicationFactor(int replicationFactor) { @@ -194,7 +201,7 @@ public class KafkaBinderConfigurationProperties { } public int getFetchSize() { - return fetchSize; + return this.fetchSize; } public void setFetchSize(int fetchSize) { @@ -202,7 +209,7 @@ public class KafkaBinderConfigurationProperties { } public int getMinPartitionCount() { - return minPartitionCount; + return this.minPartitionCount; } public void setMinPartitionCount(int minPartitionCount) { @@ -210,7 +217,7 @@ public class KafkaBinderConfigurationProperties { } public int getQueueSize() { - return queueSize; + return this.queueSize; } public void setQueueSize(int queueSize) { @@ -218,7 +225,7 @@ public class KafkaBinderConfigurationProperties { } public boolean isAutoCreateTopics() { - return autoCreateTopics; + return this.autoCreateTopics; } public void setAutoCreateTopics(boolean autoCreateTopics) { @@ -226,7 +233,7 @@ public class KafkaBinderConfigurationProperties { } public boolean isAutoAddPartitions() { - return autoAddPartitions; + return this.autoAddPartitions; } public void setAutoAddPartitions(boolean autoAddPartitions) { @@ -234,10 +241,15 @@ public class KafkaBinderConfigurationProperties { } public int getSocketBufferSize() { - return socketBufferSize; + return this.socketBufferSize; } public void setSocketBufferSize(int socketBufferSize) { this.socketBufferSize = socketBufferSize; } + + public void setConsumerGroup(String consumerGroup) { + this.consumerGroup = consumerGroup; + } + } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 5730511c2..3c90f376c 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -16,38 +16,55 @@ package org.springframework.cloud.stream.binder.kafka; +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.Assert.fail; + +import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.HashMap; import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import java.util.Properties; import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import kafka.admin.AdminUtils; -import kafka.api.TopicMetadata; +import org.I0Itec.zkclient.ZkClient; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.common.PartitionInfo; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.kafka.common.serialization.Deserializer; +import org.assertj.core.api.Condition; import org.junit.Before; import org.junit.ClassRule; +import org.junit.Ignore; import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; +import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderException; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.PartitionCapableBinderTests; +import org.springframework.cloud.stream.binder.PartitionTestSupport; import org.springframework.cloud.stream.binder.Spy; import org.springframework.cloud.stream.binder.TestUtils; import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties; -import org.springframework.cloud.stream.binder.test.junit.kafka.KafkaTestSupport; +import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.context.support.GenericApplicationContext; +import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.kafka.core.Partition; -import org.springframework.integration.kafka.core.TopicNotFoundException; -import org.springframework.integration.kafka.support.KafkaHeaders; -import org.springframework.integration.kafka.support.ProducerConfiguration; -import org.springframework.integration.kafka.support.ProducerMetadata; +import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.support.KafkaHeaders; +import org.springframework.kafka.support.TopicPartitionInitialOffset; +import org.springframework.kafka.test.core.BrokerAddress; +import org.springframework.kafka.test.rule.KafkaEmbedded; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -58,8 +75,10 @@ import org.springframework.retry.backoff.FixedBackOffPolicy; import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; -import static org.assertj.core.api.Assertions.assertThat; -import static org.junit.Assert.fail; +import kafka.admin.AdminUtils; +import kafka.api.TopicMetadata; +import kafka.utils.ZKStringSerializer$; +import kafka.utils.ZkUtils; /** * Integration tests for the {@link KafkaMessageChannelBinder}. @@ -68,13 +87,15 @@ import static org.junit.Assert.fail; * @author Mark Fisher * @author Ilayaperumal Gopinathan */ -public class KafkaBinderTests extends - PartitionCapableBinderTests, ExtendedProducerProperties> { +public class KafkaBinderTests + extends + PartitionCapableBinderTests, + ExtendedProducerProperties> { private final String CLASS_UNDER_TEST_NAME = KafkaMessageChannelBinder.class.getSimpleName(); @ClassRule - public static KafkaTestSupport kafkaTestSupport = new KafkaTestSupport(); + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 10); private KafkaTestBinder binder; @@ -92,10 +113,16 @@ public class KafkaBinderTests extends return binder; } - private KafkaBinderConfigurationProperties createConfigurationProperties() { + protected KafkaBinderConfigurationProperties createConfigurationProperties() { KafkaBinderConfigurationProperties binderConfiguration = new KafkaBinderConfigurationProperties(); - binderConfiguration.setBrokers(kafkaTestSupport.getBrokerAddress()); - binderConfiguration.setZkNodes(kafkaTestSupport.getZkConnectString()); + BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses(); + List bAddresses = new ArrayList<>(); + for (BrokerAddress bAddress : brokerAddresses) { + bAddresses.add(bAddress.toString()); + } + String[] foo = new String[bAddresses.size()]; + binderConfiguration.setBrokers(bAddresses.toArray(foo)); + binderConfiguration.setZkNodes(embeddedKafka.getZookeeperConnectionString()); return binderConfiguration; } @@ -132,23 +159,38 @@ public class KafkaBinderTests extends throw new UnsupportedOperationException("'spyOn' is not used by Kafka tests"); } + + private ConsumerFactory consumerFactory() { + Map props = new HashMap<>(); + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, configurationProperties.getKafkaConnectionString()); + props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); + props.put(ConsumerConfig.GROUP_ID_CONFIG, configurationProperties.getConsumerGroup()); + Deserializer valueDecoder = new ByteArrayDeserializer(); + Deserializer keyDecoder = new ByteArrayDeserializer(); + + return new DefaultKafkaConsumerFactory<>(props, keyDecoder, valueDecoder); + + } + @Test public void testDlqAndRetry() throws Exception { KafkaTestBinder binder = getBinder(); - ExtendedProducerProperties producerProperties = createProducerProperties(); - producerProperties.setPartitionCount(10); - DirectChannel moduleOutputChannel = createBindableChannel("output", - createProducerBindingProperties(producerProperties)); + DirectChannel moduleOutputChannel = new DirectChannel(); + DirectChannel moduleInputChannel = new DirectChannel(); QueueChannel dlqChannel = new QueueChannel(); FailingInvocationCountingMessageHandler handler = new FailingInvocationCountingMessageHandler(); + moduleInputChannel.subscribe(handler); + ExtendedProducerProperties producerProperties = createProducerProperties(); + producerProperties.setPartitionCount(2); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); consumerProperties.setMaxAttempts(3); consumerProperties.setBackOffInitialInterval(100); consumerProperties.setBackOffMaxInterval(150); consumerProperties.getExtension().setEnableDlq(true); - DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); - moduleInputChannel.subscribe(handler); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); long uniqueBindingId = System.currentTimeMillis(); + Binding producerBinding = binder.bindProducer("retryTest." + uniqueBindingId + ".0", moduleOutputChannel, producerProperties); Binding consumerBinding = binder.bindConsumer("retryTest." + uniqueBindingId + ".0", @@ -159,7 +201,7 @@ public class KafkaBinderTests extends Binding dlqConsumerBinding = binder.bindConsumer( "error.retryTest." + uniqueBindingId + ".0.testGroup", null, dlqChannel, dlqConsumerProperties); - + binderBindUnbindLatency(); String testMessagePayload = "test." + UUID.randomUUID().toString(); Message testMessage = MessageBuilder.withPayload(testMessagePayload).build(); moduleOutputChannel.send(testMessage); @@ -168,6 +210,7 @@ public class KafkaBinderTests extends assertThat(receivedMessage).isNotNull(); assertThat(receivedMessage.getPayload()).isEqualTo(testMessagePayload); assertThat(handler.getInvocationCount()).isEqualTo(consumerProperties.getMaxAttempts()); + binderBindUnbindLatency(); dlqConsumerBinding.unbind(); consumerBinding.unbind(); producerBinding.unbind(); @@ -176,16 +219,17 @@ public class KafkaBinderTests extends @Test public void testDefaultAutoCommitOnErrorWithoutDlq() throws Exception { KafkaTestBinder binder = getBinder(); + DirectChannel moduleOutputChannel = new DirectChannel(); + DirectChannel moduleInputChannel = new DirectChannel(); + FailingInvocationCountingMessageHandler handler = new FailingInvocationCountingMessageHandler(); + moduleInputChannel.subscribe(handler); ExtendedProducerProperties producerProperties = createProducerProperties(); producerProperties.setPartitionCount(10); - DirectChannel moduleOutputChannel = createBindableChannel("output", createProducerBindingProperties(producerProperties)); - FailingInvocationCountingMessageHandler handler = new FailingInvocationCountingMessageHandler(); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); consumerProperties.setMaxAttempts(1); consumerProperties.setBackOffInitialInterval(100); consumerProperties.setBackOffMaxInterval(150); - DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); - moduleInputChannel.subscribe(handler); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); long uniqueBindingId = System.currentTimeMillis(); Binding producerBinding = binder.bindProducer("retryTest." + uniqueBindingId + ".0", moduleOutputChannel, producerProperties); @@ -209,6 +253,7 @@ public class KafkaBinderTests extends QueueChannel successfulInputChannel = new QueueChannel(); consumerBinding = binder.bindConsumer("retryTest." + uniqueBindingId + ".0", "testGroup", successfulInputChannel, consumerProperties); + binderBindUnbindLatency(); String testMessage2Payload = "test." + UUID.randomUUID().toString(); Message testMessage2 = MessageBuilder.withPayload(testMessage2Payload).build(); moduleOutputChannel.send(testMessage2); @@ -224,17 +269,18 @@ public class KafkaBinderTests extends @Test public void testDefaultAutoCommitOnErrorWithDlq() throws Exception { KafkaTestBinder binder = getBinder(); + DirectChannel moduleOutputChannel = new DirectChannel(); + DirectChannel moduleInputChannel = new DirectChannel(); FailingInvocationCountingMessageHandler handler = new FailingInvocationCountingMessageHandler(); + moduleInputChannel.subscribe(handler); ExtendedProducerProperties producerProperties = createProducerProperties(); producerProperties.setPartitionCount(10); - DirectChannel moduleOutputChannel = createBindableChannel("output", createProducerBindingProperties(producerProperties)); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); consumerProperties.setMaxAttempts(3); consumerProperties.setBackOffInitialInterval(100); consumerProperties.setBackOffMaxInterval(150); consumerProperties.getExtension().setEnableDlq(true); - DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); - moduleInputChannel.subscribe(handler); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); long uniqueBindingId = System.currentTimeMillis(); Binding producerBinding = binder.bindProducer("retryTest." + uniqueBindingId + ".0", moduleOutputChannel, producerProperties); @@ -260,7 +306,7 @@ public class KafkaBinderTests extends assertThat(handledMessage).isNotNull(); assertThat(handledMessage.getPayload()).isEqualTo(testMessagePayload); assertThat(handler.getInvocationCount()).isEqualTo(consumerProperties.getMaxAttempts()); - + binderBindUnbindLatency(); dlqConsumerBinding.unbind(); consumerBinding.unbind(); @@ -275,34 +321,36 @@ public class KafkaBinderTests extends Message receivedMessage = receive(successfulInputChannel); assertThat(receivedMessage.getPayload()).isEqualTo(testMessage2Payload); + binderBindUnbindLatency(); consumerBinding.unbind(); producerBinding.unbind(); } @Test(expected = IllegalArgumentException.class) public void testValidateKafkaTopicName() { - KafkaMessageChannelBinder.validateTopicName("foo:bar"); + KafkaTopicUtils.validateTopicName("foo:bar"); } @Test public void testCompression() throws Exception { - final ProducerMetadata.CompressionType[] codecs = new ProducerMetadata.CompressionType[] { - ProducerMetadata.CompressionType.none, ProducerMetadata.CompressionType.gzip, - ProducerMetadata.CompressionType.snappy }; + final KafkaProducerProperties.CompressionType[] codecs = new KafkaProducerProperties.CompressionType[] { + KafkaProducerProperties.CompressionType.none, KafkaProducerProperties.CompressionType.gzip, + KafkaProducerProperties.CompressionType.snappy}; byte[] testPayload = new byte[2048]; Arrays.fill(testPayload, (byte) 65); KafkaTestBinder binder = getBinder(); - for (ProducerMetadata.CompressionType codec : codecs) { - ExtendedProducerProperties producerProperties = createProducerProperties(); - producerProperties.getExtension().setCompressionType(codec); - - DirectChannel moduleOutputChannel = createBindableChannel("output", - createProducerBindingProperties(producerProperties)); + for (KafkaProducerProperties.CompressionType codec : codecs) { + DirectChannel moduleOutputChannel = new DirectChannel(); QueueChannel moduleInputChannel = new QueueChannel(); + ExtendedProducerProperties producerProperties = createProducerProperties(); + producerProperties.getExtension().setCompressionType( + KafkaProducerProperties.CompressionType.valueOf(codec.toString())); Binding producerBinding = binder.bindProducer("foo.0", moduleOutputChannel, producerProperties); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); Binding consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel, - createConsumerProperties()); + consumerProperties); Message message = org.springframework.integration.support.MessageBuilder.withPayload(testPayload) .build(); // Let the consumer actually bind to the producer before sending a msg @@ -324,11 +372,13 @@ public class KafkaBinderTests extends KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); binderConfiguration.setMinPartitionCount(10); KafkaTestBinder binder = new KafkaTestBinder(binderConfiguration); - + QueueChannel moduleInputChannel = new QueueChannel(); ExtendedProducerProperties producerProperties = createProducerProperties(); producerProperties.setPartitionCount(10); - DirectChannel moduleOutputChannel = createBindableChannel("output", createProducerBindingProperties(producerProperties)); - QueueChannel moduleInputChannel = new QueueChannel(); + + DirectChannel moduleOutputChannel = createBindableChannel("output", + createProducerBindingProperties(producerProperties)); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); long uniqueBindingId = System.currentTimeMillis(); Binding producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0", @@ -343,8 +393,11 @@ public class KafkaBinderTests extends Message inbound = receive(moduleInputChannel); assertThat(inbound).isNotNull(); assertThat((byte[]) inbound.getPayload()).containsExactly(testPayload); - Collection partitions = binder.getCoreBinder().getConnectionFactory() - .getPartitions("foo" + uniqueBindingId + ".0"); + + + Collection partitions = + consumerFactory().createConsumer().partitionsFor("foo" + uniqueBindingId + ".0"); + assertThat(partitions).hasSize(10); producerBinding.unbind(); consumerBinding.unbind(); @@ -358,17 +411,19 @@ public class KafkaBinderTests extends KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); binderConfiguration.setMinPartitionCount(6); KafkaTestBinder binder = new KafkaTestBinder(binderConfiguration); - DirectChannel moduleOutputChannel = new DirectChannel(); QueueChannel moduleInputChannel = new QueueChannel(); ExtendedProducerProperties producerProperties = createProducerProperties(); producerProperties.setPartitionCount(5); producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload")); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); long uniqueBindingId = System.currentTimeMillis(); + DirectChannel moduleOutputChannel = createBindableChannel("output", + createProducerBindingProperties(producerProperties)); Binding producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, producerProperties); Binding consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null, moduleInputChannel, consumerProperties); + Thread.sleep(1000); Message message = org.springframework.integration.support.MessageBuilder.withPayload(testPayload) .build(); // Let the consumer actually bind to the producer before sending a msg @@ -377,8 +432,9 @@ public class KafkaBinderTests extends Message inbound = receive(moduleInputChannel); assertThat(inbound).isNotNull(); assertThat((byte[]) inbound.getPayload()).containsExactly(testPayload); - Collection partitions = binder.getCoreBinder().getConnectionFactory() - .getPartitions("foo" + uniqueBindingId + ".0"); + Collection partitions = + consumerFactory().createConsumer().partitionsFor("foo" + uniqueBindingId + ".0"); + assertThat(partitions).hasSize(6); producerBinding.unbind(); consumerBinding.unbind(); @@ -393,11 +449,12 @@ public class KafkaBinderTests extends binderConfiguration.setMinPartitionCount(4); KafkaTestBinder binder = new KafkaTestBinder(binderConfiguration); - DirectChannel moduleOutputChannel = new DirectChannel(); QueueChannel moduleInputChannel = new QueueChannel(); ExtendedProducerProperties producerProperties = createProducerProperties(); producerProperties.setPartitionCount(5); producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload")); + DirectChannel moduleOutputChannel = createBindableChannel("output", + createProducerBindingProperties(producerProperties)); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); long uniqueBindingId = System.currentTimeMillis(); Binding producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0", @@ -412,8 +469,8 @@ public class KafkaBinderTests extends Message inbound = receive(moduleInputChannel); assertThat(inbound).isNotNull(); assertThat((byte[]) inbound.getPayload()).containsExactly(testPayload); - Collection partitions = binder.getCoreBinder().getConnectionFactory() - .getPartitions("foo" + uniqueBindingId + ".0"); + Collection partitions = + consumerFactory().createConsumer().partitionsFor("foo" + uniqueBindingId + ".0"); assertThat(partitions).hasSize(5); producerBinding.unbind(); consumerBinding.unbind(); @@ -431,10 +488,14 @@ public class KafkaBinderTests extends QueueChannel input1 = new QueueChannel(); String testTopicName = UUID.randomUUID().toString(); - binder.bindProducer(testTopicName, output, createProducerProperties()); + Binding producerBinding = binder.bindProducer(testTopicName, output, + createProducerProperties()); String testPayload1 = "foo-" + UUID.randomUUID().toString(); output.send(new GenericMessage<>(testPayload1.getBytes())); - binder.bindConsumer(testTopicName, "startOffsets", input1, createConsumerProperties()); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); + Binding consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, + consumerProperties); Message receivedMessage1 = (Message) receive(input1); assertThat(receivedMessage1).isNotNull(); assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload1); @@ -443,32 +504,53 @@ public class KafkaBinderTests extends Message receivedMessage2 = (Message) receive(input1); assertThat(receivedMessage2).isNotNull(); assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload2); + + producerBinding.unbind(); + consumerBinding.unbind(); } @Test @SuppressWarnings("unchecked") public void testEarliest() throws Exception { - KafkaTestBinder binder = getBinder(); - DirectChannel output = new DirectChannel(); - QueueChannel input1 = new QueueChannel(); + Binding producerBinding = null; + Binding consumerBinding = null; - String testTopicName = UUID.randomUUID().toString(); - binder.bindProducer(testTopicName, output, createProducerProperties()); - String testPayload1 = "foo-" + UUID.randomUUID().toString(); - output.send(new GenericMessage<>(testPayload1.getBytes())); - ExtendedConsumerProperties properties = createConsumerProperties(); - properties.getExtension().setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest); - binder.bindConsumer(testTopicName, "startOffsets", input1, properties); - Message receivedMessage1 = (Message) receive(input1); - assertThat(receivedMessage1).isNotNull(); - String testPayload2 = "foo-" + UUID.randomUUID().toString(); - output.send(new GenericMessage<>(testPayload2.getBytes())); - Message receivedMessage2 = (Message) receive(input1); - assertThat(receivedMessage2).isNotNull(); - assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload2); + try { + KafkaTestBinder binder = getBinder(); + DirectChannel output = new DirectChannel(); + QueueChannel input1 = new QueueChannel(); + + String testTopicName = UUID.randomUUID().toString(); + producerBinding = binder.bindProducer(testTopicName, output, createProducerProperties()); + String testPayload1 = "foo-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload1.getBytes())); + ExtendedConsumerProperties properties = createConsumerProperties(); + properties.getExtension().setAutoRebalanceEnabled(false); + properties.getExtension().setStartOffset(KafkaConsumerProperties.StartOffset.earliest); + consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, properties); + Message receivedMessage1 = (Message) receive(input1); + assertThat(receivedMessage1).isNotNull(); + String testPayload2 = "foo-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload2.getBytes())); + Message receivedMessage2 = (Message) receive(input1); + assertThat(receivedMessage2).isNotNull(); + assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload2); + Thread.sleep(2000); + producerBinding.unbind(); + consumerBinding.unbind(); + } + finally { + if (consumerBinding != null) { + consumerBinding.unbind(); + } + if (producerBinding != null) { + producerBinding.unbind(); + } + } } @Test + @Ignore("Needs further discussion") @SuppressWarnings("unchecked") public void testReset() throws Exception { KafkaTestBinder binder = getBinder(); @@ -479,28 +561,29 @@ public class KafkaBinderTests extends Binding producerBinding = binder.bindProducer(testTopicName, output, createProducerProperties()); - String testPayload1 = "foo-" + UUID.randomUUID().toString(); + String testPayload1 = "foo1-" + UUID.randomUUID().toString(); output.send(new GenericMessage<>(testPayload1.getBytes())); ExtendedConsumerProperties properties = createConsumerProperties(); properties.getExtension().setResetOffsets(true); - properties.getExtension().setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest); + properties.getExtension().setStartOffset(KafkaConsumerProperties.StartOffset.earliest); Binding consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, properties); Message receivedMessage1 = (Message) receive(input1); assertThat(receivedMessage1).isNotNull(); - String testPayload2 = "foo-" + UUID.randomUUID().toString(); + assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload1); + String testPayload2 = "foo2-" + UUID.randomUUID().toString(); output.send(new GenericMessage<>(testPayload2.getBytes())); Message receivedMessage2 = (Message) receive(input1); assertThat(receivedMessage2).isNotNull(); assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload2); consumerBinding.unbind(); - String testPayload3 = "foo-" + UUID.randomUUID().toString(); + String testPayload3 = "foo3-" + UUID.randomUUID().toString(); output.send(new GenericMessage<>(testPayload3.getBytes())); ExtendedConsumerProperties properties2 = createConsumerProperties(); properties2.getExtension().setResetOffsets(true); - properties2.getExtension().setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest); + properties2.getExtension().setStartOffset(KafkaConsumerProperties.StartOffset.earliest); consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, properties2); Message receivedMessage4 = (Message) receive(input1); assertThat(receivedMessage4).isNotNull(); @@ -518,41 +601,57 @@ public class KafkaBinderTests extends @Test @SuppressWarnings("unchecked") public void testResume() throws Exception { - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties); - GenericApplicationContext context = new GenericApplicationContext(); - context.refresh(); - binder.setApplicationContext(context); - binder.afterPropertiesSet(); - DirectChannel output = new DirectChannel(); - QueueChannel input1 = new QueueChannel(); + Binding producerBinding = null; + Binding consumerBinding = null; - String testTopicName = UUID.randomUUID().toString(); - Binding producerBinding = binder.bindProducer(testTopicName, output, - createProducerProperties()); - String testPayload1 = "foo-" + UUID.randomUUID().toString(); - output.send(new GenericMessage<>(testPayload1.getBytes())); - ExtendedConsumerProperties firstConsumerProperties = createConsumerProperties(); - Binding consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, - firstConsumerProperties); - Message receivedMessage1 = (Message) receive(input1); - assertThat(receivedMessage1).isNotNull(); - String testPayload2 = "foo-" + UUID.randomUUID().toString(); - output.send(new GenericMessage<>(testPayload2.getBytes())); - Message receivedMessage2 = (Message) receive(input1); - assertThat(receivedMessage2).isNotNull(); - assertThat(new String(receivedMessage2.getPayload())).isNotNull(); - consumerBinding.unbind(); + try { + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties); + GenericApplicationContext context = new GenericApplicationContext(); + context.refresh(); + binder.setApplicationContext(context); + binder.afterPropertiesSet(); + DirectChannel output = new DirectChannel(); + QueueChannel input1 = new QueueChannel(); - String testPayload3 = "foo-" + UUID.randomUUID().toString(); - output.send(new GenericMessage<>(testPayload3.getBytes())); + String testTopicName = UUID.randomUUID().toString(); + producerBinding = binder.bindProducer(testTopicName, output, + createProducerProperties()); + String testPayload1 = "foo1-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload1.getBytes())); + ExtendedConsumerProperties firstConsumerProperties = createConsumerProperties(); + firstConsumerProperties.getExtension().setAutoRebalanceEnabled(false); + consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, + firstConsumerProperties); + Message receivedMessage1 = (Message) receive(input1); + assertThat(receivedMessage1).isNotNull(); + String testPayload2 = "foo2-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload2.getBytes())); + Message receivedMessage2 = (Message) receive(input1); + assertThat(receivedMessage2).isNotNull(); + assertThat(new String(receivedMessage2.getPayload())).isNotNull(); + consumerBinding.unbind(); - consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, createConsumerProperties()); - Message receivedMessage3 = (Message) receive(input1); - assertThat(receivedMessage3).isNotNull(); - assertThat(new String(receivedMessage3.getPayload())).isEqualTo(testPayload3); - consumerBinding.unbind(); - producerBinding.unbind(); + String testPayload3 = "foo3-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload3.getBytes())); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, consumerProperties); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); + Message receivedMessage3 = (Message) receive(input1); + assertThat(receivedMessage3).isNotNull(); + assertThat(new String(receivedMessage3.getPayload())).isEqualTo(testPayload3); + + Thread.sleep(2000); + } + finally { + if (consumerBinding != null) { + consumerBinding.unbind(); + } + if (producerBinding != null) { + producerBinding.unbind(); + } + } } @Test @@ -568,9 +667,8 @@ public class KafkaBinderTests extends properties.getExtension().setSync(true); Binding producerBinding = binder.bindProducer(testTopicName, output, properties); DirectFieldAccessor accessor = new DirectFieldAccessor(extractEndpoint(producerBinding)); - ProducerConfiguration producerConfiguration = (ProducerConfiguration) accessor - .getPropertyValue("delegate"); - assertThat(producerConfiguration.getProducerMetadata().isSync()) + KafkaProducerMessageHandler wrappedInstance = (KafkaProducerMessageHandler) accessor.getWrappedInstance(); + assertThat(new DirectFieldAccessor(wrappedInstance).getPropertyValue("sync").equals(Boolean.TRUE)) .withFailMessage("Kafka Sync Producer should have been enabled."); producerBinding.unbind(); } @@ -590,8 +688,8 @@ public class KafkaBinderTests extends backOffPolicy.setBackOffPeriod(1000); metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); binder.setMetadataRetryOperations(metatadataRetrievalRetryOperations); + DirectChannel output = new DirectChannel(); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - DirectChannel output = createBindableChannel("output", createConsumerBindingProperties(consumerProperties)); String testTopicName = "nonexisting" + System.currentTimeMillis(); try { binder.doBindConsumer(testTopicName, "test", output, consumerProperties); @@ -601,92 +699,114 @@ public class KafkaBinderTests extends assertThat(e).isInstanceOf(BinderException.class); assertThat(e).hasMessageContaining("Topic " + testTopicName + " does not exist"); } - - try { - binder.getConnectionFactory().getPartitions(testTopicName); - fail(); - } - catch (Exception e) { - assertThat(e).isInstanceOf(TopicNotFoundException.class); - } } @Test public void testAutoConfigureTopicsDisabledSucceedsIfTopicExisting() throws Exception { - String testTopicName = "existing" + System.currentTimeMillis(); - AdminUtils.createTopic(kafkaTestSupport.getZkClient(), testTopicName, 5, 1, new Properties()); KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + + final ZkClient zkClient = new ZkClient(configurationProperties.getZkConnectionString(), + configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), + ZKStringSerializer$.MODULE$); + + final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); + + String testTopicName = "existing" + System.currentTimeMillis(); + AdminUtils.createTopic(zkUtils, testTopicName, 5, 1, new Properties()); + configurationProperties.setAutoCreateTopics(false); KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties); GenericApplicationContext context = new GenericApplicationContext(); context.refresh(); binder.setApplicationContext(context); binder.afterPropertiesSet(); + DirectChannel output = new DirectChannel(); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - DirectChannel output = createBindableChannel("output", createConsumerBindingProperties(consumerProperties)); Binding binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties); binding.unbind(); } @Test public void testAutoAddPartitionsDisabledFailsIfTopicUnderpartitioned() throws Exception { - String testTopicName = "existing" + System.currentTimeMillis(); - AdminUtils.createTopic(kafkaTestSupport.getZkClient(), testTopicName, 1, 1, new Properties()); KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + + final ZkClient zkClient = new ZkClient(configurationProperties.getZkConnectionString(), + configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), + ZKStringSerializer$.MODULE$); + + final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); + + String testTopicName = "existing" + System.currentTimeMillis(); + AdminUtils.createTopic(zkUtils, testTopicName, 1, 1, new Properties()); configurationProperties.setAutoAddPartitions(false); KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties); GenericApplicationContext context = new GenericApplicationContext(); context.refresh(); binder.setApplicationContext(context); binder.afterPropertiesSet(); + DirectChannel output = new DirectChannel(); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); // this consumer must consume from partition 2 consumerProperties.setInstanceCount(3); consumerProperties.setInstanceIndex(2); - DirectChannel output = createBindableChannel("output", createConsumerBindingProperties(consumerProperties)); try { binder.doBindConsumer(testTopicName, "test", output, consumerProperties); } catch (Exception e) { assertThat(e).isInstanceOf(BinderException.class); - assertThat(e).hasMessageContaining( - "The number of expected partitions was: 3, but 1 has been found instead"); + assertThat(e) + .hasMessageContaining("The number of expected partitions was: 3, but 1 has been found instead"); } } @Test public void testAutoAddPartitionsDisabledSucceedsIfTopicPartitionedCorrectly() throws Exception { + Binding binding = null; + try { + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - String testTopicName = "existing" + System.currentTimeMillis(); - AdminUtils.createTopic(kafkaTestSupport.getZkClient(), testTopicName, 6, 1, new Properties()); - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - configurationProperties.setAutoAddPartitions(false); - KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties); - GenericApplicationContext context = new GenericApplicationContext(); - RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); - metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); - FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); - backOffPolicy.setBackOffPeriod(1000); - metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); - binder.setMetadataRetryOperations(metatadataRetrievalRetryOperations); - context.refresh(); - binder.setApplicationContext(context); - binder.afterPropertiesSet(); - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - // this consumer must consume from partition 2 - consumerProperties.setInstanceCount(3); - consumerProperties.setInstanceIndex(2); - DirectChannel output = createBindableChannel("output", createConsumerBindingProperties(consumerProperties)); - Binding binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties); + final ZkClient zkClient = new ZkClient(configurationProperties.getZkConnectionString(), + configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), + ZKStringSerializer$.MODULE$); - Partition[] listenedPartitions = TestUtils.getPropertyValue(binding, - "endpoint.messageListenerContainer.partitions", Partition[].class); + final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); - assertThat(listenedPartitions).hasSize(2); - assertThat(listenedPartitions).contains(new Partition(testTopicName, 2), new Partition(testTopicName, 5)); - Collection partitions = binder.getConnectionFactory().getPartitions(testTopicName); - assertThat(partitions).hasSize(6); - binding.unbind(); + String testTopicName = "existing" + System.currentTimeMillis(); + AdminUtils.createTopic(zkUtils, testTopicName, 6, 1, new Properties()); + configurationProperties.setAutoAddPartitions(false); + KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties); + GenericApplicationContext context = new GenericApplicationContext(); + RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); + metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); + FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); + backOffPolicy.setBackOffPeriod(1000); + metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); + binder.setMetadataRetryOperations(metatadataRetrievalRetryOperations); + context.refresh(); + binder.setApplicationContext(context); + binder.afterPropertiesSet(); + DirectChannel output = new DirectChannel(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + // this consumer must consume from partition 2 + consumerProperties.setInstanceCount(3); + consumerProperties.setInstanceIndex(2); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); + + binding = binder.doBindConsumer(testTopicName, "test-x", output, consumerProperties); + + TopicPartitionInitialOffset[] listenedPartitions = TestUtils.getPropertyValue(binding, + "endpoint.messageListenerContainer.containerProperties.topicPartitions", + TopicPartitionInitialOffset[].class); + assertThat(listenedPartitions).hasSize(2); + assertThat(listenedPartitions).contains(new TopicPartitionInitialOffset(testTopicName, 2), + new TopicPartitionInitialOffset(testTopicName, 5)); + Collection partitions = + consumerFactory().createConsumer().partitionsFor(testTopicName); + assertThat(partitions).hasSize(6); + } + finally { + binding.unbind(); + } } @Test @@ -704,8 +824,8 @@ public class KafkaBinderTests extends backOffPolicy.setBackOffPeriod(1000); metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); binder.setMetadataRetryOperations(metatadataRetrievalRetryOperations); + DirectChannel output = new DirectChannel(); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - DirectChannel output = createBindableChannel("output", createConsumerBindingProperties(consumerProperties)); String testTopicName = "nonexisting" + System.currentTimeMillis(); Binding binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties); binding.unbind(); @@ -714,8 +834,16 @@ public class KafkaBinderTests extends @Test public void testPartitionCountNotReduced() throws Exception { String testTopicName = "existing" + System.currentTimeMillis(); - AdminUtils.createTopic(kafkaTestSupport.getZkClient(), testTopicName, 6, 1, new Properties()); + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + + final ZkClient zkClient = new ZkClient(configurationProperties.getZkConnectionString(), + configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), + ZKStringSerializer$.MODULE$); + + final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); + + AdminUtils.createTopic(zkUtils, testTopicName, 6, 1, new Properties()); configurationProperties.setAutoAddPartitions(true); KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties); GenericApplicationContext context = new GenericApplicationContext(); @@ -733,15 +861,22 @@ public class KafkaBinderTests extends Binding binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties); binding.unbind(); TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(testTopicName, - kafkaTestSupport.getZkClient()); + zkUtils); assertThat(topicMetadata.partitionsMetadata().size()).isEqualTo(6); } @Test public void testPartitionCountIncreasedIfAutoAddPartitionsSet() throws Exception { - String testTopicName = "existing" + System.currentTimeMillis(); - AdminUtils.createTopic(kafkaTestSupport.getZkClient(), testTopicName, 1, 1, new Properties()); KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + + final ZkClient zkClient = new ZkClient(configurationProperties.getZkConnectionString(), + configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), + ZKStringSerializer$.MODULE$); + + final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); + + String testTopicName = "existing" + System.currentTimeMillis(); + AdminUtils.createTopic(zkUtils, testTopicName, 1, 1, new Properties()); configurationProperties.setMinPartitionCount(6); configurationProperties.setAutoAddPartitions(true); KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties); @@ -760,10 +895,315 @@ public class KafkaBinderTests extends Binding binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties); binding.unbind(); TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(testTopicName, - kafkaTestSupport.getZkClient()); + zkUtils); assertThat(topicMetadata.partitionsMetadata().size()).isEqualTo(6); } + @Test + @Override + @SuppressWarnings("unchecked") + public void testSendAndReceiveMultipleTopics() throws Exception { + Binder binder = getBinder(); + + DirectChannel moduleOutputChannel1 = createBindableChannel("output1", + createProducerBindingProperties(createProducerProperties())); + DirectChannel moduleOutputChannel2 = createBindableChannel("output2", + createProducerBindingProperties(createProducerProperties())); + + QueueChannel moduleInputChannel = new QueueChannel(); + + Binding producerBinding1 = binder.bindProducer("foo.x", moduleOutputChannel1, + createProducerProperties()); + Binding producerBinding2 = binder.bindProducer("foo.y", moduleOutputChannel2, + createProducerProperties()); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); + Binding consumerBinding1 = binder.bindConsumer("foo.x", "test", moduleInputChannel, + consumerProperties); + Binding consumerBinding2 = binder.bindConsumer("foo.y", "test", moduleInputChannel, + consumerProperties); + + String testPayload1 = "foo" + UUID.randomUUID().toString(); + Message message1 = org.springframework.integration.support.MessageBuilder.withPayload( + testPayload1.getBytes()).build(); + String testPayload2 = "foo" + UUID.randomUUID().toString(); + Message message2 = org.springframework.integration.support.MessageBuilder.withPayload( + testPayload2.getBytes()).build(); + + // Let the consumer actually bind to the producer before sending a msg + binderBindUnbindLatency(); + moduleOutputChannel1.send(message1); + moduleOutputChannel2.send(message2); + + + Message[] messages = new Message[2]; + messages[0] = receive(moduleInputChannel); + messages[1] = receive(moduleInputChannel); + + assertThat(messages[0]).isNotNull(); + assertThat(messages[1]).isNotNull(); + assertThat(messages).extracting("payload").containsExactlyInAnyOrder(testPayload1.getBytes(), + testPayload2.getBytes()); + + producerBinding1.unbind(); + producerBinding2.unbind(); + + consumerBinding1.unbind(); + consumerBinding2.unbind(); + } + + @Test + @Override + @SuppressWarnings("unchecked") + public void testTwoRequiredGroups() throws Exception { + Binder binder = getBinder(); + ExtendedProducerProperties producerProperties = createProducerProperties(); + + DirectChannel output = createBindableChannel("output", createProducerBindingProperties(producerProperties)); + + String testDestination = "testDestination" + UUID.randomUUID().toString().replace("-", ""); + + producerProperties.setRequiredGroups("test1", "test2"); + Binding producerBinding = binder.bindProducer(testDestination, output, producerProperties); + + String testPayload = "foo-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload.getBytes())); + + QueueChannel inbound1 = new QueueChannel(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); + Binding consumerBinding1 = binder.bindConsumer(testDestination, "test1", inbound1, + consumerProperties); + QueueChannel inbound2 = new QueueChannel(); + Binding consumerBinding2 = binder.bindConsumer(testDestination, "test2", inbound2, + consumerProperties); + + Message receivedMessage1 = receive(inbound1); + assertThat(receivedMessage1).isNotNull(); + assertThat(new String((byte[]) receivedMessage1.getPayload())).isEqualTo(testPayload); + Message receivedMessage2 = receive(inbound2); + assertThat(receivedMessage2).isNotNull(); + assertThat(new String((byte[]) receivedMessage2.getPayload())).isEqualTo(testPayload); + + consumerBinding1.unbind(); + consumerBinding2.unbind(); + producerBinding.unbind(); + } + + @Test + @Override + @SuppressWarnings("unchecked") + public void testPartitionedModuleSpEL() throws Exception { + Binder binder = getBinder(); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.setConcurrency(2); + consumerProperties.setInstanceIndex(0); + consumerProperties.setInstanceCount(3); + consumerProperties.setPartitioned(true); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); + QueueChannel input0 = new QueueChannel(); + input0.setBeanName("test.input0S"); + Binding input0Binding = binder.bindConsumer("part.0", "test", input0, consumerProperties); + consumerProperties.setInstanceIndex(1); + QueueChannel input1 = new QueueChannel(); + input1.setBeanName("test.input1S"); + Binding input1Binding = binder.bindConsumer("part.0", "test", input1, consumerProperties); + consumerProperties.setInstanceIndex(2); + QueueChannel input2 = new QueueChannel(); + input2.setBeanName("test.input2S"); + Binding input2Binding = binder.bindConsumer("part.0", "test", input2, consumerProperties); + + ExtendedProducerProperties producerProperties = createProducerProperties(); + producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload")); + producerProperties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("hashCode()")); + producerProperties.setPartitionCount(3); + + DirectChannel output = createBindableChannel("output", createProducerBindingProperties(producerProperties)); + output.setBeanName("test.output"); + Binding outputBinding = binder.bindProducer("part.0", output, producerProperties); + try { + Object endpoint = extractEndpoint(outputBinding); + assertThat(getEndpointRouting(endpoint)) + .contains(getExpectedRoutingBaseDestination("part.0", "test") + "-' + headers['partition']"); + } + catch (UnsupportedOperationException ignored) { + } + + Message message2 = org.springframework.integration.support.MessageBuilder.withPayload(2) + .setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "foo") + .setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 42) + .setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 43).build(); + output.send(message2); + output.send(new GenericMessage<>(1)); + output.send(new GenericMessage<>(0)); + + Message receive0 = receive(input0); + assertThat(receive0).isNotNull(); + Message receive1 = receive(input1); + assertThat(receive1).isNotNull(); + Message receive2 = receive(input2); + assertThat(receive2).isNotNull(); + + Condition> correlationHeadersForPayload2 = new Condition>() { + + @Override + public boolean matches(Message value) { + IntegrationMessageHeaderAccessor accessor = new IntegrationMessageHeaderAccessor(value); + return "foo".equals(accessor.getCorrelationId()) && 42 == accessor.getSequenceNumber() + && 43 == accessor.getSequenceSize(); + } + }; + + if (usesExplicitRouting()) { + assertThat(receive0.getPayload()).isEqualTo(0); + assertThat(receive1.getPayload()).isEqualTo(1); + assertThat(receive2.getPayload()).isEqualTo(2); + assertThat(receive2).has(correlationHeadersForPayload2); + } + else { + List> receivedMessages = Arrays.asList(receive0, receive1, receive2); + assertThat(receivedMessages).extracting("payload").containsExactlyInAnyOrder(0, 1, 2); + Condition> payloadIs2 = new Condition>() { + + @Override + public boolean matches(Message value) { + return value.getPayload().equals(2); + } + }; + assertThat(receivedMessages).filteredOn(payloadIs2).areExactly(1, correlationHeadersForPayload2); + + } + input0Binding.unbind(); + input1Binding.unbind(); + input2Binding.unbind(); + outputBinding.unbind(); + } + + @Test + @Override + @SuppressWarnings("unchecked") + public void testPartitionedModuleJava() throws Exception { + Binder binder = getBinder(); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.setConcurrency(2); + consumerProperties.setInstanceCount(3); + consumerProperties.setInstanceIndex(0); + consumerProperties.setPartitioned(true); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); + QueueChannel input0 = new QueueChannel(); + input0.setBeanName("test.input0J"); + Binding input0Binding = binder.bindConsumer("partJ.0", "test", input0, consumerProperties); + consumerProperties.setInstanceIndex(1); + QueueChannel input1 = new QueueChannel(); + input1.setBeanName("test.input1J"); + Binding input1Binding = binder.bindConsumer("partJ.0", "test", input1, consumerProperties); + consumerProperties.setInstanceIndex(2); + QueueChannel input2 = new QueueChannel(); + input2.setBeanName("test.input2J"); + Binding input2Binding = binder.bindConsumer("partJ.0", "test", input2, consumerProperties); + + ExtendedProducerProperties producerProperties = createProducerProperties(); + producerProperties.setPartitionKeyExtractorClass(PartitionTestSupport.class); + producerProperties.setPartitionSelectorClass(PartitionTestSupport.class); + producerProperties.setPartitionCount(3); + DirectChannel output = createBindableChannel("output", createProducerBindingProperties(producerProperties)); + output.setBeanName("test.output"); + Binding outputBinding = binder.bindProducer("partJ.0", output, producerProperties); + if (usesExplicitRouting()) { + Object endpoint = extractEndpoint(outputBinding); + assertThat(getEndpointRouting(endpoint)). + contains(getExpectedRoutingBaseDestination("partJ.0", "test") + "-' + headers['partition']"); + } + + output.send(new GenericMessage<>(2)); + output.send(new GenericMessage<>(1)); + output.send(new GenericMessage<>(0)); + + Message receive0 = receive(input0); + assertThat(receive0).isNotNull(); + Message receive1 = receive(input1); + assertThat(receive1).isNotNull(); + Message receive2 = receive(input2); + assertThat(receive2).isNotNull(); + + if (usesExplicitRouting()) { + assertThat(receive0.getPayload()).isEqualTo(0); + assertThat(receive1.getPayload()).isEqualTo(1); + assertThat(receive2.getPayload()).isEqualTo(2); + } + else { + List> receivedMessages = Arrays.asList(receive0, receive1, receive2); + assertThat(receivedMessages).extracting("payload").containsExactlyInAnyOrder(0, 1, 2); + } + + input0Binding.unbind(); + input1Binding.unbind(); + input2Binding.unbind(); + outputBinding.unbind(); + } + + @Test + @Override + @SuppressWarnings("unchecked") + public void testAnonymousGroup() throws Exception { + Binder binder = getBinder(); + BindingProperties producerBindingProperties = createProducerBindingProperties(createProducerProperties()); + DirectChannel output = createBindableChannel("output", producerBindingProperties); + Binding producerBinding = binder.bindProducer("defaultGroup.0", output, + producerBindingProperties.getProducer()); + + QueueChannel input1 = new QueueChannel(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + //consumerProperties.getExtension().setAutoRebalanceEnabled(false); + Binding binding1 = binder.bindConsumer("defaultGroup.0", null, input1, + consumerProperties); + + QueueChannel input2 = new QueueChannel(); + Binding binding2 = binder.bindConsumer("defaultGroup.0", null, input2, + consumerProperties); + //Since we don't provide any topic info, let Kafka bind the consumer successfully + Thread.sleep(1000); + String testPayload1 = "foo-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload1.getBytes())); + + Message receivedMessage1 = (Message) receive(input1); + assertThat(receivedMessage1).isNotNull(); + assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload1); + + Message receivedMessage2 = (Message) receive(input2); + assertThat(receivedMessage2).isNotNull(); + assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload1); + + binding2.unbind(); + + String testPayload2 = "foo-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload2.getBytes())); + + binding2 = binder.bindConsumer("defaultGroup.0", null, input2, consumerProperties); + //Since we don't provide any topic info, let Kafka bind the consumer successfully + Thread.sleep(1000); + String testPayload3 = "foo-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload3.getBytes())); + + receivedMessage1 = (Message) receive(input1); + assertThat(receivedMessage1).isNotNull(); + assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload2); + receivedMessage1 = (Message) receive(input1); + assertThat(receivedMessage1).isNotNull(); + assertThat(new String(receivedMessage1.getPayload())).isNotNull(); + + receivedMessage2 = (Message) receive(input2); + assertThat(receivedMessage2).isNotNull(); + assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload3); + + producerBinding.unbind(); + binding1.unbind(); + binding2.unbind(); + } + private static final class FailingInvocationCountingMessageHandler implements MessageHandler { private int invocationCount; diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java index 12ea4eae0..69d9af362 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java @@ -27,9 +27,9 @@ import org.springframework.context.support.GenericApplicationContext; import org.springframework.integration.codec.Codec; import org.springframework.integration.codec.kryo.KryoRegistrar; import org.springframework.integration.codec.kryo.PojoCodec; -import org.springframework.integration.kafka.support.LoggingProducerListener; -import org.springframework.integration.kafka.support.ProducerListener; import org.springframework.integration.tuple.TupleKryoRegistrar; +import org.springframework.kafka.support.LoggingProducerListener; +import org.springframework.kafka.support.ProducerListener; import com.esotericsoftware.kryo.Kryo; import com.esotericsoftware.kryo.Registration; diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java index ad2f8112a..83e0eee69 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java @@ -18,14 +18,12 @@ package org.springframework.cloud.stream.binder.kafka; import java.util.Arrays; -import org.junit.Ignore; import org.junit.Test; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.HeaderMode; -import org.springframework.context.Lifecycle; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; @@ -48,14 +46,15 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { @Override public void testPartitionedModuleJava() throws Exception { KafkaTestBinder binder = getBinder(); - ExtendedProducerProperties producerProperties = createProducerProperties(); - producerProperties.setHeaderMode(HeaderMode.raw); - producerProperties.setPartitionKeyExtractorClass(RawKafkaPartitionTestSupport.class); - producerProperties.setPartitionSelectorClass(RawKafkaPartitionTestSupport.class); - producerProperties.setPartitionCount(6); + ExtendedProducerProperties properties = createProducerProperties(); + properties.setHeaderMode(HeaderMode.raw); + properties.setPartitionKeyExtractorClass(RawKafkaPartitionTestSupport.class); + properties.setPartitionSelectorClass(RawKafkaPartitionTestSupport.class); + properties.setPartitionCount(6); - DirectChannel output = createBindableChannel("output", createProducerBindingProperties(producerProperties)); - Binding outputBinding = binder.bindProducer("partJ.0", output, producerProperties); + DirectChannel output = createBindableChannel("output", createProducerBindingProperties(properties)); + output.setBeanName("test.output"); + Binding outputBinding = binder.bindProducer("partJ.0", output, properties); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); consumerProperties.setConcurrency(2); @@ -63,6 +62,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { consumerProperties.setInstanceIndex(0); consumerProperties.setPartitioned(true); consumerProperties.setHeaderMode(HeaderMode.raw); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); QueueChannel input0 = new QueueChannel(); input0.setBeanName("test.input0J"); Binding input0Binding = binder.bindConsumer("partJ.0", "test", input0, consumerProperties); @@ -75,9 +75,9 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { input2.setBeanName("test.input2J"); Binding input2Binding = binder.bindConsumer("partJ.0", "test", input2, consumerProperties); - output.send(new GenericMessage<>(new byte[] { (byte) 0 })); - output.send(new GenericMessage<>(new byte[] { (byte) 1 })); - output.send(new GenericMessage<>(new byte[] { (byte) 2 })); + output.send(new GenericMessage<>(new byte[] {(byte) 0})); + output.send(new GenericMessage<>(new byte[] {(byte) 1})); + output.send(new GenericMessage<>(new byte[] {(byte) 2})); Message receive0 = receive(input0); assertThat(receive0).isNotNull(); @@ -99,26 +99,31 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { @Override public void testPartitionedModuleSpEL() throws Exception { KafkaTestBinder binder = getBinder(); - ExtendedProducerProperties producerProperties = createProducerProperties(); - producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload[0]")); - producerProperties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("hashCode()")); - producerProperties.setPartitionCount(6); - producerProperties.setHeaderMode(HeaderMode.raw); - DirectChannel output = createBindableChannel("output", createProducerBindingProperties(producerProperties)); + ExtendedProducerProperties properties = createProducerProperties(); + properties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload[0]")); + properties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("hashCode()")); + properties.setPartitionCount(6); + properties.setHeaderMode(HeaderMode.raw); + + DirectChannel output = createBindableChannel("output", createProducerBindingProperties(properties)); output.setBeanName("test.output"); - Binding outputBinding = binder.bindProducer("part.0", output, producerProperties); + Binding outputBinding = binder.bindProducer("part.0", output, properties); try { - Lifecycle endpoint = extractEndpoint(outputBinding); - assertThat(getEndpointRouting(endpoint)).contains("part.0-' + headers['partition']"); + Object endpoint = extractEndpoint(outputBinding); + assertThat(getEndpointRouting(endpoint)) + .contains(getExpectedRoutingBaseDestination("part.0", "test") + "-' + headers['partition']"); } catch (UnsupportedOperationException ignored) { } + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); consumerProperties.setConcurrency(2); consumerProperties.setInstanceIndex(0); consumerProperties.setInstanceCount(3); consumerProperties.setPartitioned(true); consumerProperties.setHeaderMode(HeaderMode.raw); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); QueueChannel input0 = new QueueChannel(); input0.setBeanName("test.input0S"); Binding input0Binding = binder.bindConsumer("part.0", "test", input0, consumerProperties); @@ -131,13 +136,13 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { input2.setBeanName("test.input2S"); Binding input2Binding = binder.bindConsumer("part.0", "test", input2, consumerProperties); - Message message2 = MessageBuilder.withPayload(new byte[] { 2 }) + Message message2 = MessageBuilder.withPayload(new byte[] {2}) .setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "foo") .setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 42) .setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 43).build(); output.send(message2); - output.send(new GenericMessage<>(new byte[] { 1 })); - output.send(new GenericMessage<>(new byte[] { 0 })); + output.send(new GenericMessage<>(new byte[] {1})); + output.send(new GenericMessage<>(new byte[] {0})); Message receive0 = receive(input0); assertThat(receive0).isNotNull(); Message receive1 = receive(input1); @@ -156,15 +161,14 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { @Override public void testSendAndReceive() throws Exception { KafkaTestBinder binder = getBinder(); + DirectChannel moduleOutputChannel = new DirectChannel(); + QueueChannel moduleInputChannel = new QueueChannel(); ExtendedProducerProperties producerProperties = createProducerProperties(); - DirectChannel moduleOutputChannel = createBindableChannel("output", - createProducerBindingProperties(producerProperties)); producerProperties.setHeaderMode(HeaderMode.raw); Binding producerBinding = binder.bindProducer("foo.0", moduleOutputChannel, producerProperties); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); consumerProperties.setHeaderMode(HeaderMode.raw); - QueueChannel moduleInputChannel = new QueueChannel(); Binding consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel, consumerProperties); Message message = MessageBuilder.withPayload("foo".getBytes()).build(); @@ -178,14 +182,6 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { consumerBinding.unbind(); } - // Ignored, since raw mode does not support headers - @Test - @Override - @Ignore - public void testSendAndReceiveNoOriginalContentType() throws Exception { - - } - @Test public void testSendAndReceiveWithExplicitConsumerGroup() { KafkaTestBinder binder = getBinder(); @@ -196,9 +192,11 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { QueueChannel module3InputChannel = new QueueChannel(); ExtendedProducerProperties producerProperties = createProducerProperties(); producerProperties.setHeaderMode(HeaderMode.raw); - Binding producerBinding = binder.bindProducer("baz.0", moduleOutputChannel, producerProperties); + Binding producerBinding = binder.bindProducer("baz.0", moduleOutputChannel, + producerProperties); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); consumerProperties.setHeaderMode(HeaderMode.raw); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); Binding input1Binding = binder.bindConsumer("baz.0", "test", module1InputChannel, consumerProperties); // A new module is using the tap as an input channel