diff --git a/pom.xml b/pom.xml index b961acbc6..97095761a 100644 --- a/pom.xml +++ b/pom.xml @@ -21,8 +21,6 @@ spring-cloud-stream-binder-kafka spring-cloud-starter-stream-kafka spring-cloud-stream-binder-kafka-docs - spring-cloud-stream-binder-kafka-0.9-test - spring-cloud-stream-binder-kafka-0.10.0-test spring-cloud-stream-binder-kafka-0.10.1-test spring-cloud-stream-binder-kafka-0.10.2-test spring-cloud-stream-binder-kafka-core diff --git a/spring-cloud-stream-binder-kafka-0.10.0-test/pom.xml b/spring-cloud-stream-binder-kafka-0.10.0-test/pom.xml deleted file mode 100644 index c3ae29af4..000000000 --- a/spring-cloud-stream-binder-kafka-0.10.0-test/pom.xml +++ /dev/null @@ -1,115 +0,0 @@ - - - 4.0.0 - - org.springframework.cloud - spring-cloud-stream-binder-kafka-parent - 1.3.0.BUILD-SNAPSHOT - - spring-cloud-stream-binder-kafka-0.10.0-test - Spring Cloud Stream Kafka Binder 0.10.0 Tests - http://projects.spring.io/spring-cloud - - Pivotal Software, Inc. - http://www.spring.io - - - ${basedir}/../.. - - 0.10.0.1 - 1.1.2.RELEASE - 2.1.0.RELEASE - - - - - org.springframework.cloud - spring-cloud-stream-binder-kafka-core - 1.3.0.BUILD-SNAPSHOT - - - org.springframework.cloud - spring-cloud-stream-binder-kafka - ${project.version} - test - - - org.springframework.kafka - spring-kafka - test - - - org.apache.kafka - kafka_2.11 - test - - - org.slf4j - slf4j-log4j12 - - - - - org.apache.kafka - kafka-clients - test - - - org.springframework.kafka - spring-kafka-test - test - - - org.springframework.integration - spring-integration-kafka - - - org.springframework.cloud - spring-cloud-stream-binder-kafka - ${project.version} - test-jar - test - - - org.springframework.cloud - spring-cloud-stream-binder-kafka-0.10.2-test - ${project.version} - test-jar - test - - - org.springframework.cloud - spring-cloud-stream-binder-test - test - - - org.springframework.cloud - spring-cloud-stream-schema - ${spring-cloud-stream.version} - test - - - io.confluent - kafka-avro-serializer - 3.0.1 - test - - - io.confluent - kafka-schema-registry - 3.0.1 - test - - - - - - confluent - http://packages.confluent.io/maven/ - - - - diff --git a/spring-cloud-stream-binder-kafka-0.10.0-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka_0_10_0_BinderTests.java b/spring-cloud-stream-binder-kafka-0.10.0-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka_0_10_0_BinderTests.java deleted file mode 100644 index e999627f5..000000000 --- a/spring-cloud-stream-binder-kafka-0.10.0-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka_0_10_0_BinderTests.java +++ /dev/null @@ -1,28 +0,0 @@ -/* - * Copyright 2014-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; - -/** - * Integration tests for the {@link KafkaMessageChannelBinder}. - * - * This test specifically tests for the 0.10.0.x version of Kafka. - * - * @author Soby Chacko - */ -public class Kafka_0_10_0_BinderTests extends Kafka_0_10_2_BinderTests { - -} diff --git a/spring-cloud-stream-binder-kafka-0.9-test/pom.xml b/spring-cloud-stream-binder-kafka-0.9-test/pom.xml deleted file mode 100644 index 0995694dc..000000000 --- a/spring-cloud-stream-binder-kafka-0.9-test/pom.xml +++ /dev/null @@ -1,78 +0,0 @@ - - - 4.0.0 - - org.springframework.cloud - spring-cloud-stream-binder-kafka-parent - 1.3.0.BUILD-SNAPSHOT - - spring-cloud-stream-binder-kafka-0.9-test - Spring Cloud Stream Kafka Binder 0.9 Tests - http://projects.spring.io/spring-cloud - - Pivotal Software, Inc. - http://www.spring.io - - - ${basedir}/../.. - - 0.9.0.1 - 1.0.5.RELEASE - 2.0.1.RELEASE - - - - - org.springframework.cloud - spring-cloud-stream-binder-kafka - ${project.version} - test - - - org.springframework.kafka - spring-kafka - test - - - org.apache.kafka - kafka_2.11 - test - - - org.slf4j - slf4j-log4j12 - - - - - org.apache.kafka - kafka-clients - test - - - org.springframework.kafka - spring-kafka-test - test - - - org.springframework.integration - spring-integration-kafka - - - org.springframework.cloud - spring-cloud-stream-binder-kafka - ${project.version} - test-jar - test - - - org.springframework.cloud - spring-cloud-stream-binder-test - test - - - - diff --git a/spring-cloud-stream-binder-kafka-0.9-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09TestBinder.java b/spring-cloud-stream-binder-kafka-0.9-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09TestBinder.java deleted file mode 100644 index adb53e52a..000000000 --- a/spring-cloud-stream-binder-kafka-0.9-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09TestBinder.java +++ /dev/null @@ -1,60 +0,0 @@ -/* - * Copyright 2015-2016 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.binder.kafka; - -import org.springframework.cloud.stream.binder.kafka.admin.AdminUtilsOperation; -import org.springframework.cloud.stream.binder.kafka.admin.Kafka09AdminUtilsOperation; -import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; -import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; -import org.springframework.context.support.GenericApplicationContext; -import org.springframework.kafka.support.LoggingProducerListener; -import org.springframework.kafka.support.ProducerListener; - -/** - * Test support class for {@link KafkaMessageChannelBinder}. Creates a binder that uses - * an embedded Kafka cluster. - * @author Eric Bottard - * @author Marius Bogoevici - * @author David Turanski - * @author Gary Russell - * @author Soby Chacko - */ -public class Kafka09TestBinder extends AbstractKafkaTestBinder { - - public Kafka09TestBinder(KafkaBinderConfigurationProperties binderConfiguration) { - try { - AdminUtilsOperation adminUtilsOperation = new Kafka09AdminUtilsOperation(); - KafkaTopicProvisioner provisioningProvider = - new KafkaTopicProvisioner(binderConfiguration, adminUtilsOperation); - provisioningProvider.afterPropertiesSet(); - - KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(binderConfiguration, provisioningProvider); - binder.setCodec(getCodec()); - ProducerListener producerListener = new LoggingProducerListener(); - binder.setProducerListener(producerListener); - GenericApplicationContext context = new GenericApplicationContext(); - context.refresh(); - binder.setApplicationContext(context); - binder.afterPropertiesSet(); - this.setBinder(binder); - } - catch (Exception e) { - throw new RuntimeException(e); - } - } - -} diff --git a/spring-cloud-stream-binder-kafka-0.9-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka_09_BinderTests.java b/spring-cloud-stream-binder-kafka-0.9-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka_09_BinderTests.java deleted file mode 100644 index e9c470ce7..000000000 --- a/spring-cloud-stream-binder-kafka-0.9-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka_09_BinderTests.java +++ /dev/null @@ -1,158 +0,0 @@ -/* - * Copyright 2014-2016 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.binder.kafka; - -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Properties; - -import kafka.utils.ZKStringSerializer$; -import kafka.utils.ZkUtils; -import org.I0Itec.zkclient.ZkClient; -import org.apache.kafka.clients.consumer.ConsumerConfig; -import org.apache.kafka.common.serialization.ByteArrayDeserializer; -import org.apache.kafka.common.serialization.Deserializer; -import org.junit.Before; -import org.junit.ClassRule; - -import org.springframework.cloud.stream.binder.Binder; -import org.springframework.cloud.stream.binder.Spy; -import org.springframework.cloud.stream.binder.kafka.admin.Kafka09AdminUtilsOperation; -import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; -import org.springframework.kafka.core.ConsumerFactory; -import org.springframework.kafka.core.DefaultKafkaConsumerFactory; -import org.springframework.kafka.support.KafkaHeaders; -import org.springframework.kafka.test.core.BrokerAddress; -import org.springframework.kafka.test.rule.KafkaEmbedded; - -/** - * Integration tests for the {@link KafkaMessageChannelBinder}. - * - * @author Eric Bottard - * @author Marius Bogoevici - * @author Mark Fisher - * @author Ilayaperumal Gopinathan - */ -public class Kafka_09_BinderTests extends KafkaBinderTests { - - private final String CLASS_UNDER_TEST_NAME = KafkaMessageChannelBinder.class.getSimpleName(); - - @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 10); - - private Kafka09TestBinder binder; - - private Kafka09AdminUtilsOperation adminUtilsOperation = new Kafka09AdminUtilsOperation(); - - @Override - protected void binderBindUnbindLatency() throws InterruptedException { - Thread.sleep(500); - } - - @Override - protected Kafka09TestBinder getBinder() { - if (binder == null) { - KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); - binder = new Kafka09TestBinder(binderConfiguration); - } - return binder; - } - - protected KafkaBinderConfigurationProperties createConfigurationProperties() { - KafkaBinderConfigurationProperties binderConfiguration = new KafkaBinderConfigurationProperties(); - BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses(); - List bAddresses = new ArrayList<>(); - for (BrokerAddress bAddress : brokerAddresses) { - bAddresses.add(bAddress.toString()); - } - String[] foo = new String[bAddresses.size()]; - binderConfiguration.setBrokers(bAddresses.toArray(foo)); - binderConfiguration.setZkNodes(embeddedKafka.getZookeeperConnectionString()); - return binderConfiguration; - } - - @Override - protected int partitionSize(String topic) { - return consumerFactory().createConsumer().partitionsFor(topic).size(); - } - - @Override - protected ZkUtils getZkUtils(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) { - final ZkClient zkClient = new ZkClient(kafkaBinderConfigurationProperties.getZkConnectionString(), - kafkaBinderConfigurationProperties.getZkSessionTimeout(), kafkaBinderConfigurationProperties.getZkConnectionTimeout(), - ZKStringSerializer$.MODULE$); - - return new ZkUtils(zkClient, null, false); - } - - @Override - protected void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions, int replicationFactor, Properties topicConfig) { - adminUtilsOperation.invokeCreateTopic(zkUtils, topic, partitions, replicationFactor, new Properties()); - } - - @Override - protected int invokePartitionSize(String topic, ZkUtils zkUtils) { - return adminUtilsOperation.partitionSize(topic, zkUtils); - } - - @Override - public String getKafkaOffsetHeaderKey() { - return KafkaHeaders.OFFSET; - } - - @Override - protected Binder getBinder(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) { - return new Kafka09TestBinder(kafkaBinderConfigurationProperties); - } - - @Before - public void init() { - String multiplier = System.getenv("KAFKA_TIMEOUT_MULTIPLIER"); - if (multiplier != null) { - timeoutMultiplier = Double.parseDouble(multiplier); - } - } - - @Override - protected boolean usesExplicitRouting() { - return false; - } - - @Override - protected String getClassUnderTestName() { - return CLASS_UNDER_TEST_NAME; - } - - @Override - public Spy spyOn(final String name) { - throw new UnsupportedOperationException("'spyOn' is not used by Kafka tests"); - } - - private ConsumerFactory consumerFactory() { - Map props = new HashMap<>(); - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, configurationProperties.getKafkaConnectionString()); - props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); - Deserializer valueDecoder = new ByteArrayDeserializer(); - Deserializer keyDecoder = new ByteArrayDeserializer(); - - return new DefaultKafkaConsumerFactory<>(props, keyDecoder, valueDecoder); - } - -} 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 40cebc2bd..235274c5a 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 @@ -16,10 +16,18 @@ package org.springframework.cloud.stream.binder.kafka.properties; +import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.producer.ProducerConfig; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; /** @@ -31,13 +39,16 @@ import org.springframework.util.StringUtils; @ConfigurationProperties(prefix = "spring.cloud.stream.kafka.binder") public class KafkaBinderConfigurationProperties { - private String[] zkNodes = new String[] {"localhost"}; + @Autowired(required = false) + private KafkaProperties kafkaProperties; - private Map configuration = new HashMap<>(); + private String[] zkNodes = new String[] { "localhost" }; + + private Map configuration = new HashMap<>(); private String defaultZkPort = "2181"; - private String[] brokers = new String[] {"localhost"}; + private String[] brokers = new String[] { "localhost" }; private String defaultBrokerPort = "9092"; @@ -249,16 +260,78 @@ public class KafkaBinderConfigurationProperties { this.socketBufferSize = socketBufferSize; } - public Map getConfiguration() { + public Map getConfiguration() { return configuration; } - public void setConfiguration(Map configuration) { + public void setConfiguration(Map configuration) { this.configuration = configuration; } + public Map getConsumerConfiguration() { + Map consumerConfiguration = new HashMap<>(); + // If Spring Boot Kafka properties are present, add them with lowest precedence + if (this.kafkaProperties != null) { + consumerConfiguration.putAll(this.kafkaProperties.buildConsumerProperties()); + } + // Copy configured binder properties + for (Map.Entry configurationEntry : this.configuration.entrySet()) { + if (ConsumerConfig.configNames().contains(configurationEntry.getKey())) { + consumerConfiguration.put(configurationEntry.getKey(), configurationEntry.getValue()); + } + } + // Override Spring Boot bootstrap server setting if left to default with the value + // configured in the binder + if (ObjectUtils.isEmpty(consumerConfiguration.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG))) { + consumerConfiguration.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, getKafkaConnectionString()); + } + else { + Object boostrapServersConfig = consumerConfiguration.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG); + if (boostrapServersConfig instanceof List) { + @SuppressWarnings("unchecked") + List bootStrapServers = (List) consumerConfiguration + .get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG); + if (bootStrapServers.size() == 1 && bootStrapServers.get(0).equals("localhost:9092")) { + consumerConfiguration.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, getKafkaConnectionString()); + } + } + } + return Collections.unmodifiableMap(consumerConfiguration); + } + + public Map getProducerConfiguration() { + Map producerConfiguration = new HashMap<>(); + // If Spring Boot Kafka properties are present, add them with lowest precedence + if (this.kafkaProperties != null) { + producerConfiguration.putAll(this.kafkaProperties.buildProducerProperties()); + } + // Copy configured binder properties + for (Map.Entry configurationEntry : configuration.entrySet()) { + if (ProducerConfig.configNames().contains(configurationEntry.getKey())) { + producerConfiguration.put(configurationEntry.getKey(), configurationEntry.getValue()); + } + } + // Override Spring Boot bootstrap server setting if left to default with the value + // configured in the binder + if (ObjectUtils.isEmpty(producerConfiguration.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) { + producerConfiguration.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, getKafkaConnectionString()); + } + else { + Object boostrapServersConfig = producerConfiguration.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG); + if (boostrapServersConfig instanceof List) { + @SuppressWarnings("unchecked") + List bootStrapServers = (List) producerConfiguration + .get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG); + if (bootStrapServers.size() == 1 && bootStrapServers.get(0).equals("localhost:9092")) { + producerConfiguration.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, getKafkaConnectionString()); + } + } + } + return Collections.unmodifiableMap(producerConfiguration); + } + public JaasLoginModuleConfiguration getJaas() { - return jaas; + return this.jaas; } public void setJaas(JaasLoginModuleConfiguration jaas) { diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java index d3d5c7c82..b8098e88c 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java @@ -112,8 +112,8 @@ public class KafkaBinderMetrics implements PublicMetrics { Map props = new HashMap<>(); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); - if (!ObjectUtils.isEmpty(binderConfigurationProperties.getConfiguration())) { - props.putAll(binderConfigurationProperties.getConfiguration()); + if (!ObjectUtils.isEmpty(binderConfigurationProperties.getConsumerConfiguration())) { + props.putAll(binderConfigurationProperties.getConsumerConfiguration()); } if (!props.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) { props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, 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 ec32fc27a..87ef2ab48 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 @@ -180,8 +180,8 @@ public class KafkaMessageChannelBinder extends props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); props.put(ProducerConfig.ACKS_CONFIG, String.valueOf(this.configurationProperties.getRequiredAcks())); - if (!ObjectUtils.isEmpty(configurationProperties.getConfiguration())) { - props.putAll(configurationProperties.getConfiguration()); + if (!ObjectUtils.isEmpty(configurationProperties.getProducerConfiguration())) { + props.putAll(configurationProperties.getProducerConfiguration()); } if (ObjectUtils.isEmpty(props.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) { props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString()); @@ -337,8 +337,8 @@ public class KafkaMessageChannelBinder extends props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, anonymous ? "latest" : "earliest"); props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup); - if (!ObjectUtils.isEmpty(configurationProperties.getConfiguration())) { - props.putAll(configurationProperties.getConfiguration()); + if (!ObjectUtils.isEmpty(configurationProperties.getConsumerConfiguration())) { + props.putAll(configurationProperties.getConsumerConfiguration()); } if (ObjectUtils.isEmpty(props.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG))) { props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString()); 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 6dc5c43b5..5f0f55dd0 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 @@ -18,15 +18,11 @@ package org.springframework.cloud.stream.binder.kafka.config; import java.io.IOException; import java.util.HashMap; -import java.util.List; import java.util.Map; -import javax.annotation.PostConstruct; - import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.kafka.clients.consumer.ConsumerConfig; -import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.serialization.ByteArrayDeserializer; import org.apache.kafka.common.utils.AppInfoParser; @@ -35,7 +31,6 @@ import org.springframework.boot.actuate.endpoint.PublicMetrics; 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; @@ -76,8 +71,7 @@ import org.springframework.util.ObjectUtils; */ @Configuration @ConditionalOnMissingBean(Binder.class) -@Import({ KryoCodecAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class, - KafkaBinderConfiguration.KafkaPropertiesConfiguration.class }) +@Import({ KryoCodecAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class}) @EnableConfigurationProperties({ KafkaBinderConfigurationProperties.class, KafkaExtendedBindingProperties.class }) public class KafkaBinderConfiguration { @@ -127,8 +121,8 @@ public class KafkaBinderConfiguration { Map props = new HashMap<>(); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); - if (!ObjectUtils.isEmpty(configurationProperties.getConfiguration())) { - props.putAll(configurationProperties.getConfiguration()); + if (!ObjectUtils.isEmpty(configurationProperties.getConsumerConfiguration())) { + props.putAll(configurationProperties.getConsumerConfiguration()); } if (!props.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) { props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString()); @@ -185,52 +179,4 @@ public class KafkaBinderConfiguration { private JaasLoginModuleConfiguration zookeeper; } - - @ConditionalOnClass(name = "org.springframework.boot.autoconfigure.kafka.KafkaProperties") - public static class KafkaPropertiesConfiguration { - - // KafkaProperties can still be unavailable if KafkaAutoConfiguration is disabled. - @Autowired(required = false) - private KafkaProperties kafkaProperties; - - @Autowired - private KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties; - - @PostConstruct - public void init() { - Map configuration = this.kafkaBinderConfigurationProperties.getConfiguration(); - if (this.kafkaProperties != null) { - for (Map.Entry properties : this.kafkaProperties.getProperties().entrySet()) { - if (!configuration.containsKey(properties.getKey())) { - configuration.put(properties.getKey(), properties.getValue()); - } - } - for (Map.Entry producerProperties : this.kafkaProperties.buildProducerProperties() - .entrySet()) { - if (!configuration.containsKey(producerProperties.getKey())) { - configuration.put(producerProperties.getKey(), producerProperties.getValue()); - } - } - for (Map.Entry consumerProperties : this.kafkaProperties.buildConsumerProperties() - .entrySet()) { - if (!configuration.containsKey(consumerProperties.getKey())) { - configuration.put(consumerProperties.getKey(), consumerProperties.getValue()); - } - } - if (ObjectUtils.isEmpty(configuration.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) { - configuration.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, - kafkaBinderConfigurationProperties.getKafkaConnectionString()); - } - else { - @SuppressWarnings("unchecked") - List bootStrapServers = (List) configuration - .get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG); - if (bootStrapServers.size() == 1 && bootStrapServers.get(0).equals("localhost:9092")) { - configuration.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, - kafkaBinderConfigurationProperties.getKafkaConnectionString()); - } - } - } - } - } } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderAutoConfigurationPropertiesTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderAutoConfigurationPropertiesTest.java index a26665bee..5f21fafdc 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderAutoConfigurationPropertiesTest.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderAutoConfigurationPropertiesTest.java @@ -77,7 +77,9 @@ public class KafkaBinderAutoConfigurationPropertiesTest { producerFactory); assertTrue(producerConfigs.get("batch.size").equals(10)); assertTrue(producerConfigs.get("key.serializer").equals(LongSerializer.class)); + assertTrue(producerConfigs.get("key.deserializer") == null); assertTrue(producerConfigs.get("value.serializer").equals(LongSerializer.class)); + assertTrue(producerConfigs.get("value.deserializer") == null); assertTrue(producerConfigs.get("compression.type").equals("snappy")); List bootstrapServers = new ArrayList<>(); bootstrapServers.add("10.98.09.199:9092"); @@ -96,7 +98,9 @@ public class KafkaBinderAutoConfigurationPropertiesTest { Map consumerConfigs = (Map) ReflectionUtils.getField(consumerFactoryConfigField, consumerFactory); assertTrue(consumerConfigs.get("key.deserializer").equals(LongDeserializer.class)); + assertTrue(consumerConfigs.get("key.serializer") == null); assertTrue(consumerConfigs.get("value.deserializer").equals(LongDeserializer.class)); + assertTrue(consumerConfigs.get("value.serialized") == null); assertTrue(consumerConfigs.get("group.id").equals("groupIdFromBootConfig")); assertTrue(consumerConfigs.get("auto.offset.reset").equals("earliest")); assertTrue((((List) consumerConfigs.get("bootstrap.servers")).containsAll(bootstrapServers))); 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 50aab787f..bd6541051 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 @@ -1299,7 +1299,7 @@ public abstract class KafkaBinderTests extends Binding binding = null; try { KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - Map propertiesToOverride = configurationProperties.getConfiguration(); + 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); 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 3e4bce62e..36a3b8f8b 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 @@ -67,7 +67,7 @@ public class KafkaBinderUnitTests { // binder level setting binderConfigurationProperties.setConfiguration( - Collections.singletonMap(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")); + Collections.singletonMap(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")); factory = method.invoke(binder, false, "foo", ecp); configs = TestUtils.getPropertyValue(factory, "configs", Map.class); assertThat(configs.get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)).isEqualTo("latest");