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; }