From 50b8955dfc6330c150195d84a9c64c1ba95b0bef Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 17 Nov 2017 18:06:46 -0500 Subject: [PATCH] Upgrade to Spring Kafka 2.1, Kafka 1.0.0 Resolves #259 - Remove the kafka server dependency from build/binder artifact - Remove AdminUtilOperation and KafkaAdminUtilOperation - Rely on AdminClient for provisioning operations - Update KStream components with the new changes - Test updates - Polishing Add timeout to all the blocking AdminClient calls Addressing PR review comments Addressing PR review comments Addressing PR review Update SK, SIK to 2.1.0.RELEASE and 3.0.0.RELEASE respectively Update Kafka Streams class name changes --- pom.xml | 52 +- spring-cloud-stream-binder-kafka-core/pom.xml | 4 - .../kafka/admin/AdminUtilsOperation.java | 75 - .../kafka/admin/KafkaAdminUtilsOperation.java | 54 - .../KafkaBinderConfigurationProperties.java | 8 +- .../provisioning/KafkaTopicProvisioner.java | 244 ++- spring-cloud-stream-binder-kafka/pom.xml | 10 - .../kafka/KafkaMessageChannelBinder.java | 12 +- .../config/KafkaBinderConfiguration.java | 16 +- ...afkaBinderConfigurationPropertiesTest.java | 4 +- .../kafka/KafkaBinderConfigurationTest.java | 5 +- .../stream/binder/kafka/KafkaBinderTests.java | 1796 ++++++++--------- .../binder/kafka/KafkaBinderUnitTests.java | 7 +- .../stream/binder/kafka/KafkaTestBinder.java | 16 +- spring-cloud-stream-binder-kstream/pom.xml | 31 +- .../kstream/KStreamBoundElementFactory.java | 6 +- .../config/KStreamBinderConfiguration.java | 19 +- ...KStreamBinderSupportAutoConfiguration.java | 18 +- ...treamInteractiveQueryIntegrationTests.java | 10 +- 19 files changed, 1100 insertions(+), 1287 deletions(-) delete mode 100644 spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/AdminUtilsOperation.java delete mode 100644 spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/KafkaAdminUtilsOperation.java diff --git a/pom.xml b/pom.xml index 72430d831..cf1c22819 100644 --- a/pom.xml +++ b/pom.xml @@ -12,9 +12,9 @@ 1.8 - 2.0.0.RELEASE - 0.11.0.0 - 3.0.0.M2 + 2.1.0.RELEASE + 3.0.0.RELEASE + 1.0.0 2.0.0.BUILD-SNAPSHOT @@ -42,25 +42,6 @@ spring-cloud-stream ${spring-cloud-stream.version} - - org.apache.kafka - kafka_2.11 - ${kafka.version} - - - jline - jline - - - org.slf4j - slf4j-log4j12 - - - log4j - log4j - - - org.apache.kafka kafka-clients @@ -88,12 +69,6 @@ test ${spring-kafka.version} - - org.apache.kafka - kafka_2.11 - test - ${kafka.version} - org.apache.kafka kafka-streams @@ -105,6 +80,27 @@ + + org.apache.kafka + kafka_2.11 + test + test + ${kafka.version} + + + jline + jline + + + org.slf4j + slf4j-log4j12 + + + log4j + log4j + + + diff --git a/spring-cloud-stream-binder-kafka-core/pom.xml b/spring-cloud-stream-binder-kafka-core/pom.xml index 7bc68d357..c50b86b46 100644 --- a/spring-cloud-stream-binder-kafka-core/pom.xml +++ b/spring-cloud-stream-binder-kafka-core/pom.xml @@ -38,10 +38,6 @@ org.apache.kafka kafka-clients - - org.apache.kafka - kafka_2.11 - org.springframework.integration spring-integration-kafka diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/AdminUtilsOperation.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/AdminUtilsOperation.java deleted file mode 100644 index 4cb2b8bff..000000000 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/AdminUtilsOperation.java +++ /dev/null @@ -1,75 +0,0 @@ -/* - * Copyright 2002-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.admin; - -import java.util.Properties; - -import kafka.utils.ZkUtils; - -/** - * API around {@link kafka.admin.AdminUtils} to support - * various versions of Kafka brokers. - * - * Note: Implementations that support Kafka brokers other than 0.10, need to use - * a possible strategy that involves reflection around {@link kafka.admin.AdminUtils}. - * - * @author Soby Chacko - */ -public interface AdminUtilsOperation { - - /** - * Invoke {@link kafka.admin.AdminUtils#addPartitions} - * - * @param zkUtils Zookeeper utils - * @param topic name of the topic - * @param numPartitions - * @param replicaAssignmentStr - * @param checkBrokerAvailable - */ - void invokeAddPartitions(ZkUtils zkUtils, String topic, int numPartitions, - String replicaAssignmentStr, boolean checkBrokerAvailable); - - /** - * Invoke {@link kafka.admin.AdminUtils#fetchTopicMetadataFromZk} - * - * @param topic name - * @param zkUtils zookeeper utils - * @return error code - */ - short errorCodeFromTopicMetadata(String topic, ZkUtils zkUtils); - - /** - * Find partition size from Kafka broker using {@link kafka.admin.AdminUtils} - * - * @param topic name - * @param zkUtils zookeeper utils - * @return partition size - */ - int partitionSize(String topic, ZkUtils zkUtils); - - /** - * Inovke {@link kafka.admin.AdminUtils#createTopic} - * - * @param zkUtils zookeeper utils - * @param topic name - * @param partitions - * @param replicationFactor - * @param topicConfig - */ - void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions, - int replicationFactor, Properties topicConfig); -} diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/KafkaAdminUtilsOperation.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/KafkaAdminUtilsOperation.java deleted file mode 100644 index 9cad3aa71..000000000 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/KafkaAdminUtilsOperation.java +++ /dev/null @@ -1,54 +0,0 @@ -/* - * Copyright 2002-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.admin; - -import java.util.Properties; - -import kafka.admin.AdminUtils; -import kafka.utils.ZkUtils; -import org.apache.kafka.common.requests.MetadataResponse; - -/** - * @author Soby Chacko - */ -public class KafkaAdminUtilsOperation implements AdminUtilsOperation { - - public void invokeAddPartitions(ZkUtils zkUtils, String topic, int numPartitions, - String replicaAssignmentStr, boolean checkBrokerAvailable) { - AdminUtils.addPartitions(zkUtils, topic, numPartitions, replicaAssignmentStr, checkBrokerAvailable, null); - } - - public short errorCodeFromTopicMetadata(String topic, ZkUtils zkUtils) { - - MetadataResponse.TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topic, zkUtils); - return topicMetadata.error().code(); - } - - @SuppressWarnings("unchecked") - public int partitionSize(String topic, ZkUtils zkUtils) { - - MetadataResponse.TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topic, zkUtils); - return topicMetadata.partitionMetadata().size(); - } - - public void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions, - int replicationFactor, Properties topicConfig) { - - AdminUtils.createTopic(zkUtils, topic, partitions, replicationFactor, - topicConfig, null); - } -} diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java index 7659cf24a..4debb72c8 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java @@ -40,6 +40,8 @@ import org.springframework.util.StringUtils; @ConfigurationProperties(prefix = "spring.cloud.stream.kafka.binder") public class KafkaBinderConfigurationProperties { + private static final String DEFAULT_KAFKA_CONNECTION_STRING = "localhost:9092"; + private final Transaction transaction = new Transaction(); @Autowired(required = false) @@ -99,7 +101,7 @@ public class KafkaBinderConfigurationProperties { private JaasLoginModuleConfiguration jaas; /** - * The bean name of a custom header mapper to use instead of a {@link DefaultKafkaHeaderMapper}. + * The bean name of a custom header mapper to use instead of a {@link org.springframework.kafka.support.DefaultKafkaHeaderMapper}. */ private String headerMapperBeanName; @@ -115,6 +117,10 @@ public class KafkaBinderConfigurationProperties { return toConnectionString(this.brokers, this.defaultBrokerPort); } + public String getDefaultKafkaConnectionString() { + return DEFAULT_KAFKA_CONNECTION_STRING; + } + public String[] getHeaders() { return this.headers; } diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java index bab31e0c2..8f3a56a3c 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java @@ -17,21 +17,32 @@ package org.springframework.cloud.stream.binder.kafka.provisioning; import java.util.Collection; -import java.util.Properties; +import java.util.Collections; +import java.util.Map; +import java.util.Set; import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; -import kafka.common.ErrorMapping; -import kafka.utils.ZkUtils; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.kafka.clients.admin.AdminClient; +import org.apache.kafka.clients.admin.AdminClientConfig; +import org.apache.kafka.clients.admin.CreatePartitionsResult; +import org.apache.kafka.clients.admin.CreateTopicsResult; +import org.apache.kafka.clients.admin.DescribeTopicsResult; +import org.apache.kafka.clients.admin.ListTopicsResult; +import org.apache.kafka.clients.admin.NewPartitions; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.admin.TopicDescription; +import org.apache.kafka.common.KafkaFuture; import org.apache.kafka.common.PartitionInfo; -import org.apache.kafka.common.security.JaasUtils; import org.springframework.beans.factory.InitializingBean; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.cloud.stream.binder.BinderException; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; -import org.springframework.cloud.stream.binder.kafka.admin.AdminUtilsOperation; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; @@ -45,6 +56,7 @@ import org.springframework.retry.backoff.ExponentialBackOffPolicy; import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; /** @@ -58,18 +70,30 @@ import org.springframework.util.StringUtils; public class KafkaTopicProvisioner implements ProvisioningProvider, ExtendedProducerProperties>, InitializingBean { + private static final int DEFAULT_OPERATION_TIMEOUT = 30; + private final Log logger = LogFactory.getLog(getClass()); - private final KafkaBinderConfigurationProperties configurationProperties; + private KafkaBinderConfigurationProperties configurationProperties; - private final AdminUtilsOperation adminUtilsOperation; + private final AdminClient adminClient; private RetryOperations metadataRetryOperations; + private int operationTimeout = DEFAULT_OPERATION_TIMEOUT; + public KafkaTopicProvisioner(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties, - AdminUtilsOperation adminUtilsOperation) { + KafkaProperties kafkaProperties) { + Assert.isTrue(kafkaProperties != null, "KafkaProperties cannot be null"); + Map adminClientProperties = kafkaProperties.buildAdminProperties(); + String kafkaConnectionString = kafkaBinderConfigurationProperties.getKafkaConnectionString(); + + if (ObjectUtils.isEmpty(adminClientProperties.get(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG)) + || !kafkaConnectionString.equals(kafkaBinderConfigurationProperties.getDefaultKafkaConnectionString())) { + adminClientProperties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConnectionString); + } this.configurationProperties = kafkaBinderConfigurationProperties; - this.adminUtilsOperation = adminUtilsOperation; + this.adminClient = AdminClient.create(adminClientProperties); } /** @@ -103,14 +127,20 @@ public class KafkaTopicProvisioner implements ProvisioningProvider> all = describeTopicsResult.all(); + + try { + Map topicDescriptions = all.get(operationTimeout, TimeUnit.SECONDS); + TopicDescription topicDescription = topicDescriptions.get(name); + int partitions = topicDescription.partitions().size(); + return new KafkaProducerDestination(name, partitions); + } + catch (Exception e) { + throw new ProvisioningException("Problems encountered with partitions finding", e); + } } else { return new KafkaProducerDestination(name); @@ -127,30 +157,67 @@ public class KafkaTopicProvisioner implements ProvisioningProvider> all = describeTopicsResult.all(); + try { + Map topicDescriptions = all.get(operationTimeout, TimeUnit.SECONDS); + TopicDescription topicDescription = topicDescriptions.get(name); + int partitions = topicDescription.partitions().size(); + ConsumerDestination dlqTopic = createDlqIfNeedBe(name, group, properties, anonymous, partitions); + if (dlqTopic != null) return dlqTopic; + return new KafkaConsumerDestination(name, partitions); + } + catch (Exception e) { + throw new ProvisioningException("provisioning exception", e); } - return new KafkaConsumerDestination(name, partitions); } return new KafkaConsumerDestination(name); } - private void createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(final String topicName, final int partitionCount, - boolean tolerateLowerPartitionsOnBroker) { - if (this.configurationProperties.isAutoCreateTopics() && adminUtilsOperation != null) { + private ConsumerDestination createDlqIfNeedBe(String name, String group, + ExtendedConsumerProperties properties, + boolean anonymous, int partitions) { + if (properties.getExtension().isEnableDlq() && !anonymous) { + String dlqTopic = StringUtils.hasText(properties.getExtension().getDlqName()) ? + properties.getExtension().getDlqName() : "error." + name + "." + group; + try { + createTopicAndPartitions(dlqTopic, partitions, properties.getExtension().isAutoRebalanceEnabled()); + } + catch (Throwable throwable) { + if (throwable instanceof Error) { + throw (Error) throwable; + } + else { + throw new ProvisioningException("provisioning exception", throwable); + } + } + return new KafkaConsumerDestination(name, partitions, dlqTopic); + } + return null; + } + + private void createTopic(String name, int partitionCount, boolean tolerateLowerPartitionsOnBroker) { + try { + createTopicIfNecessary(name, partitionCount, tolerateLowerPartitionsOnBroker); + } + catch (Throwable throwable) { + if (throwable instanceof Error) { + throw (Error) throwable; + } + else { + throw new ProvisioningException("provisioning exception", throwable); + } + } + } + + private void createTopicIfNecessary(final String topicName, final int partitionCount, + boolean tolerateLowerPartitionsOnBroker) throws Throwable { + if (this.configurationProperties.isAutoCreateTopics() && adminClient != null) { createTopicAndPartitions(topicName, partitionCount, tolerateLowerPartitionsOnBroker); } - else if (this.configurationProperties.isAutoCreateTopics() && adminUtilsOperation == null) { + else if (this.configurationProperties.isAutoCreateTopics() && adminClient == null) { this.logger.warn("Auto creation of topics is enabled, but Kafka AdminUtils class is not present on the classpath. " + "No topic will be created by the binder"); } @@ -164,71 +231,68 @@ public class KafkaTopicProvisioner implements ProvisioningProvider> namesFutures = listTopicsResult.names(); - final ZkUtils zkUtils = ZkUtils.apply(this.configurationProperties.getZkConnectionString(), - this.configurationProperties.getZkSessionTimeout(), - this.configurationProperties.getZkConnectionTimeout(), - JaasUtils.isZkSecurityEnabled()); - try { - short errorCode = adminUtilsOperation.errorCodeFromTopicMetadata(topicName, zkUtils); - if (errorCode == ErrorMapping.NoError()) { - // only consider minPartitionCount for resizing if autoAddPartitions is true - int effectivePartitionCount = this.configurationProperties.isAutoAddPartitions() - ? Math.max(this.configurationProperties.getMinPartitionCount(), partitionCount) - : partitionCount; - int partitionSize = adminUtilsOperation.partitionSize(topicName, zkUtils); - - if (partitionSize < effectivePartitionCount) { - if (this.configurationProperties.isAutoAddPartitions()) { - adminUtilsOperation.invokeAddPartitions(zkUtils, topicName, effectivePartitionCount, null, false); - } - else if (tolerateLowerPartitionsOnBroker) { - logger.warn("The number of expected partitions was: " + partitionCount + ", but " - + partitionSize + (partitionSize > 1 ? " have " : " has ") + "been found instead." - + "There will be " + (effectivePartitionCount - partitionSize) + " idle consumers"); - } - else { - throw new ProvisioningException("The number of expected partitions was: " + partitionCount + ", but " - + partitionSize + (partitionSize > 1 ? " have " : " has ") + "been found instead." - + "Consider either increasing the partition count of the topic or enabling " + - "`autoAddPartitions`"); - } + Set names = namesFutures.get(operationTimeout, TimeUnit.SECONDS); + if (names.contains(topicName)) { + // only consider minPartitionCount for resizing if autoAddPartitions is true + int effectivePartitionCount = this.configurationProperties.isAutoAddPartitions() + ? Math.max(this.configurationProperties.getMinPartitionCount(), partitionCount) + : partitionCount; + DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singletonList(topicName)); + KafkaFuture> topicDescriptionsFuture = describeTopicsResult.all(); + Map topicDescriptions = topicDescriptionsFuture.get(operationTimeout, TimeUnit.SECONDS); + TopicDescription topicDescription = topicDescriptions.get(topicName); + int partitionSize = topicDescription.partitions().size(); + if (partitionSize < effectivePartitionCount) { + if (this.configurationProperties.isAutoAddPartitions()) { + CreatePartitionsResult partitions = adminClient.createPartitions( + Collections.singletonMap(topicName, NewPartitions.increaseTo(effectivePartitionCount))); + partitions.all().get(operationTimeout, TimeUnit.SECONDS); + } + else if (tolerateLowerPartitionsOnBroker) { + logger.warn("The number of expected partitions was: " + partitionCount + ", but " + + partitionSize + (partitionSize > 1 ? " have " : " has ") + "been found instead." + + "There will be " + (effectivePartitionCount - partitionSize) + " idle consumers"); + } + else { + throw new ProvisioningException("The number of expected partitions was: " + partitionCount + ", but " + + partitionSize + (partitionSize > 1 ? " have " : " has ") + "been found instead." + + "Consider either increasing the partition count of the topic or enabling " + + "`autoAddPartitions`"); } } - else if (errorCode == ErrorMapping.UnknownTopicOrPartitionCode()) { - // always consider minPartitionCount for topic creation - final int effectivePartitionCount = Math.max(this.configurationProperties.getMinPartitionCount(), - partitionCount); + } + else if (!names.contains(topicName)) { + // always consider minPartitionCount for topic creation + final int effectivePartitionCount = Math.max(this.configurationProperties.getMinPartitionCount(), + partitionCount); + this.metadataRetryOperations.execute(context -> { - this.metadataRetryOperations.execute(context -> { - - try { - adminUtilsOperation.invokeCreateTopic(zkUtils, topicName, effectivePartitionCount, - configurationProperties.getReplicationFactor(), new Properties()); - } - catch (Exception e) { - String exceptionClass = e.getClass().getName(); - if (exceptionClass.equals("kafka.common.TopicExistsException") - || exceptionClass.equals("org.apache.kafka.common.errors.TopicExistsException")) { + NewTopic newTopic = new NewTopic(topicName, effectivePartitionCount, + (short) configurationProperties.getReplicationFactor()); + CreateTopicsResult createTopicsResult = adminClient.createTopics(Collections.singletonList(newTopic)); + try { + createTopicsResult.all().get(operationTimeout, TimeUnit.SECONDS); + } + catch (Exception e) { + if (e instanceof ExecutionException) { + String exceptionMessage = e.getMessage(); + if (exceptionMessage.contains("org.apache.kafka.common.errors.TopicExistsException")) { if (logger.isWarnEnabled()) { logger.warn("Attempt to create topic: " + topicName + ". Topic already exists."); } } - else { - throw e; - } } - return null; - }); - } - else { - throw new ProvisioningException("Error fetching Kafka topic metadata: ", - ErrorMapping.exceptionFor(errorCode)); - } - } - finally { - zkUtils.close(); + else { + logger.error("Failed to create topics", e.getCause()); + throw e.getCause(); + } + } + return null; + }); } } diff --git a/spring-cloud-stream-binder-kafka/pom.xml b/spring-cloud-stream-binder-kafka/pom.xml index 4fd4134e7..2c7fbd7f5 100644 --- a/spring-cloud-stream-binder-kafka/pom.xml +++ b/spring-cloud-stream-binder-kafka/pom.xml @@ -37,10 +37,6 @@ spring-cloud-stream-binder-test test - - org.apache.kafka - kafka_2.11 - org.apache.kafka kafka-clients @@ -59,12 +55,6 @@ spring-kafka-test test - - org.apache.kafka - kafka_2.11 - test - test - org.springframework.cloud spring-cloud-stream-binder-test 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 0c9db3b37..fe7979bec 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 @@ -609,26 +609,26 @@ public class KafkaMessageChannelBinder extends } } - public static class TopicInformation { + static class TopicInformation { private final String consumerGroup; private final Collection partitionInfos; - public TopicInformation(String consumerGroup, Collection partitionInfos) { + TopicInformation(String consumerGroup, Collection partitionInfos) { this.consumerGroup = consumerGroup; this.partitionInfos = partitionInfos; } - public String getConsumerGroup() { + String getConsumerGroup() { return consumerGroup; } - public boolean isConsumerTopic() { + boolean isConsumerTopic() { return consumerGroup != null; } - public Collection getPartitionInfos() { + Collection getPartitionInfos() { return partitionInfos; } @@ -645,7 +645,7 @@ public class KafkaMessageChannelBinder extends } @SuppressWarnings("unchecked") - public void sendToDlq(ConsumerRecord consumerRecord, Headers headers) { + void sendToDlq(ConsumerRecord consumerRecord, Headers headers) { K key = (K)consumerRecord.key(); V value = (V)consumerRecord.value(); ProducerRecord producerRecord = new ProducerRecord<>(this.dlqName, consumerRecord.partition(), 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 380c3b8ad..c11dfe585 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 @@ -27,16 +27,14 @@ import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.common.serialization.ByteArrayDeserializer; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; 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.KafkaBinderMetrics; import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder; -import org.springframework.cloud.stream.binder.kafka.admin.AdminUtilsOperation; -import org.springframework.cloud.stream.binder.kafka.admin.KafkaAdminUtilsOperation; import org.springframework.cloud.stream.binder.kafka.properties.JaasLoginModuleConfiguration; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBindingProperties; @@ -81,12 +79,12 @@ public class KafkaBinderConfiguration { @Autowired private ApplicationContext context; - @Autowired(required = false) - private AdminUtilsOperation adminUtilsOperation; + @Autowired + private KafkaProperties kafkaProperties; @Bean KafkaTopicProvisioner provisioningProvider() { - return new KafkaTopicProvisioner(this.configurationProperties, this.adminUtilsOperation); + return new KafkaTopicProvisioner(this.configurationProperties, this.kafkaProperties); } @Bean @@ -127,12 +125,6 @@ public class KafkaBinderConfiguration { return new KafkaBinderMetrics(kafkaMessageChannelBinder, configurationProperties); } - @Bean(name = "adminUtilsOperation") - @ConditionalOnClass(name = "kafka.admin.AdminUtils") - public AdminUtilsOperation kafka10AdminUtilsOperation() { - return new KafkaAdminUtilsOperation(); - } - @Bean public KafkaJaasLoginModuleInitializer jaasInitializer() throws IOException { return new KafkaJaasLoginModuleInitializer(); diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationPropertiesTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationPropertiesTest.java index ee40aaac7..b0c57421e 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationPropertiesTest.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationPropertiesTest.java @@ -28,6 +28,7 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; @@ -47,7 +48,8 @@ import static org.junit.Assert.assertTrue; * @author Ilayaperumal Gopinathan */ @RunWith(SpringJUnit4ClassRunner.class) -@SpringBootTest(classes = { KafkaBinderConfiguration.class, KafkaBinderConfigurationPropertiesTest.class }) +@SpringBootTest(classes = { KafkaBinderConfiguration.class, KafkaAutoConfiguration.class, + KafkaBinderConfigurationPropertiesTest.class }) @TestPropertySource(locations = "classpath:binder-config.properties") public class KafkaBinderConfigurationPropertiesTest { diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationTest.java index 23ba781a9..9a0f633ff 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationTest.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationTest.java @@ -22,6 +22,7 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration; import org.springframework.kafka.support.ProducerListener; @@ -34,7 +35,9 @@ import static org.junit.Assert.assertNotNull; * @author Ilayaperumal Gopinathan */ @RunWith(SpringJUnit4ClassRunner.class) -@SpringBootTest(classes = { KafkaBinderConfiguration.class, KafkaBinderConfigurationTest.class }) +@SpringBootTest(classes = { KafkaBinderConfiguration.class, + KafkaAutoConfiguration.class, + KafkaBinderConfigurationTest.class }) public class KafkaBinderConfigurationTest { @Autowired 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 06d506479..ef2ed0fec 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 @@ -26,7 +26,6 @@ import java.util.Iterator; 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; @@ -34,11 +33,12 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import com.fasterxml.jackson.databind.ObjectMapper; - -import kafka.utils.ZKStringSerializer$; -import kafka.utils.ZkUtils; - -import org.I0Itec.zkclient.ZkClient; +import org.apache.kafka.clients.admin.AdminClient; +import org.apache.kafka.clients.admin.AdminClientConfig; +import org.apache.kafka.clients.admin.CreateTopicsResult; +import org.apache.kafka.clients.admin.DescribeTopicsResult; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.admin.TopicDescription; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; @@ -46,6 +46,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.KafkaFuture; import org.apache.kafka.common.serialization.ByteArrayDeserializer; import org.apache.kafka.common.serialization.ByteArraySerializer; import org.apache.kafka.common.serialization.Deserializer; @@ -61,6 +62,7 @@ import org.junit.Test; import org.junit.rules.ExpectedException; import org.springframework.beans.DirectFieldAccessor; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.Binding; @@ -71,11 +73,10 @@ 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.admin.KafkaAdminUtilsOperation; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; -import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties.StandardHeaders; import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; +import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; import org.springframework.cloud.stream.binder.kafka.utils.KafkaTopicUtils; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.provisioning.ProvisioningException; @@ -111,9 +112,6 @@ import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; -import org.springframework.retry.backoff.FixedBackOffPolicy; -import org.springframework.retry.policy.SimpleRetryPolicy; -import org.springframework.retry.support.RetryTemplate; import org.springframework.util.Assert; import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; @@ -134,6 +132,8 @@ import static org.mockito.Mockito.mock; public class KafkaBinderTests extends PartitionCapableBinderTests, ExtendedProducerProperties> { + private static final int DEFAULT_OPERATION_TIMEOUT = 30; + @Rule public ExpectedException expectedProvisioningException = ExpectedException.none(); @@ -144,7 +144,7 @@ public class KafkaBinderTests extends private KafkaTestBinder binder; - private final KafkaAdminUtilsOperation adminUtilsOperation = new KafkaAdminUtilsOperation(); + private AdminClient adminClient; @Override protected ExtendedConsumerProperties createConsumerProperties() { @@ -173,11 +173,30 @@ public class KafkaBinderTests extends protected KafkaTestBinder getBinder() { if (binder == null) { KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); - binder = new KafkaTestBinder(binderConfiguration); + KafkaTopicProvisioner kafkaTopicProvisioner = new KafkaTopicProvisioner(binderConfiguration, new KafkaProperties()); + try { + kafkaTopicProvisioner.afterPropertiesSet(); + } + catch (Exception e) { + throw new RuntimeException(e); + } + binder = new KafkaTestBinder(binderConfiguration, kafkaTopicProvisioner); } return binder; } + private Binder getBinder(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) { + KafkaTopicProvisioner provisioningProvider = + new KafkaTopicProvisioner(kafkaBinderConfigurationProperties, new KafkaProperties()); + try { + provisioningProvider.afterPropertiesSet(); + } + catch (Exception e) { + throw new RuntimeException(e); + } + return new KafkaTestBinder(kafkaBinderConfigurationProperties, provisioningProvider); + } + private KafkaBinderConfigurationProperties createConfigurationProperties() { KafkaBinderConfigurationProperties binderConfiguration = new KafkaBinderConfigurationProperties(); BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses(); @@ -195,36 +214,44 @@ public class KafkaBinderTests extends return consumerFactory().createConsumer().partitionsFor(topic).size(); } - private ZkUtils getZkUtils(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) { - final ZkClient zkClient = new ZkClient(kafkaBinderConfigurationProperties.getZkConnectionString(), - kafkaBinderConfigurationProperties.getZkSessionTimeout(), kafkaBinderConfigurationProperties.getZkConnectionTimeout(), - ZKStringSerializer$.MODULE$); + private void invokeCreateTopic(String topic, int partitions, int replicationFactor) throws Throwable { - return new ZkUtils(zkClient, null, false); - } - - private void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions, int replicationFactor, Properties topicConfig) { - adminUtilsOperation.invokeCreateTopic(zkUtils, topic, partitions, replicationFactor, new Properties()); - } - - private int invokePartitionSize(String topic, ZkUtils zkUtils) { - return adminUtilsOperation.partitionSize(topic, zkUtils); + NewTopic newTopic = new NewTopic(topic, partitions, + (short) replicationFactor); + CreateTopicsResult topics = adminClient.createTopics(Collections.singletonList(newTopic)); + topics.all().get(DEFAULT_OPERATION_TIMEOUT, TimeUnit.SECONDS); } private String getKafkaOffsetHeaderKey() { return KafkaHeaders.OFFSET; } - private Binder getBinder(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) { - return new KafkaTestBinder(kafkaBinderConfigurationProperties); - } - @Before public void init() { String multiplier = System.getenv("KAFKA_TIMEOUT_MULTIPLIER"); if (multiplier != null) { timeoutMultiplier = Double.parseDouble(multiplier); } + + BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses(); + List bAddresses = new ArrayList<>(); + for (BrokerAddress bAddress : brokerAddresses) { + bAddresses.add(bAddress.toString()); + } + String[] foo = new String[bAddresses.size()]; + + Map adminConfigs = new HashMap<>(); + adminConfigs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bAddresses.toArray(foo)[0]); + adminClient = AdminClient.create(adminConfigs); + } + + private int invokePartitionSize(String topic) throws Throwable { + + DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singletonList(topic)); + KafkaFuture> all = describeTopicsResult.all(); + Map stringTopicDescriptionMap = all.get(DEFAULT_OPERATION_TIMEOUT, TimeUnit.SECONDS); + TopicDescription topicDescription = stringTopicDescriptionMap.get(topic); + return topicDescription.partitions().size(); } @Override @@ -254,7 +281,7 @@ public class KafkaBinderTests extends return new DefaultKafkaConsumerFactory<>(props, keyDecoder, valueDecoder); } - @SuppressWarnings({ "rawtypes", "unchecked" }) + @SuppressWarnings({"rawtypes", "unchecked"}) @Test public void testTrustedPackages() throws Exception { Binder binder = getBinder(); @@ -320,7 +347,7 @@ public class KafkaBinderTests extends Binding producerBinding = binder.bindProducer("bar.0", moduleOutputChannel, producerBindingProperties.getProducer()); - consumerProperties.getExtension().setTrustedPackages(new String[] {"org.springframework.util"}); + consumerProperties.getExtension().setTrustedPackages(new String[]{"org.springframework.util"}); Binding consumerBinding = binder.bindConsumer("bar.0", "testSendAndReceiveNoOriginalContentType", moduleInputChannel, consumerProperties); @@ -552,7 +579,7 @@ public class KafkaBinderTests extends ExtendedProducerProperties producerProperties = createProducerProperties(); producerProperties.setPartitionCount(2); - producerProperties.getExtension().setHeaderPatterns(new String[] { MessageHeaders.CONTENT_TYPE }); + producerProperties.getExtension().setHeaderPatterns(new String[]{MessageHeaders.CONTENT_TYPE}); DirectChannel moduleOutputChannel = createBindableChannel("output", createProducerBindingProperties(producerProperties)); @@ -824,25 +851,6 @@ public class KafkaBinderTests extends producerBinding.unbind(); } - @Test - @SuppressWarnings("unchecked") - public void testAutoCreateTopicsEnabledSucceeds() throws Exception { - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - configurationProperties.setAutoCreateTopics(true); - Binder binder = getBinder(configurationProperties); - RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); - metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); - FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); - backOffPolicy.setBackOffPeriod(1000); - metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - String testTopicName = "nonexisting" + System.currentTimeMillis(); - DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); - - Binding binding = binder.bindConsumer(testTopicName, "test", moduleInputChannel, consumerProperties); - binding.unbind(); - } - @Test(expected = IllegalArgumentException.class) public void testValidateKafkaTopicName() { KafkaTopicUtils.validateTopicName("foo:bar"); @@ -852,9 +860,9 @@ public class KafkaBinderTests extends @SuppressWarnings("unchecked") //TODO: This test needs to be rethought - sending byte[] without explicit content type - yet being converted by the json converter public void testCompression() throws Exception { - final KafkaProducerProperties.CompressionType[] codecs = new KafkaProducerProperties.CompressionType[] { + final KafkaProducerProperties.CompressionType[] codecs = new KafkaProducerProperties.CompressionType[]{ KafkaProducerProperties.CompressionType.none, KafkaProducerProperties.CompressionType.gzip, - KafkaProducerProperties.CompressionType.snappy }; + KafkaProducerProperties.CompressionType.snappy}; byte[] testPayload = new byte[2048]; Arrays.fill(testPayload, (byte) 65); Binder binder = getBinder(); @@ -899,206 +907,6 @@ public class KafkaBinderTests extends } } - @Test - @SuppressWarnings("unchecked") - public void testCustomPartitionCountOverridesDefaultIfLarger() throws Exception { - byte[] testPayload = new byte[2048]; - Arrays.fill(testPayload, (byte) 65); - KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); - binderConfiguration.setMinPartitionCount(10); - Binder binder = getBinder(binderConfiguration); - QueueChannel moduleInputChannel = new QueueChannel(); - ExtendedProducerProperties producerProperties = createProducerProperties(); - producerProperties.setPartitionCount(10); - - DirectChannel moduleOutputChannel = createBindableChannel("output", - createProducerBindingProperties(producerProperties)); - - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - long uniqueBindingId = System.currentTimeMillis(); - Binding producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0", - moduleOutputChannel, producerProperties); - Binding consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null, - moduleInputChannel, consumerProperties); - Message message = org.springframework.integration.support.MessageBuilder.withPayload(testPayload) - .build(); - // Let the consumer actually bind to the producer before sending a msg - binderBindUnbindLatency(); - moduleOutputChannel.send(message); - Message inbound = receive(moduleInputChannel); - assertThat(inbound).isNotNull(); - assertThat((byte[]) inbound.getPayload()).containsExactly(testPayload); - - assertThat(partitionSize("foo" + uniqueBindingId + ".0")).isEqualTo(10); - producerBinding.unbind(); - consumerBinding.unbind(); - } - - @Test - @SuppressWarnings("unchecked") - public void testCustomPartitionCountDoesNotOverridePartitioningIfSmaller() throws Exception { - - byte[] testPayload = new byte[2048]; - Arrays.fill(testPayload, (byte) 65); - KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); - binderConfiguration.setMinPartitionCount(6); - Binder binder = getBinder(binderConfiguration); - 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 - binderBindUnbindLatency(); - moduleOutputChannel.send(message); - Message inbound = receive(moduleInputChannel); - assertThat(inbound).isNotNull(); - assertThat((byte[]) inbound.getPayload()).containsExactly(testPayload); - - assertThat(partitionSize("foo" + uniqueBindingId + ".0")).isEqualTo(6); - producerBinding.unbind(); - consumerBinding.unbind(); - } - - @Test - @SuppressWarnings("unchecked") - public void testDynamicKeyExpression() throws Exception { - Binder binder = getBinder(createConfigurationProperties()); - QueueChannel moduleInputChannel = new QueueChannel(); - ExtendedProducerProperties producerProperties = createProducerProperties(); - producerProperties.getExtension().getConfiguration().put("key.serializer", StringSerializer.class.getName()); - producerProperties.getExtension().setMessageKeyExpression(spelExpressionParser.parseExpression("headers.key")); - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - String uniqueBindingId = UUID.randomUUID().toString(); - 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 = MessageBuilder.withPayload("somePayload").setHeader("key", "myDynamicKey").build(); - // Let the consumer actually bind to the producer before sending a msg - binderBindUnbindLatency(); - moduleOutputChannel.send(message); - Message inbound = receive(moduleInputChannel); - assertThat(inbound).isNotNull(); - String receivedKey = new String(inbound.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, byte[].class)); - assertThat(receivedKey).isEqualTo("myDynamicKey"); - producerBinding.unbind(); - consumerBinding.unbind(); - } - - @Test - @SuppressWarnings("unchecked") - public void testCustomPartitionCountOverridesPartitioningIfLarger() throws Exception { - - byte[] testPayload = new byte[2048]; - Arrays.fill(testPayload, (byte) 65); - KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); - binderConfiguration.setMinPartitionCount(4); - Binder binder = getBinder(binderConfiguration); - - 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", - moduleOutputChannel, producerProperties); - Binding consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null, - moduleInputChannel, consumerProperties); - Message message = org.springframework.integration.support.MessageBuilder.withPayload(testPayload) - .build(); - // Let the consumer actually bind to the producer before sending a msg - binderBindUnbindLatency(); - moduleOutputChannel.send(message); - Message inbound = receive(moduleInputChannel); - assertThat(inbound).isNotNull(); - assertThat((byte[]) inbound.getPayload()).containsExactly(testPayload); - assertThat(partitionSize("foo" + uniqueBindingId + ".0")).isEqualTo(5); - producerBinding.unbind(); - consumerBinding.unbind(); - } - - @Test - @SuppressWarnings("unchecked") - public void testDefaultConsumerStartsAtEarliest() throws Exception { - Binder binder = getBinder(createConfigurationProperties()); - GenericApplicationContext context = new GenericApplicationContext(); - context.refresh(); - - BindingProperties producerBindingProperties = createProducerBindingProperties(createProducerProperties()); - DirectChannel output = createBindableChannel("output", producerBindingProperties); - //QueueChannel moduleInputChannel = new QueueChannel(); - - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - consumerProperties.getExtension().setAutoRebalanceEnabled(false); - - DirectChannel input1 = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); - - String testTopicName = UUID.randomUUID().toString(); - Binding producerBinding = binder.bindProducer(testTopicName, output, - createProducerProperties()); - String testPayload1 = "foo-" + UUID.randomUUID().toString(); - output.send(new GenericMessage<>(testPayload1.getBytes())); - - Binding consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, - consumerProperties); - - CountDownLatch latch = new CountDownLatch(1); - AtomicReference> inboundMessageRef1 = new AtomicReference<>(); - MessageHandler messageHandler = message1 -> { - try { - inboundMessageRef1.set((Message) message1); - } - finally { - latch.countDown(); - } - }; - input1.subscribe(messageHandler); - Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); - - - assertThat(inboundMessageRef1.get()).isNotNull(); - assertThat(inboundMessageRef1.get().getPayload()).isEqualTo(testPayload1); - - String testPayload2 = "foo-" + UUID.randomUUID().toString(); - input1.unsubscribe(messageHandler); - output.send(new GenericMessage<>(testPayload2.getBytes())); - - CountDownLatch latch1 = new CountDownLatch(1); - AtomicReference> inboundMessageRef2 = new AtomicReference<>(); - input1.subscribe(message1 -> { - try { - inboundMessageRef2.set((Message) message1); - } - finally { - latch1.countDown(); - } - }); - Assert.isTrue(latch1.await(5, TimeUnit.SECONDS), "Failed to receive message"); - - assertThat(inboundMessageRef2.get()).isNotNull(); - assertThat(inboundMessageRef2.get().getPayload()).isEqualTo(testPayload2); - - producerBinding.unbind(); - consumerBinding.unbind(); - } - @Test @SuppressWarnings("unchecked") public void testEarliest() throws Exception { @@ -1109,7 +917,6 @@ public class KafkaBinderTests extends Binder binder = getBinder(); BindingProperties producerBindingProperties = createProducerBindingProperties(createProducerProperties()); DirectChannel output = createBindableChannel("output", producerBindingProperties); - //QueueChannel moduleInputChannel = new QueueChannel(); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); consumerProperties.getExtension().setAutoRebalanceEnabled(false); @@ -1169,95 +976,6 @@ public class KafkaBinderTests extends } } - @Test - @SuppressWarnings("unchecked") - public void testResume() throws Exception { - Binding producerBinding = null; - Binding consumerBinding = null; - - try { - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - Binder binder = getBinder(configurationProperties); - - BindingProperties producerBindingProperties = createProducerBindingProperties(createProducerProperties()); - DirectChannel output = createBindableChannel("output", producerBindingProperties); - - DirectChannel input1 = createBindableChannel("input", createConsumerBindingProperties(createConsumerProperties())); - - - String testTopicName = UUID.randomUUID().toString(); - producerBinding = binder.bindProducer(testTopicName, output, - producerBindingProperties.getProducer()); - String testPayload1 = "foo1-" + UUID.randomUUID().toString(); - output.send(new GenericMessage<>(testPayload1)); - ExtendedConsumerProperties firstConsumerProperties = createConsumerProperties(); - consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, - firstConsumerProperties); - CountDownLatch latch = new CountDownLatch(1); - AtomicReference> inboundMessageRef1 = new AtomicReference<>(); - MessageHandler messageHandler = message1 -> { - try { - inboundMessageRef1.set((Message) message1); - } - finally { - latch.countDown(); - } - }; - input1.subscribe(messageHandler); - Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); - - assertThat(inboundMessageRef1.get()).isNotNull(); - assertThat(inboundMessageRef1.get().getPayload()).isNotNull(); - String testPayload2 = "foo2-" + UUID.randomUUID().toString(); - output.send(new GenericMessage<>(testPayload2.getBytes())); - input1.unsubscribe(messageHandler); - CountDownLatch latch1 = new CountDownLatch(1); - AtomicReference> inboundMessageRef2 = new AtomicReference<>(); - MessageHandler messageHandler1 = message1 -> { - try { - inboundMessageRef2.set((Message) message1); - } - finally { - latch1.countDown(); - } - }; - input1.subscribe(messageHandler1); - Assert.isTrue(latch1.await(5, TimeUnit.SECONDS), "Failed to receive message"); - assertThat(inboundMessageRef2.get()).isNotNull(); - assertThat(inboundMessageRef2.get().getPayload()).isNotNull(); - consumerBinding.unbind(); - - Thread.sleep(2000); - String testPayload3 = "foo3-" + UUID.randomUUID().toString(); - output.send(new GenericMessage<>(testPayload3.getBytes())); - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, consumerProperties); - input1.unsubscribe(messageHandler1); - CountDownLatch latch2 = new CountDownLatch(1); - AtomicReference> inboundMessageRef3 = new AtomicReference<>(); - MessageHandler messageHandler2 = message1 -> { - try { - inboundMessageRef3.set((Message) message1); - } - finally { - latch2.countDown(); - } - }; - input1.subscribe(messageHandler2); - Assert.isTrue(latch2.await(5, TimeUnit.SECONDS), "Failed to receive message"); - assertThat(inboundMessageRef3.get()).isNotNull(); - assertThat(new String(inboundMessageRef3.get().getPayload())).isEqualTo(testPayload3); - } - finally { - if (consumerBinding != null) { - consumerBinding.unbind(); - } - if (producerBinding != null) { - producerBinding.unbind(); - } - } - } - @Test @Override @SuppressWarnings("unchecked") @@ -1497,9 +1215,9 @@ public class KafkaBinderTests extends ObjectMapper om = new ObjectMapper(); if (usesExplicitRouting()) { - assertThat(om.readValue((byte[])receive0.getPayload(), Integer.class)).isEqualTo(0); - assertThat(om.readValue((byte[])receive1.getPayload(), Integer.class)).isEqualTo(1); - assertThat(om.readValue((byte[])receive2.getPayload(), Integer.class)).isEqualTo(2); + assertThat(om.readValue((byte[]) receive0.getPayload(), Integer.class)).isEqualTo(0); + assertThat(om.readValue((byte[]) receive1.getPayload(), Integer.class)).isEqualTo(1); + assertThat(om.readValue((byte[]) receive2.getPayload(), Integer.class)).isEqualTo(2); assertThat(receive2).has(correlationHeadersForPayload2); } else { @@ -1529,20 +1247,12 @@ public class KafkaBinderTests extends @Test @Override - @SuppressWarnings({ "unchecked", "rawtypes" }) + @SuppressWarnings({"unchecked", "rawtypes"}) public void testPartitionedModuleJava() throws Exception { Binder binder = getBinder(); KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - final ZkClient zkClient; - zkClient = new ZkClient(configurationProperties.getZkConnectionString(), - configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), - ZKStringSerializer$.MODULE$); - - final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); - invokeCreateTopic(zkUtils, "partJ.0", 8, 1, new Properties()); - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); consumerProperties.setConcurrency(2); consumerProperties.setInstanceCount(4); @@ -1593,10 +1303,10 @@ public class KafkaBinderTests extends assertThat(receive3).isNotNull(); ObjectMapper om = new ObjectMapper(); - assertThat(om.readValue((byte[])receive0.getPayload(), Integer.class)).isEqualTo(0); - assertThat(om.readValue((byte[])receive1.getPayload(), Integer.class)).isEqualTo(1); - assertThat(om.readValue((byte[])receive2.getPayload(), Integer.class)).isEqualTo(2); - assertThat(om.readValue((byte[])receive3.getPayload(), Integer.class)).isEqualTo(3); + assertThat(om.readValue((byte[]) receive0.getPayload(), Integer.class)).isEqualTo(0); + assertThat(om.readValue((byte[]) receive1.getPayload(), Integer.class)).isEqualTo(1); + assertThat(om.readValue((byte[]) receive2.getPayload(), Integer.class)).isEqualTo(2); + assertThat(om.readValue((byte[]) receive3.getPayload(), Integer.class)).isEqualTo(3); input0Binding.unbind(); input1Binding.unbind(); @@ -1663,513 +1373,6 @@ public class KafkaBinderTests extends binding2.unbind(); } - @Test - @SuppressWarnings("unchecked") - public void testSyncProducerMetadata() throws Exception { - Binder binder = getBinder(createConfigurationProperties()); - DirectChannel output = new DirectChannel(); - String testTopicName = UUID.randomUUID().toString(); - ExtendedProducerProperties properties = createProducerProperties(); - properties.getExtension().setSync(true); - Binding producerBinding = binder.bindProducer(testTopicName, output, properties); - DirectFieldAccessor accessor = new DirectFieldAccessor(extractEndpoint(producerBinding)); - KafkaProducerMessageHandler wrappedInstance = (KafkaProducerMessageHandler) accessor.getWrappedInstance(); - assertThat(new DirectFieldAccessor(wrappedInstance).getPropertyValue("sync").equals(Boolean.TRUE)) - .withFailMessage("Kafka Sync Producer should have been enabled."); - producerBinding.unbind(); - } - - @Test - @SuppressWarnings("unchecked") - public void testAutoCreateTopicsDisabledOnBinderStillWorksAsLongAsBrokerCreatesTopic() throws Exception { - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - configurationProperties.setAutoCreateTopics(false); - Binder binder = getBinder(configurationProperties); - RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); - metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); - FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); - backOffPolicy.setBackOffPeriod(1000); - metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); - - BindingProperties producerBindingProperties = createProducerBindingProperties(createProducerProperties()); - DirectChannel output = createBindableChannel("output", producerBindingProperties); - //QueueChannel moduleInputChannel = new QueueChannel(); - - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - - DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); - - String testTopicName = "createdByBroker-" + System.currentTimeMillis(); - - Binding producerBinding = binder.bindProducer(testTopicName, output, - producerBindingProperties.getProducer()); - - String testPayload = "foo1-" + UUID.randomUUID().toString(); - output.send(new GenericMessage<>(testPayload)); - - Binding consumerBinding = binder.bindConsumer(testTopicName, "test", input, consumerProperties); - CountDownLatch latch = new CountDownLatch(1); - AtomicReference> inboundMessageRef = new AtomicReference<>(); - input.subscribe(message1 -> { - try { - inboundMessageRef.set((Message) message1); - } - finally { - latch.countDown(); - } - }); - Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); - - assertThat(inboundMessageRef.get()).isNotNull(); - assertThat(inboundMessageRef.get().getPayload()).isEqualTo(testPayload); - - producerBinding.unbind(); - consumerBinding.unbind(); - } - - @Test - @SuppressWarnings("unchecked") - public void testAutoConfigureTopicsDisabledSucceedsIfTopicExisting() throws Exception { - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - - final ZkUtils zkUtils = getZkUtils(configurationProperties); - - String testTopicName = "existing" + System.currentTimeMillis(); - invokeCreateTopic(zkUtils, testTopicName, 5, 1, new Properties()); - - configurationProperties.setAutoCreateTopics(false); - Binder binder = getBinder(configurationProperties); - - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - - DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); - Binding binding = binder.bindConsumer(testTopicName, "test", input, consumerProperties); - binding.unbind(); - } - - @Test - @SuppressWarnings("unchecked") - public void testPartitionCountIncreasedIfAutoAddPartitionsSet() throws Exception { - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - - final ZkUtils zkUtils = getZkUtils(configurationProperties); - - String testTopicName = "existing" + System.currentTimeMillis(); - invokeCreateTopic(zkUtils, testTopicName, 6, 1, new Properties()); - configurationProperties.setMinPartitionCount(6); - configurationProperties.setAutoAddPartitions(true); - Binder binder = getBinder(configurationProperties); - RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); - metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); - FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); - backOffPolicy.setBackOffPeriod(1000); - metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); - - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - - DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); - - Binding binding = binder.bindConsumer(testTopicName, "test", input, consumerProperties); - binding.unbind(); - assertThat(invokePartitionSize(testTopicName, zkUtils)).isEqualTo(6); - } - - @Test - @SuppressWarnings("unchecked") - public void testAutoAddPartitionsDisabledSucceedsIfTopicUnderPartitionedAndAutoRebalanceEnabled() throws Exception { - 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(); - invokeCreateTopic(zkUtils, testTopicName, 1, 1, new Properties()); - configurationProperties.setAutoAddPartitions(false); - Binder binder = getBinder(configurationProperties); - GenericApplicationContext context = new GenericApplicationContext(); - context.refresh(); - - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - - DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); - - // this consumer must consume from partition 2 - consumerProperties.setInstanceCount(3); - consumerProperties.setInstanceIndex(2); - Binding binding = binder.bindConsumer(testTopicName, "test", input, consumerProperties); - binding.unbind(); - assertThat(invokePartitionSize(testTopicName, zkUtils)).isEqualTo(1); - } - - @Test - @SuppressWarnings("unchecked") - public void testAutoAddPartitionsDisabledFailsIfTopicUnderPartitionedAndAutoRebalanceDisabled() throws Exception { - 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(); - invokeCreateTopic(zkUtils, testTopicName, 1, 1, new Properties()); - configurationProperties.setAutoAddPartitions(false); - Binder binder = getBinder(configurationProperties); - GenericApplicationContext context = new GenericApplicationContext(); - context.refresh(); - - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - DirectChannel output = createBindableChannel("output", createConsumerBindingProperties(consumerProperties)); - // this consumer must consume from partition 2 - consumerProperties.setInstanceCount(3); - consumerProperties.setInstanceIndex(2); - consumerProperties.getExtension().setAutoRebalanceEnabled(false); - expectedProvisioningException.expect(ProvisioningException.class); - expectedProvisioningException - .expectMessage("The number of expected partitions was: 3, but 1 has been found instead"); - Binding binding = binder.bindConsumer(testTopicName, "test", output, consumerProperties); - if (binding != null) { - binding.unbind(); - } - } - - @Test - @SuppressWarnings("unchecked") - public void testAutoAddPartitionsDisabledSucceedsIfTopicPartitionedCorrectly() throws Exception { - Binding binding = null; - try { - 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(); - invokeCreateTopic(zkUtils, testTopicName, 6, 1, new Properties()); - configurationProperties.setAutoAddPartitions(false); - Binder binder = getBinder(configurationProperties); - RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); - metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); - FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); - backOffPolicy.setBackOffPeriod(1000); - metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); - - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - - DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); - - // this consumer must consume from partition 2 - consumerProperties.setInstanceCount(3); - consumerProperties.setInstanceIndex(2); - consumerProperties.getExtension().setAutoRebalanceEnabled(false); - - binding = binder.bindConsumer(testTopicName, "test-x", input, consumerProperties); - - TopicPartitionInitialOffset[] listenedPartitions = TestUtils.getPropertyValue(binding, - "lifecycle.messageListenerContainer.containerProperties.topicPartitions", - TopicPartitionInitialOffset[].class); - assertThat(listenedPartitions).hasSize(2); - assertThat(listenedPartitions).contains(new TopicPartitionInitialOffset(testTopicName, 2), - new TopicPartitionInitialOffset(testTopicName, 5)); - int partitions = invokePartitionSize(testTopicName, zkUtils); - assertThat(partitions).isEqualTo(6); - } - finally { - if (binding != null) { - binding.unbind(); - } - } - } - - @Test - @SuppressWarnings("unchecked") - public void testPartitionCountNotReduced() throws Exception { - String testTopicName = "existing" + System.currentTimeMillis(); - - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - - final ZkClient zkClient; - zkClient = new ZkClient(configurationProperties.getZkConnectionString(), - configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), - ZKStringSerializer$.MODULE$); - - final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); - invokeCreateTopic(zkUtils, testTopicName, 6, 1, new Properties()); - configurationProperties.setAutoAddPartitions(true); - Binder binder = getBinder(configurationProperties); - GenericApplicationContext context = new GenericApplicationContext(); - context.refresh(); - RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); - metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); - FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); - backOffPolicy.setBackOffPeriod(1000); - metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); - - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); - - Binding binding = binder.bindConsumer(testTopicName, "test", input, consumerProperties); - binding.unbind(); - - assertThat(partitionSize(testTopicName)).isEqualTo(6); - } - - @Test - @SuppressWarnings("unchecked") - public void testConsumerDefaultDeserializer() throws Exception { - Binding binding = null; - try { - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - final ZkUtils zkUtils = getZkUtils(configurationProperties); - String testTopicName = "existing" + System.currentTimeMillis(); - invokeCreateTopic(zkUtils, testTopicName, 5, 1, new Properties()); - configurationProperties.setAutoCreateTopics(false); - Binder binder = getBinder(configurationProperties); - - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); - - binding = binder.bindConsumer(testTopicName, "test", input, consumerProperties); - DirectFieldAccessor consumerAccessor = new DirectFieldAccessor(getKafkaConsumer(binding)); - assertTrue(consumerAccessor.getPropertyValue("keyDeserializer") instanceof ByteArrayDeserializer); - assertTrue(consumerAccessor.getPropertyValue("valueDeserializer") instanceof ByteArrayDeserializer); - } - finally { - if (binding != null) { - binding.unbind(); - } - } - } - - @Test - @SuppressWarnings("unchecked") - public void testConsumerCustomDeserializer() throws Exception { - Binding binding = null; - try { - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - Map propertiesToOverride = configurationProperties.getConfiguration(); - propertiesToOverride.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); - propertiesToOverride.put("value.deserializer", "org.apache.kafka.common.serialization.LongDeserializer"); - configurationProperties.setConfiguration(propertiesToOverride); - final ZkUtils zkUtils = getZkUtils(configurationProperties); - String testTopicName = "existing" + System.currentTimeMillis(); - invokeCreateTopic(zkUtils, testTopicName, 5, 1, new Properties()); - configurationProperties.setAutoCreateTopics(false); - Binder binder = getBinder(configurationProperties); - - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); - - binding = binder.bindConsumer(testTopicName, "test", input, consumerProperties); - DirectFieldAccessor consumerAccessor = new DirectFieldAccessor(getKafkaConsumer(binding)); - assertTrue("Expected StringDeserializer as a custom key deserializer", - consumerAccessor.getPropertyValue("keyDeserializer") instanceof StringDeserializer); - assertTrue("Expected LongDeserializer as a custom value deserializer", - consumerAccessor.getPropertyValue("valueDeserializer") instanceof LongDeserializer); - } - finally { - if (binding != null) { - binding.unbind(); - } - } - } - - private KafkaConsumer getKafkaConsumer(Binding binding) { - DirectFieldAccessor bindingAccessor = new DirectFieldAccessor(binding); - KafkaMessageDrivenChannelAdapter adapter = (KafkaMessageDrivenChannelAdapter) bindingAccessor - .getPropertyValue("lifecycle"); - DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter); - ConcurrentMessageListenerContainer messageListenerContainer = - (ConcurrentMessageListenerContainer) adapterAccessor.getPropertyValue("messageListenerContainer"); - DirectFieldAccessor containerAccessor = new DirectFieldAccessor(messageListenerContainer); - DefaultKafkaConsumerFactory consumerFactory = (DefaultKafkaConsumerFactory) containerAccessor - .getPropertyValue("consumerFactory"); - return (KafkaConsumer) consumerFactory.createConsumer(); - } - - @Test - @SuppressWarnings({ "unchecked", "rawtypes" }) - public void testNativeSerializationWithCustomSerializerDeserializer() throws Exception { - Binding producerBinding = null; - Binding consumerBinding = null; - try { - Integer testPayload = 10; - Message message = MessageBuilder.withPayload(testPayload).build(); - SubscribableChannel moduleOutputChannel = new DirectChannel(); - String testTopicName = "existing" + System.currentTimeMillis(); - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - final ZkClient zkClient; - zkClient = new ZkClient(configurationProperties.getZkConnectionString(), - configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), - ZKStringSerializer$.MODULE$); - final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); - invokeCreateTopic(zkUtils, testTopicName, 6, 1, new Properties()); - configurationProperties.setAutoAddPartitions(true); - Binder binder = getBinder(configurationProperties); - QueueChannel moduleInputChannel = new QueueChannel(); - ExtendedProducerProperties producerProperties = createProducerProperties(); - producerProperties.setUseNativeEncoding(true); - producerProperties.getExtension().getConfiguration().put("value.serializer", - "org.apache.kafka.common.serialization.IntegerSerializer"); - producerBinding = binder.bindProducer(testTopicName, moduleOutputChannel, producerProperties); - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - consumerProperties.getExtension().setAutoRebalanceEnabled(false); - consumerProperties.getExtension().getConfiguration().put("value.deserializer", - "org.apache.kafka.common.serialization.IntegerDeserializer"); - consumerProperties.getExtension().setStandardHeaders(StandardHeaders.both); - consumerBinding = binder.bindConsumer(testTopicName, "test", moduleInputChannel, consumerProperties); - // Let the consumer actually bind to the producer before sending a msg - binderBindUnbindLatency(); - moduleOutputChannel.send(message); - Message inbound = receive(moduleInputChannel, 500); - assertThat(inbound).isNotNull(); - assertThat(inbound.getPayload()).isEqualTo(10); - assertThat(inbound.getHeaders()).doesNotContainKey("contentType"); - assertThat(inbound.getHeaders().getId()).isNotNull(); - assertThat(inbound.getHeaders().getTimestamp()).isNotNull(); - } - finally { - if (producerBinding != null) { - producerBinding.unbind(); - } - if (consumerBinding != null) { - consumerBinding.unbind(); - } - } - } - - @Test - @SuppressWarnings({ "unchecked", "rawtypes" }) - public void testNativeSerializationWithCustomSerializerDeserializerBytesPayload() throws Exception { - Binding producerBinding = null; - Binding consumerBinding = null; - try { - byte[] testPayload = new byte[1]; - Message message = MessageBuilder.withPayload(testPayload) - .setHeader(MessageHeaders.CONTENT_TYPE, "something/funky") - .build(); - SubscribableChannel moduleOutputChannel = new DirectChannel(); - String testTopicName = "existing" + System.currentTimeMillis(); - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - final ZkClient zkClient; - zkClient = new ZkClient(configurationProperties.getZkConnectionString(), - configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), - ZKStringSerializer$.MODULE$); - final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); - invokeCreateTopic(zkUtils, testTopicName, 1, 1, new Properties()); - configurationProperties.setAutoAddPartitions(true); - Binder binder = getBinder(configurationProperties); - ConfigurableApplicationContext context = TestUtils.getPropertyValue(binder, "binder.applicationContext", - ConfigurableApplicationContext.class); - MessagingMessageConverter converter = new MessagingMessageConverter(); - converter.setGenerateMessageId(true); - converter.setGenerateTimestamp(true); - context.getBeanFactory().registerSingleton("testConverter", converter); - QueueChannel moduleInputChannel = new QueueChannel(); - ExtendedProducerProperties producerProperties = createProducerProperties(); - producerProperties.setUseNativeEncoding(true); - producerProperties.getExtension() - .getConfiguration() - .put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName()); - producerBinding = binder.bindProducer(testTopicName, moduleOutputChannel, producerProperties); - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - consumerProperties.getExtension().setAutoRebalanceEnabled(false); - consumerProperties.getExtension() - .getConfiguration() - .put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName()); - consumerProperties.getExtension().setConverterBeanName("testConverter"); - consumerBinding = binder.bindConsumer(testTopicName, "test", moduleInputChannel, consumerProperties); - // Let the consumer actually bind to the producer before sending a msg - binderBindUnbindLatency(); - moduleOutputChannel.send(message); - Message inbound = receive(moduleInputChannel, 500); - assertThat(inbound).isNotNull(); - assertThat(inbound.getPayload()).isEqualTo(new byte[1]); - assertThat(inbound.getHeaders()).containsKey("contentType"); - assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()).isEqualTo("something/funky"); - assertThat(inbound.getHeaders().getId()).isNotNull(); - assertThat(inbound.getHeaders().getTimestamp()).isNotNull(); - } - finally { - if (producerBinding != null) { - producerBinding.unbind(); - } - if (consumerBinding != null) { - consumerBinding.unbind(); - } - } - } - - @Test - @SuppressWarnings("unchecked") - public void testBuiltinSerialization() throws Exception { - Binding producerBinding = null; - Binding consumerBinding = null; - try { - String testPayload = "test"; - Message message = MessageBuilder.withPayload(testPayload) - .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN) - .build(); - - - ExtendedProducerProperties producerProperties = createProducerProperties(); - - DirectChannel moduleOutputChannel = createBindableChannel("output", - createProducerBindingProperties(producerProperties)); - - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - consumerProperties.getExtension().setAutoRebalanceEnabled(false); - - DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); - - String testTopicName = "existing" + System.currentTimeMillis(); - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - final ZkClient zkClient; - zkClient = new ZkClient(configurationProperties.getZkConnectionString(), - configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), - ZKStringSerializer$.MODULE$); - final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); - invokeCreateTopic(zkUtils, testTopicName, 6, 1, new Properties()); - configurationProperties.setAutoAddPartitions(true); - Binder binder = getBinder(configurationProperties); - producerBinding = binder.bindProducer(testTopicName, moduleOutputChannel, producerProperties); - - consumerBinding = binder.bindConsumer(testTopicName, "test", moduleInputChannel, consumerProperties); - // Let the consumer actually bind to the producer before sending a msg - binderBindUnbindLatency(); - moduleOutputChannel.send(message); - CountDownLatch latch = new CountDownLatch(1); - AtomicReference> inboundMessageRef = new AtomicReference<>(); - moduleInputChannel.subscribe(message1 -> { - try { - inboundMessageRef.set((Message) message1); - } - finally { - latch.countDown(); - } - }); - Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); - - assertThat(inboundMessageRef.get()).isNotNull(); - assertThat(inboundMessageRef.get().getPayload()).isEqualTo("test"); - assertThat(inboundMessageRef.get().getHeaders()).containsEntry("contentType", MimeTypeUtils.TEXT_PLAIN); - } - finally { - if (producerBinding != null) { - producerBinding.unbind(); - } - if (consumerBinding != null) { - consumerBinding.unbind(); - } - } - } - @Test @SuppressWarnings("unchecked") public void testPartitionedModuleJavaWithRawMode() throws Exception { @@ -2203,9 +1406,9 @@ public class KafkaBinderTests extends input2.setBeanName("test.input2J"); Binding input2Binding = binder.bindConsumer("partJ.raw.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(); @@ -2263,13 +1466,13 @@ public class KafkaBinderTests extends input2.setBeanName("test.input2S"); Binding input2Binding = binder.bindConsumer("part.raw.0", "test", input2, consumerProperties); - Message message2 = org.springframework.integration.support.MessageBuilder.withPayload(new byte[] { 2 }) + Message message2 = org.springframework.integration.support.MessageBuilder.withPayload(new byte[]{2}) .setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "kafkaBinderTestCommonsDelegate") .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); @@ -2285,7 +1488,7 @@ public class KafkaBinderTests extends } @Test - @SuppressWarnings({ "unchecked", "rawtypes" }) + @SuppressWarnings({"unchecked", "rawtypes"}) public void testSendAndReceiveWithRawMode() throws Exception { Binder binder = getBinder(); @@ -2326,12 +1529,797 @@ public class KafkaBinderTests extends consumerBinding.unbind(); } + @SuppressWarnings({"rawtypes", "unchecked"}) + @Test + public void testProducerErrorChannel() throws Exception { + AbstractKafkaTestBinder binder = getBinder(); + DirectChannel moduleOutputChannel = createBindableChannel("output", new BindingProperties()); + ExtendedProducerProperties producerProps = new ExtendedProducerProperties<>( + new KafkaProducerProperties()); + producerProps.setHeaderMode(HeaderMode.none); + producerProps.setErrorChannelEnabled(true); + Binding producerBinding = binder.bindProducer("ec.0", moduleOutputChannel, producerProps); + final Message message = MessageBuilder.withPayload("bad").setHeader(MessageHeaders.CONTENT_TYPE, "application/json") + .build(); + SubscribableChannel ec = binder.getApplicationContext().getBean("ec.0.errors", SubscribableChannel.class); + final AtomicReference> errorMessage = new AtomicReference<>(); + final CountDownLatch latch = new CountDownLatch(2); + ec.subscribe(message1 -> { + errorMessage.set(message1); + latch.countDown(); + }); + SubscribableChannel globalEc = binder.getApplicationContext() + .getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, SubscribableChannel.class); + globalEc.subscribe(message12 -> latch.countDown()); + KafkaProducerMessageHandler endpoint = TestUtils.getPropertyValue(producerBinding, "lifecycle", + KafkaProducerMessageHandler.class); + final RuntimeException fooException = new RuntimeException("foo"); + final AtomicReference sent = new AtomicReference<>(); + new DirectFieldAccessor(endpoint).setPropertyValue("kafkaTemplate", + new KafkaTemplate(mock(ProducerFactory.class)) { + + @Override // SIK < 2.3 + public ListenableFuture send(String topic, Object payload) { + sent.set(payload); + SettableListenableFuture future = new SettableListenableFuture<>(); + future.setException(fooException); + return future; + } + + @Override // SIK 2.3+ + public ListenableFuture send(ProducerRecord record) { + sent.set(record.value()); + SettableListenableFuture future = new SettableListenableFuture<>(); + future.setException(fooException); + return future; + } + + }); + + moduleOutputChannel.send(message); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(errorMessage.get()).isInstanceOf(ErrorMessage.class); + assertThat(errorMessage.get().getPayload()).isInstanceOf(KafkaSendFailureException.class); + KafkaSendFailureException exception = (KafkaSendFailureException) errorMessage.get().getPayload(); + assertThat(exception.getCause()).isSameAs(fooException); + assertThat(new String((byte[]) exception.getFailedMessage().getPayload(), StandardCharsets.UTF_8)).isEqualTo(message.getPayload()); + assertThat(exception.getRecord().value()).isSameAs(sent.get()); + producerBinding.unbind(); + } + + @Test + @SuppressWarnings("unchecked") + public void testAutoCreateTopicsEnabledSucceeds() throws Exception { + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + configurationProperties.setAutoCreateTopics(true); + Binder binder = getBinder(configurationProperties); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + String testTopicName = "nonexisting" + System.currentTimeMillis(); + DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); + + Binding binding = binder.bindConsumer(testTopicName, "test", moduleInputChannel, consumerProperties); + binding.unbind(); + } + + @Test + @SuppressWarnings("unchecked") + public void testCustomPartitionCountOverridesDefaultIfLarger() throws Exception { + byte[] testPayload = new byte[2048]; + Arrays.fill(testPayload, (byte) 65); + KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); + binderConfiguration.setMinPartitionCount(10); + Binder binder = getBinder(binderConfiguration); + QueueChannel moduleInputChannel = new QueueChannel(); + ExtendedProducerProperties producerProperties = createProducerProperties(); + producerProperties.setPartitionCount(10); + + DirectChannel moduleOutputChannel = createBindableChannel("output", + createProducerBindingProperties(producerProperties)); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + long uniqueBindingId = System.currentTimeMillis(); + Binding producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0", + moduleOutputChannel, producerProperties); + Binding consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null, + moduleInputChannel, consumerProperties); + Message message = org.springframework.integration.support.MessageBuilder.withPayload(testPayload) + .build(); + // Let the consumer actually bind to the producer before sending a msg + binderBindUnbindLatency(); + moduleOutputChannel.send(message); + Message inbound = receive(moduleInputChannel); + assertThat(inbound).isNotNull(); + assertThat((byte[]) inbound.getPayload()).containsExactly(testPayload); + + assertThat(partitionSize("foo" + uniqueBindingId + ".0")).isEqualTo(10); + producerBinding.unbind(); + consumerBinding.unbind(); + } + + @Test + @SuppressWarnings("unchecked") + public void testCustomPartitionCountDoesNotOverridePartitioningIfSmaller() throws Exception { + byte[] testPayload = new byte[2048]; + Arrays.fill(testPayload, (byte) 65); + KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); + binderConfiguration.setMinPartitionCount(6); + Binder binder = getBinder(binderConfiguration); + 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 + binderBindUnbindLatency(); + moduleOutputChannel.send(message); + Message inbound = receive(moduleInputChannel); + assertThat(inbound).isNotNull(); + assertThat((byte[]) inbound.getPayload()).containsExactly(testPayload); + + assertThat(partitionSize("foo" + uniqueBindingId + ".0")).isEqualTo(6); + producerBinding.unbind(); + consumerBinding.unbind(); + } + + @Test + @SuppressWarnings("unchecked") + public void testDynamicKeyExpression() throws Exception { + Binder binder = getBinder(createConfigurationProperties()); + QueueChannel moduleInputChannel = new QueueChannel(); + ExtendedProducerProperties producerProperties = createProducerProperties(); + producerProperties.getExtension().getConfiguration().put("key.serializer", StringSerializer.class.getName()); + producerProperties.getExtension().setMessageKeyExpression(spelExpressionParser.parseExpression("headers.key")); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + String uniqueBindingId = UUID.randomUUID().toString(); + 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 = MessageBuilder.withPayload("somePayload").setHeader("key", "myDynamicKey").build(); + // Let the consumer actually bind to the producer before sending a msg + binderBindUnbindLatency(); + moduleOutputChannel.send(message); + Message inbound = receive(moduleInputChannel); + assertThat(inbound).isNotNull(); + String receivedKey = new String(inbound.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, byte[].class)); + assertThat(receivedKey).isEqualTo("myDynamicKey"); + producerBinding.unbind(); + consumerBinding.unbind(); + } + + @Test + @SuppressWarnings("unchecked") + public void testCustomPartitionCountOverridesPartitioningIfLarger() throws Exception { + byte[] testPayload = new byte[2048]; + Arrays.fill(testPayload, (byte) 65); + KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); + binderConfiguration.setMinPartitionCount(4); + Binder binder = getBinder(binderConfiguration); + + 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", + moduleOutputChannel, producerProperties); + Binding consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null, + moduleInputChannel, consumerProperties); + Message message = org.springframework.integration.support.MessageBuilder.withPayload(testPayload) + .build(); + // Let the consumer actually bind to the producer before sending a msg + binderBindUnbindLatency(); + moduleOutputChannel.send(message); + Message inbound = receive(moduleInputChannel); + assertThat(inbound).isNotNull(); + assertThat((byte[]) inbound.getPayload()).containsExactly(testPayload); + assertThat(partitionSize("foo" + uniqueBindingId + ".0")).isEqualTo(5); + producerBinding.unbind(); + consumerBinding.unbind(); + } + + @Test + @SuppressWarnings("unchecked") + public void testDefaultConsumerStartsAtEarliest() throws Exception { + Binder binder = getBinder(createConfigurationProperties()); + GenericApplicationContext context = new GenericApplicationContext(); + context.refresh(); + + BindingProperties producerBindingProperties = createProducerBindingProperties(createProducerProperties()); + DirectChannel output = createBindableChannel("output", producerBindingProperties); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); + + DirectChannel input1 = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); + + String testTopicName = UUID.randomUUID().toString(); + Binding producerBinding = binder.bindProducer(testTopicName, output, + createProducerProperties()); + String testPayload1 = "foo-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload1.getBytes())); + + Binding consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, + consumerProperties); + + CountDownLatch latch = new CountDownLatch(1); + AtomicReference> inboundMessageRef1 = new AtomicReference<>(); + MessageHandler messageHandler = message1 -> { + try { + inboundMessageRef1.set((Message) message1); + } + finally { + latch.countDown(); + } + }; + input1.subscribe(messageHandler); + Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); + assertThat(inboundMessageRef1.get()).isNotNull(); + assertThat(inboundMessageRef1.get().getPayload()).isEqualTo(testPayload1); + + String testPayload2 = "foo-" + UUID.randomUUID().toString(); + input1.unsubscribe(messageHandler); + output.send(new GenericMessage<>(testPayload2.getBytes())); + + CountDownLatch latch1 = new CountDownLatch(1); + AtomicReference> inboundMessageRef2 = new AtomicReference<>(); + input1.subscribe(message1 -> { + try { + inboundMessageRef2.set((Message) message1); + } + finally { + latch1.countDown(); + } + }); + Assert.isTrue(latch1.await(5, TimeUnit.SECONDS), "Failed to receive message"); + + assertThat(inboundMessageRef2.get()).isNotNull(); + assertThat(inboundMessageRef2.get().getPayload()).isEqualTo(testPayload2); + + producerBinding.unbind(); + consumerBinding.unbind(); + } + + @Test + @SuppressWarnings("unchecked") + public void testResume() throws Exception { + Binding producerBinding = null; + Binding consumerBinding = null; + + try { + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + Binder binder = getBinder(configurationProperties); + + BindingProperties producerBindingProperties = createProducerBindingProperties(createProducerProperties()); + DirectChannel output = createBindableChannel("output", producerBindingProperties); + + DirectChannel input1 = createBindableChannel("input", createConsumerBindingProperties(createConsumerProperties())); + + String testTopicName = UUID.randomUUID().toString(); + producerBinding = binder.bindProducer(testTopicName, output, + producerBindingProperties.getProducer()); + String testPayload1 = "foo1-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload1)); + ExtendedConsumerProperties firstConsumerProperties = createConsumerProperties(); + consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, + firstConsumerProperties); + CountDownLatch latch = new CountDownLatch(1); + AtomicReference> inboundMessageRef1 = new AtomicReference<>(); + MessageHandler messageHandler = message1 -> { + try { + inboundMessageRef1.set((Message) message1); + } + finally { + latch.countDown(); + } + }; + input1.subscribe(messageHandler); + Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); + + assertThat(inboundMessageRef1.get()).isNotNull(); + assertThat(inboundMessageRef1.get().getPayload()).isNotNull(); + String testPayload2 = "foo2-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload2.getBytes())); + input1.unsubscribe(messageHandler); + CountDownLatch latch1 = new CountDownLatch(1); + AtomicReference> inboundMessageRef2 = new AtomicReference<>(); + MessageHandler messageHandler1 = message1 -> { + try { + inboundMessageRef2.set((Message) message1); + } + finally { + latch1.countDown(); + } + }; + input1.subscribe(messageHandler1); + Assert.isTrue(latch1.await(5, TimeUnit.SECONDS), "Failed to receive message"); + assertThat(inboundMessageRef2.get()).isNotNull(); + assertThat(inboundMessageRef2.get().getPayload()).isNotNull(); + consumerBinding.unbind(); + + Thread.sleep(2000); + String testPayload3 = "foo3-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload3.getBytes())); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, consumerProperties); + input1.unsubscribe(messageHandler1); + CountDownLatch latch2 = new CountDownLatch(1); + AtomicReference> inboundMessageRef3 = new AtomicReference<>(); + MessageHandler messageHandler2 = message1 -> { + try { + inboundMessageRef3.set((Message) message1); + } + finally { + latch2.countDown(); + } + }; + input1.subscribe(messageHandler2); + Assert.isTrue(latch2.await(5, TimeUnit.SECONDS), "Failed to receive message"); + assertThat(inboundMessageRef3.get()).isNotNull(); + assertThat(new String(inboundMessageRef3.get().getPayload())).isEqualTo(testPayload3); + } + finally { + if (consumerBinding != null) { + consumerBinding.unbind(); + } + if (producerBinding != null) { + producerBinding.unbind(); + } + } + } + + @Test + @SuppressWarnings("unchecked") + public void testSyncProducerMetadata() throws Exception { + Binder binder = getBinder(createConfigurationProperties()); + DirectChannel output = new DirectChannel(); + String testTopicName = UUID.randomUUID().toString(); + ExtendedProducerProperties properties = createProducerProperties(); + properties.getExtension().setSync(true); + Binding producerBinding = binder.bindProducer(testTopicName, output, properties); + DirectFieldAccessor accessor = new DirectFieldAccessor(extractEndpoint(producerBinding)); + KafkaProducerMessageHandler wrappedInstance = (KafkaProducerMessageHandler) accessor.getWrappedInstance(); + assertThat(new DirectFieldAccessor(wrappedInstance).getPropertyValue("sync").equals(Boolean.TRUE)) + .withFailMessage("Kafka Sync Producer should have been enabled."); + producerBinding.unbind(); + } + + @Test + @SuppressWarnings("unchecked") + public void testAutoCreateTopicsDisabledOnBinderStillWorksAsLongAsBrokerCreatesTopic() throws Exception { + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + configurationProperties.setAutoCreateTopics(false); + Binder binder = getBinder(configurationProperties); + BindingProperties producerBindingProperties = createProducerBindingProperties(createProducerProperties()); + DirectChannel output = createBindableChannel("output", producerBindingProperties); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + + DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); + + String testTopicName = "createdByBroker-" + System.currentTimeMillis(); + + Binding producerBinding = binder.bindProducer(testTopicName, output, + producerBindingProperties.getProducer()); + + String testPayload = "foo1-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload)); + + Binding consumerBinding = binder.bindConsumer(testTopicName, "test", input, consumerProperties); + CountDownLatch latch = new CountDownLatch(1); + AtomicReference> inboundMessageRef = new AtomicReference<>(); + input.subscribe(message1 -> { + try { + inboundMessageRef.set((Message) message1); + } + finally { + latch.countDown(); + } + }); + Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); + + assertThat(inboundMessageRef.get()).isNotNull(); + assertThat(inboundMessageRef.get().getPayload()).isEqualTo(testPayload); + + producerBinding.unbind(); + consumerBinding.unbind(); + } + + @Test + @SuppressWarnings("unchecked") + public void testAutoConfigureTopicsDisabledSucceedsIfTopicExisting() throws Throwable { + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + + String testTopicName = "existing" + System.currentTimeMillis(); + invokeCreateTopic(testTopicName, 5, 1); + configurationProperties.setAutoCreateTopics(false); + Binder binder = getBinder(configurationProperties); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + + DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); + Binding binding = binder.bindConsumer(testTopicName, "test", input, consumerProperties); + binding.unbind(); + } + + @Test + @SuppressWarnings("unchecked") + public void testPartitionCountIncreasedIfAutoAddPartitionsSet() throws Throwable { + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + + String testTopicName = "existing" + System.currentTimeMillis(); + configurationProperties.setMinPartitionCount(6); + configurationProperties.setAutoAddPartitions(true); + Binder binder = getBinder(configurationProperties); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); + + Binding binding = binder.bindConsumer(testTopicName, "test", input, consumerProperties); + binding.unbind(); + assertThat(invokePartitionSize(testTopicName)).isEqualTo(6); + } + + @Test + @SuppressWarnings("unchecked") + public void testAutoAddPartitionsDisabledSucceedsIfTopicUnderPartitionedAndAutoRebalanceEnabled() throws Throwable { + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + + String testTopicName = "existing" + System.currentTimeMillis(); + invokeCreateTopic(testTopicName, 1, 1); + configurationProperties.setAutoAddPartitions(false); + Binder binder = getBinder(configurationProperties); + GenericApplicationContext context = new GenericApplicationContext(); + context.refresh(); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + + DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); + + // this consumer must consume from partition 2 + consumerProperties.setInstanceCount(3); + consumerProperties.setInstanceIndex(2); + Binding binding = binder.bindConsumer(testTopicName, "test", input, consumerProperties); + binding.unbind(); + assertThat(invokePartitionSize(testTopicName)).isEqualTo(1); + } + + @Test + @SuppressWarnings("unchecked") + public void testAutoAddPartitionsDisabledFailsIfTopicUnderPartitionedAndAutoRebalanceDisabled() throws Throwable { + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + + String testTopicName = "existing" + System.currentTimeMillis(); + invokeCreateTopic(testTopicName, 1, 1); + configurationProperties.setAutoAddPartitions(false); + Binder binder = getBinder(configurationProperties); + GenericApplicationContext context = new GenericApplicationContext(); + context.refresh(); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + DirectChannel output = createBindableChannel("output", createConsumerBindingProperties(consumerProperties)); + // this consumer must consume from partition 2 + consumerProperties.setInstanceCount(3); + consumerProperties.setInstanceIndex(2); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); + expectedProvisioningException.expect(ProvisioningException.class); + expectedProvisioningException + .expectMessage("The number of expected partitions was: 3, but 1 has been found instead"); + Binding binding = binder.bindConsumer(testTopicName, "test", output, consumerProperties); + if (binding != null) { + binding.unbind(); + } + } + + @Test + @SuppressWarnings("unchecked") + public void testAutoAddPartitionsDisabledSucceedsIfTopicPartitionedCorrectly() throws Throwable { + Binding binding = null; + try { + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + + String testTopicName = "existing" + System.currentTimeMillis(); + invokeCreateTopic(testTopicName, 6, 1); + configurationProperties.setAutoAddPartitions(false); + Binder binder = getBinder(configurationProperties); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + + DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); + + // this consumer must consume from partition 2 + consumerProperties.setInstanceCount(3); + consumerProperties.setInstanceIndex(2); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); + + binding = binder.bindConsumer(testTopicName, "test-x", input, consumerProperties); + + TopicPartitionInitialOffset[] listenedPartitions = TestUtils.getPropertyValue(binding, + "lifecycle.messageListenerContainer.containerProperties.topicPartitions", + TopicPartitionInitialOffset[].class); + assertThat(listenedPartitions).hasSize(2); + assertThat(listenedPartitions).contains(new TopicPartitionInitialOffset(testTopicName, 2), + new TopicPartitionInitialOffset(testTopicName, 5)); + int partitions = invokePartitionSize(testTopicName); + assertThat(partitions).isEqualTo(6); + } + finally { + if (binding != null) { + binding.unbind(); + } + } + } + + @Test + @SuppressWarnings("unchecked") + public void testPartitionCountNotReduced() throws Throwable { + String testTopicName = "existing" + System.currentTimeMillis(); + + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + + invokeCreateTopic(testTopicName, 6, 1); + configurationProperties.setAutoAddPartitions(true); + Binder binder = getBinder(configurationProperties); + GenericApplicationContext context = new GenericApplicationContext(); + context.refresh(); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); + + Binding binding = binder.bindConsumer(testTopicName, "test", input, consumerProperties); + binding.unbind(); + + assertThat(partitionSize(testTopicName)).isEqualTo(6); + } + + @Test + @SuppressWarnings("unchecked") + public void testConsumerDefaultDeserializer() throws Throwable { + Binding binding = null; + try { + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + String testTopicName = "existing" + System.currentTimeMillis(); + invokeCreateTopic(testTopicName, 5, 1); + configurationProperties.setAutoCreateTopics(false); + Binder binder = getBinder(configurationProperties); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); + + binding = binder.bindConsumer(testTopicName, "test", input, consumerProperties); + DirectFieldAccessor consumerAccessor = new DirectFieldAccessor(getKafkaConsumer(binding)); + assertTrue(consumerAccessor.getPropertyValue("keyDeserializer") instanceof ByteArrayDeserializer); + assertTrue(consumerAccessor.getPropertyValue("valueDeserializer") instanceof ByteArrayDeserializer); + } + finally { + if (binding != null) { + binding.unbind(); + } + } + } + + @Test + @SuppressWarnings("unchecked") + public void testConsumerCustomDeserializer() throws Exception { + Binding binding = null; + try { + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + Map propertiesToOverride = configurationProperties.getConfiguration(); + propertiesToOverride.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + propertiesToOverride.put("value.deserializer", "org.apache.kafka.common.serialization.LongDeserializer"); + configurationProperties.setConfiguration(propertiesToOverride); + String testTopicName = "existing" + System.currentTimeMillis(); + configurationProperties.setAutoCreateTopics(false); + Binder binder = getBinder(configurationProperties); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); + + binding = binder.bindConsumer(testTopicName, "test", input, consumerProperties); + DirectFieldAccessor consumerAccessor = new DirectFieldAccessor(getKafkaConsumer(binding)); + assertTrue("Expected StringDeserializer as a custom key deserializer", + consumerAccessor.getPropertyValue("keyDeserializer") instanceof StringDeserializer); + assertTrue("Expected LongDeserializer as a custom value deserializer", + consumerAccessor.getPropertyValue("valueDeserializer") instanceof LongDeserializer); + } + finally { + if (binding != null) { + binding.unbind(); + } + } + } + + @Test + @SuppressWarnings({"unchecked", "rawtypes"}) + public void testNativeSerializationWithCustomSerializerDeserializer() throws Exception { + Binding producerBinding = null; + Binding consumerBinding = null; + try { + Integer testPayload = 10; + Message message = MessageBuilder.withPayload(testPayload).build(); + SubscribableChannel moduleOutputChannel = new DirectChannel(); + String testTopicName = "existing" + System.currentTimeMillis(); + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + configurationProperties.setAutoAddPartitions(true); + Binder binder = getBinder(configurationProperties); + QueueChannel moduleInputChannel = new QueueChannel(); + ExtendedProducerProperties producerProperties = createProducerProperties(); + producerProperties.setUseNativeEncoding(true); + producerProperties.getExtension().getConfiguration().put("value.serializer", + "org.apache.kafka.common.serialization.IntegerSerializer"); + producerBinding = binder.bindProducer(testTopicName, moduleOutputChannel, producerProperties); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); + consumerProperties.getExtension().getConfiguration().put("value.deserializer", + "org.apache.kafka.common.serialization.IntegerDeserializer"); + consumerProperties.getExtension().setStandardHeaders(KafkaConsumerProperties.StandardHeaders.both); + consumerBinding = binder.bindConsumer(testTopicName, "test", moduleInputChannel, consumerProperties); + // Let the consumer actually bind to the producer before sending a msg + binderBindUnbindLatency(); + moduleOutputChannel.send(message); + Message inbound = receive(moduleInputChannel, 500); + assertThat(inbound).isNotNull(); + assertThat(inbound.getPayload()).isEqualTo(10); + assertThat(inbound.getHeaders()).doesNotContainKey("contentType"); + assertThat(inbound.getHeaders().getId()).isNotNull(); + assertThat(inbound.getHeaders().getTimestamp()).isNotNull(); + } + finally { + if (producerBinding != null) { + producerBinding.unbind(); + } + if (consumerBinding != null) { + consumerBinding.unbind(); + } + } + } + + private KafkaConsumer getKafkaConsumer(Binding binding) { + DirectFieldAccessor bindingAccessor = new DirectFieldAccessor(binding); + KafkaMessageDrivenChannelAdapter adapter = (KafkaMessageDrivenChannelAdapter) bindingAccessor + .getPropertyValue("lifecycle"); + DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter); + ConcurrentMessageListenerContainer messageListenerContainer = + (ConcurrentMessageListenerContainer) adapterAccessor.getPropertyValue("messageListenerContainer"); + DirectFieldAccessor containerAccessor = new DirectFieldAccessor(messageListenerContainer); + DefaultKafkaConsumerFactory consumerFactory = (DefaultKafkaConsumerFactory) containerAccessor + .getPropertyValue("consumerFactory"); + return (KafkaConsumer) consumerFactory.createConsumer(); + } + + @Test + @SuppressWarnings({"unchecked", "rawtypes"}) + public void testNativeSerializationWithCustomSerializerDeserializerBytesPayload() throws Exception { + Binding producerBinding = null; + Binding consumerBinding = null; + try { + byte[] testPayload = new byte[1]; + Message message = MessageBuilder.withPayload(testPayload) + .setHeader(MessageHeaders.CONTENT_TYPE, "something/funky") + .build(); + SubscribableChannel moduleOutputChannel = new DirectChannel(); + String testTopicName = "existing" + System.currentTimeMillis(); + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + configurationProperties.setAutoAddPartitions(true); + Binder binder = getBinder(configurationProperties); + ConfigurableApplicationContext context = TestUtils.getPropertyValue(binder, "binder.applicationContext", + ConfigurableApplicationContext.class); + MessagingMessageConverter converter = new MessagingMessageConverter(); + converter.setGenerateMessageId(true); + converter.setGenerateTimestamp(true); + context.getBeanFactory().registerSingleton("testConverter", converter); + QueueChannel moduleInputChannel = new QueueChannel(); + ExtendedProducerProperties producerProperties = createProducerProperties(); + producerProperties.setUseNativeEncoding(true); + producerProperties.getExtension() + .getConfiguration() + .put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName()); + producerBinding = binder.bindProducer(testTopicName, moduleOutputChannel, producerProperties); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); + consumerProperties.getExtension() + .getConfiguration() + .put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName()); + consumerProperties.getExtension().setConverterBeanName("testConverter"); + consumerBinding = binder.bindConsumer(testTopicName, "test", moduleInputChannel, consumerProperties); + // Let the consumer actually bind to the producer before sending a msg + binderBindUnbindLatency(); + moduleOutputChannel.send(message); + Message inbound = receive(moduleInputChannel, 500); + assertThat(inbound).isNotNull(); + assertThat(inbound.getPayload()).isEqualTo(new byte[1]); + assertThat(inbound.getHeaders()).containsKey("contentType"); + assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()).isEqualTo("something/funky"); + assertThat(inbound.getHeaders().getId()).isNotNull(); + assertThat(inbound.getHeaders().getTimestamp()).isNotNull(); + } + finally { + if (producerBinding != null) { + producerBinding.unbind(); + } + if (consumerBinding != null) { + consumerBinding.unbind(); + } + } + } + + @Test + @SuppressWarnings("unchecked") + public void testBuiltinSerialization() throws Exception { + Binding producerBinding = null; + Binding consumerBinding = null; + try { + String testPayload = "test"; + Message message = MessageBuilder.withPayload(testPayload) + .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN) + .build(); + + ExtendedProducerProperties producerProperties = createProducerProperties(); + + DirectChannel moduleOutputChannel = createBindableChannel("output", + createProducerBindingProperties(producerProperties)); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); + + DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); + + String testTopicName = "existing" + System.currentTimeMillis(); + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + configurationProperties.setAutoAddPartitions(true); + Binder binder = getBinder(configurationProperties); + producerBinding = binder.bindProducer(testTopicName, moduleOutputChannel, producerProperties); + + consumerBinding = binder.bindConsumer(testTopicName, "test", moduleInputChannel, consumerProperties); + // Let the consumer actually bind to the producer before sending a msg + binderBindUnbindLatency(); + moduleOutputChannel.send(message); + CountDownLatch latch = new CountDownLatch(1); + AtomicReference> inboundMessageRef = new AtomicReference<>(); + moduleInputChannel.subscribe(message1 -> { + try { + inboundMessageRef.set((Message) message1); + } + finally { + latch.countDown(); + } + }); + Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); + + assertThat(inboundMessageRef.get()).isNotNull(); + assertThat(inboundMessageRef.get().getPayload()).isEqualTo("test"); + assertThat(inboundMessageRef.get().getHeaders()).containsEntry("contentType", MimeTypeUtils.TEXT_PLAIN); + } + finally { + if (producerBinding != null) { + producerBinding.unbind(); + } + if (consumerBinding != null) { + consumerBinding.unbind(); + } + } + } + /* * Verify that a consumer configured to handle embedded headers can handle * all three variants. */ @Test - @SuppressWarnings({ "unchecked", "rawtypes" }) + @SuppressWarnings({"unchecked", "rawtypes"}) public void testSendAndReceiveWithMixedMode() throws Exception { KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); binderConfiguration.setHeaders("foo"); @@ -2415,64 +2403,6 @@ public class KafkaBinderTests extends consumerBinding.unbind(); } - @SuppressWarnings({ "rawtypes", "unchecked" }) - @Test - public void testProducerErrorChannel() throws Exception { - AbstractKafkaTestBinder binder = getBinder(); - DirectChannel moduleOutputChannel = createBindableChannel("output", new BindingProperties()); - ExtendedProducerProperties producerProps = new ExtendedProducerProperties<>( - new KafkaProducerProperties()); - producerProps.setHeaderMode(HeaderMode.none); - producerProps.setErrorChannelEnabled(true); - Binding producerBinding = binder.bindProducer("ec.0", moduleOutputChannel, producerProps); - final Message message = MessageBuilder.withPayload("bad").setHeader(MessageHeaders.CONTENT_TYPE, "application/json") - .build(); - SubscribableChannel ec = binder.getApplicationContext().getBean("ec.0.errors", SubscribableChannel.class); - final AtomicReference> errorMessage = new AtomicReference<>(); - final CountDownLatch latch = new CountDownLatch(2); - ec.subscribe(message1 -> { - errorMessage.set(message1); - latch.countDown(); - }); - SubscribableChannel globalEc = binder.getApplicationContext() - .getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, SubscribableChannel.class); - globalEc.subscribe(message12 -> latch.countDown()); - KafkaProducerMessageHandler endpoint = TestUtils.getPropertyValue(producerBinding, "lifecycle", - KafkaProducerMessageHandler.class); - final RuntimeException fooException = new RuntimeException("foo"); - final AtomicReference sent = new AtomicReference<>(); - new DirectFieldAccessor(endpoint).setPropertyValue("kafkaTemplate", - new KafkaTemplate(mock(ProducerFactory.class)) { - - @Override // SIK < 2.3 - public ListenableFuture send(String topic, Object payload) { - sent.set(payload); - SettableListenableFuture future = new SettableListenableFuture<>(); - future.setException(fooException); - return future; - } - - @Override // SIK 2.3+ - public ListenableFuture send(ProducerRecord record) { - sent.set(record.value()); - SettableListenableFuture future = new SettableListenableFuture<>(); - future.setException(fooException); - return future; - } - - }); - - moduleOutputChannel.send(message); - assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); - assertThat(errorMessage.get()).isInstanceOf(ErrorMessage.class); - assertThat(errorMessage.get().getPayload()).isInstanceOf(KafkaSendFailureException.class); - KafkaSendFailureException exception = (KafkaSendFailureException) errorMessage.get().getPayload(); - assertThat(exception.getCause()).isSameAs(fooException); - assertThat(new String((byte[])exception.getFailedMessage().getPayload(), StandardCharsets.UTF_8)).isEqualTo(message.getPayload()); - assertThat(exception.getRecord().value()).isSameAs(sent.get()); - producerBinding.unbind(); - } - private 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/KafkaBinderUnitTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderUnitTests.java index 36a3b8f8b..7fac11227 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderUnitTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderUnitTests.java @@ -23,15 +23,14 @@ import java.util.Map; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.junit.Test; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; -import org.springframework.cloud.stream.binder.kafka.admin.AdminUtilsOperation; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; import org.springframework.integration.test.util.TestUtils; import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.Mockito.mock; /** * @author Gary Russell @@ -43,9 +42,7 @@ public class KafkaBinderUnitTests { @Test public void testPropertyOverrides() throws Exception { KafkaBinderConfigurationProperties binderConfigurationProperties = new KafkaBinderConfigurationProperties(); - AdminUtilsOperation adminUtilsOperation = mock(AdminUtilsOperation.class); - KafkaTopicProvisioner provisioningProvider = new KafkaTopicProvisioner(binderConfigurationProperties, - adminUtilsOperation); + KafkaTopicProvisioner provisioningProvider = new KafkaTopicProvisioner(binderConfigurationProperties, new KafkaProperties()); KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(binderConfigurationProperties, provisioningProvider); KafkaConsumerProperties consumerProps = new KafkaConsumerProperties(); 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 708281246..f89abdbb0 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 @@ -17,8 +17,6 @@ package org.springframework.cloud.stream.binder.kafka; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; -import org.springframework.cloud.stream.binder.kafka.admin.AdminUtilsOperation; -import org.springframework.cloud.stream.binder.kafka.admin.KafkaAdminUtilsOperation; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; @@ -31,6 +29,7 @@ import org.springframework.kafka.support.ProducerListener; /** * Test support class for {@link KafkaMessageChannelBinder}. + * * @author Eric Bottard * @author Marius Bogoevici * @author David Turanski @@ -39,16 +38,11 @@ import org.springframework.kafka.support.ProducerListener; */ public class KafkaTestBinder extends AbstractKafkaTestBinder { - @SuppressWarnings({ "rawtypes", "unchecked" }) - KafkaTestBinder(KafkaBinderConfigurationProperties binderConfiguration) { + @SuppressWarnings({"rawtypes", "unchecked"}) + KafkaTestBinder(KafkaBinderConfigurationProperties binderConfiguration, KafkaTopicProvisioner kafkaTopicProvisioner) { try { - AdminUtilsOperation adminUtilsOperation = new KafkaAdminUtilsOperation(); - KafkaTopicProvisioner provisioningProvider = - new KafkaTopicProvisioner(binderConfiguration, adminUtilsOperation); - provisioningProvider.afterPropertiesSet(); - KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(binderConfiguration, - provisioningProvider) { + kafkaTopicProvisioner) { /* * Some tests use multiple instance indexes for the same topic; we need to make @@ -56,7 +50,7 @@ public class KafkaTestBinder extends AbstractKafkaTestBinder { */ @Override protected String errorsBaseName(ConsumerDestination destination, String group, - ExtendedConsumerProperties consumerProperties) { + ExtendedConsumerProperties consumerProperties) { return super.errorsBaseName(destination, group, consumerProperties) + "-" + consumerProperties.getInstanceIndex(); } diff --git a/spring-cloud-stream-binder-kstream/pom.xml b/spring-cloud-stream-binder-kstream/pom.xml index 1c0ce9afb..cde1340c5 100644 --- a/spring-cloud-stream-binder-kstream/pom.xml +++ b/spring-cloud-stream-binder-kstream/pom.xml @@ -28,10 +28,6 @@ spring-boot-autoconfigure true - - org.apache.kafka - kafka_2.11 - org.apache.kafka kafka-streams @@ -46,36 +42,21 @@ test - org.springframework.kafka + org.springframework.kafka spring-kafka-test org.apache.kafka kafka_2.11 test - test - - - jline - jline - - - org.slf4j - slf4j-log4j12 - - - log4j - log4j - - - log4j - log4j - 1.2.17 - test - + log4j + log4j + 1.2.17 + test + org.springframework.cloud spring-cloud-stream-binder-test diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java index 701326a4d..5296a8f92 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java @@ -19,8 +19,8 @@ package org.springframework.cloud.stream.binder.kstream; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.kstream.KStream; -import org.apache.kafka.streams.kstream.KStreamBuilder; import org.apache.kafka.streams.kstream.KeyValueMapper; import org.springframework.aop.framework.ProxyFactory; @@ -42,13 +42,13 @@ import org.springframework.util.StringUtils; */ public class KStreamBoundElementFactory extends AbstractBindingTargetFactory { - private final KStreamBuilder kStreamBuilder; + private final StreamsBuilder kStreamBuilder; private final BindingServiceProperties bindingServiceProperties; private CompositeMessageConverterFactory compositeMessageConverterFactory; - public KStreamBoundElementFactory(KStreamBuilder streamBuilder, BindingServiceProperties bindingServiceProperties, + public KStreamBoundElementFactory(StreamsBuilder streamBuilder, BindingServiceProperties bindingServiceProperties, CompositeMessageConverterFactory compositeMessageConverterFactory) { super(KStream.class); this.bindingServiceProperties = bindingServiceProperties; diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java index b12881dff..0c187bc2b 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java @@ -21,10 +21,8 @@ import org.apache.commons.logging.LogFactory; import org.apache.kafka.streams.StreamsConfig; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.cloud.stream.binder.kafka.admin.AdminUtilsOperation; -import org.springframework.cloud.stream.binder.kafka.admin.KafkaAdminUtilsOperation; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; import org.springframework.cloud.stream.binder.kstream.KStreamBinder; @@ -39,14 +37,14 @@ import org.springframework.context.annotation.Configuration; @EnableConfigurationProperties(KStreamExtendedBindingProperties.class) public class KStreamBinderConfiguration { - @Autowired(required = false) - private AdminUtilsOperation adminUtilsOperation; - private static final Log logger = LogFactory.getLog(KStreamBinderConfiguration.class); + @Autowired + private KafkaProperties kafkaProperties; + @Bean public KafkaTopicProvisioner provisioningProvider(KafkaBinderConfigurationProperties binderConfigurationProperties) { - return new KafkaTopicProvisioner(binderConfigurationProperties, adminUtilsOperation); + return new KafkaTopicProvisioner(binderConfigurationProperties, kafkaProperties); } @Bean @@ -57,11 +55,4 @@ public class KStreamBinderConfiguration { streamsConfig); } - @Bean(name = "adminUtilsOperation") - @ConditionalOnClass(name = "kafka.admin.AdminUtils") - public AdminUtilsOperation kafka10AdminUtilsOperation() { - logger.info("AdminUtils selected: Kafka 0.10 AdminUtils"); - return new KafkaAdminUtilsOperation(); - } - } diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderSupportAutoConfiguration.java index c3d1080a7..4eae8c6b3 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderSupportAutoConfiguration.java @@ -19,8 +19,8 @@ package org.springframework.cloud.stream.binder.kstream.config; import java.util.Properties; import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.StreamsConfig; -import org.apache.kafka.streams.kstream.KStreamBuilder; import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.UnsatisfiedDependencyException; @@ -34,7 +34,7 @@ import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.context.annotation.Bean; import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration; -import org.springframework.kafka.core.KStreamBuilderFactoryBean; +import org.springframework.kafka.core.StreamsBuilderFactoryBean; import org.springframework.util.ObjectUtils; /** @@ -48,18 +48,18 @@ public class KStreamBinderSupportAutoConfiguration { return new KafkaBinderConfigurationProperties(); } - @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_KSTREAM_BUILDER_BEAN_NAME) - public KStreamBuilderFactoryBean defaultKStreamBuilder( + @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_BUILDER_BEAN_NAME) + public StreamsBuilderFactoryBean defaultKafkaStreamBuilder( @Qualifier(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME) ObjectProvider streamsConfigProvider) { StreamsConfig streamsConfig = streamsConfigProvider.getIfAvailable(); if (streamsConfig != null) { - KStreamBuilderFactoryBean kStreamBuilderFactoryBean = new KStreamBuilderFactoryBean(streamsConfig); + StreamsBuilderFactoryBean kStreamBuilderFactoryBean = new StreamsBuilderFactoryBean(streamsConfig); kStreamBuilderFactoryBean.setPhase(Integer.MAX_VALUE - 500); return kStreamBuilderFactoryBean; } else { throw new UnsatisfiedDependencyException(KafkaStreamsDefaultConfiguration.class.getName(), - KafkaStreamsDefaultConfiguration.DEFAULT_KSTREAM_BUILDER_BEAN_NAME, "streamsConfig", + KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_BUILDER_BEAN_NAME, "streamsConfig", "There is no '" + KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME + "' StreamsConfig bean in the application context.\n"); } @@ -79,19 +79,19 @@ public class KStreamBinderSupportAutoConfiguration { } @Bean - public KStreamStreamListenerResultAdapter kStreamStreamListenerResultAdapter() { + public KStreamStreamListenerResultAdapter kafkaStreamStreamListenerResultAdapter() { return new KStreamStreamListenerResultAdapter(); } @Bean - public KStreamListenerParameterAdapter kStreamListenerParameterAdapter( + public KStreamListenerParameterAdapter kafkaStreamListenerParameterAdapter( CompositeMessageConverterFactory compositeMessageConverterFactory) { return new KStreamListenerParameterAdapter( compositeMessageConverterFactory.getMessageConverterForAllRegistered()); } @Bean - public KStreamBoundElementFactory kStreamBindableTargetFactory(KStreamBuilder kStreamBuilder, + public KStreamBoundElementFactory kafkaStreamBindableTargetFactory(StreamsBuilder kStreamBuilder, BindingServiceProperties bindingServiceProperties, CompositeMessageConverterFactory compositeMessageConverterFactory) { return new KStreamBoundElementFactory(kStreamBuilder, bindingServiceProperties, diff --git a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamInteractiveQueryIntegrationTests.java b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamInteractiveQueryIntegrationTests.java index 775987371..38087dbae 100644 --- a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamInteractiveQueryIntegrationTests.java +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamInteractiveQueryIntegrationTests.java @@ -42,8 +42,8 @@ import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; -import org.springframework.kafka.core.KStreamBuilderFactoryBean; import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.StreamsBuilderFactoryBean; import org.springframework.kafka.support.serializer.JsonSerde; import org.springframework.kafka.test.rule.KafkaEmbedded; import org.springframework.kafka.test.utils.KafkaTestUtils; @@ -118,7 +118,7 @@ public class KStreamInteractiveQueryIntegrationTests { public static class ProductCountApplication { @Autowired - private KStreamBuilderFactoryBean kStreamBuilderFactoryBean; + private StreamsBuilderFactoryBean kStreamBuilderFactoryBean; @StreamListener("input") @SendTo("output") @@ -134,14 +134,14 @@ public class KStreamInteractiveQueryIntegrationTests { } @Bean - public Foo foo(KStreamBuilderFactoryBean kStreamBuilderFactoryBean) { + public Foo foo(StreamsBuilderFactoryBean kStreamBuilderFactoryBean) { return new Foo(kStreamBuilderFactoryBean); } static class Foo { - KStreamBuilderFactoryBean kStreamBuilderFactoryBean; + StreamsBuilderFactoryBean kStreamBuilderFactoryBean; - Foo(KStreamBuilderFactoryBean kStreamBuilderFactoryBean) { + Foo(StreamsBuilderFactoryBean kStreamBuilderFactoryBean) { this.kStreamBuilderFactoryBean = kStreamBuilderFactoryBean; }