diff --git a/spring-cloud-stream-binder-kafka-core/pom.xml b/spring-cloud-stream-binder-kafka-core/pom.xml index b73a30269..ec387d26e 100644 --- a/spring-cloud-stream-binder-kafka-core/pom.xml +++ b/spring-cloud-stream-binder-kafka-core/pom.xml @@ -1,59 +1,68 @@ - - 4.0.0 - - - - org.apache.maven.plugins - maven-compiler-plugin - - 1.8 - 1.8 - - - - - - org.springframework.cloud - spring-cloud-stream-binder-kafka-parent - 2.0.0.BUILD-SNAPSHOT - - spring-cloud-stream-binder-kafka-core - Spring Cloud Stream Kafka Binder Core - http://projects.spring.io/spring-cloud - - Pivotal Software, Inc. - http://www.spring.io - - + + 4.0.0 + + + + org.apache.maven.plugins + maven-compiler-plugin + + 1.8 + 1.8 + + + + + + org.springframework.cloud + spring-cloud-stream-binder-kafka-parent + 2.0.0.BUILD-SNAPSHOT + + spring-cloud-stream-binder-kafka-core + Spring Cloud Stream Kafka Binder Core + http://projects.spring.io/spring-cloud + + Pivotal Software, Inc. + http://www.spring.io + + - + - - - org.springframework.cloud - spring-cloud-stream - - - org.apache.kafka - kafka-clients - - - org.springframework.integration - spring-integration-kafka - ${spring-integration-kafka.version} - - - org.apache.avro - avro-compiler - - - - - log4j - log4j - 1.2.17 - - + + + org.springframework.cloud + spring-cloud-stream + + + org.apache.kafka + kafka-clients + + + org.springframework.integration + spring-integration-kafka + + + org.springframework.boot + spring-boot-test + test + + + org.springframework.kafka + spring-kafka-test + test + + + org.springframework.cloud + spring-cloud-stream-binder-test + test + + + log4j + log4j + 1.2.17 + + 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 0b54a6a21..2a7a0b0e7 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 @@ -1,5 +1,5 @@ /* - * Copyright 2014-2017 the original author or authors. + * Copyright 2014-2018 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. @@ -26,6 +26,7 @@ import java.util.concurrent.TimeUnit; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.clients.admin.AdminClient; import org.apache.kafka.clients.admin.AdminClientConfig; import org.apache.kafka.clients.admin.CreatePartitionsResult; @@ -74,25 +75,20 @@ public class KafkaTopicProvisioner implements ProvisioningProvider 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; + normalalizeBootPropsWithBinder(adminClientProperties, kafkaProperties, kafkaBinderConfigurationProperties); this.adminClient = AdminClient.create(adminClientProperties); } @@ -166,7 +162,9 @@ public class KafkaTopicProvisioner implements ProvisioningProvider adminProps, KafkaProperties bootProps, + KafkaBinderConfigurationProperties binderProps) { + // First deal with the outlier + String kafkaConnectionString = binderProps.getKafkaConnectionString(); + if (ObjectUtils.isEmpty(adminProps.get(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG)) + || !kafkaConnectionString.equals(binderProps.getDefaultKafkaConnectionString())) { + adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConnectionString); + } + // Now override any boot values with binder values + Map binderProperties = binderProps.getConfiguration(); + Set adminConfigNames = AdminClientConfig.configNames(); + binderProperties.forEach((key, value) -> { + if (key.equals(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG)) { + throw new IllegalStateException( + "Set binder bootstrap servers via the 'brokers' property, not 'configuration'"); + } + if (adminConfigNames.contains(key)) { + Object replaced = adminProps.put(key, value); + if (replaced != null && this.logger.isDebugEnabled()) { + logger.debug("Overrode boot property: [" + key + "], from: [" + replaced + "] to: [" + value + "]"); + } + } + }); + } + private ConsumerDestination createDlqIfNeedBe(String name, String group, ExtendedConsumerProperties properties, boolean anonymous, int partitions) { diff --git a/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisionerTests.java b/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisionerTests.java new file mode 100644 index 000000000..7334f7af4 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisionerTests.java @@ -0,0 +1,98 @@ +/* + * Copyright 2018 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.provisioning; + +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import org.apache.kafka.clients.CommonClientConfigs; +import org.apache.kafka.clients.admin.AdminClient; +import org.apache.kafka.clients.admin.AdminClientConfig; +import org.apache.kafka.common.config.SslConfigs; +import org.apache.kafka.common.network.SslChannelBuilder; +import org.junit.Test; + +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; +import org.springframework.core.io.ClassPathResource; +import org.springframework.kafka.test.utils.KafkaTestUtils; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; + + +/** + * @author Gary Russell + * @since 2.0 + * + */ +public class KafkaTopicProvisionerTests { + + @SuppressWarnings("rawtypes") + @Test + public void bootPropertiesOverriddenExceptServers() throws Exception { + KafkaProperties bootConfig = new KafkaProperties(); + bootConfig.getProperties().put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT"); + bootConfig.setBootstrapServers(Collections.singletonList("localhost:1234")); + KafkaBinderConfigurationProperties binderConfig = new KafkaBinderConfigurationProperties(); + binderConfig.getConfiguration().put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SSL"); + ClassPathResource ts = new ClassPathResource("test.truststore.ks"); + binderConfig.getConfiguration().put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, ts.getFile().getAbsolutePath()); + binderConfig.setBrokers("localhost:9092"); + KafkaTopicProvisioner provisioner = new KafkaTopicProvisioner(binderConfig, bootConfig); + AdminClient adminClient = KafkaTestUtils.getPropertyValue(provisioner, "adminClient", AdminClient.class); + assertThat(KafkaTestUtils.getPropertyValue(adminClient, "client.selector.channelBuilder")).isInstanceOf(SslChannelBuilder.class); + Map configs = KafkaTestUtils.getPropertyValue(adminClient, "client.selector.channelBuilder.configs", Map.class); + assertThat(((List) configs.get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG)).get(0)).isEqualTo("localhost:1234"); + } + + @SuppressWarnings("rawtypes") + @Test + public void bootPropertiesOverriddenIncludingServers() throws Exception { + KafkaProperties bootConfig = new KafkaProperties(); + bootConfig.getProperties().put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT"); + bootConfig.setBootstrapServers(Collections.singletonList("localhost:9092")); + KafkaBinderConfigurationProperties binderConfig = new KafkaBinderConfigurationProperties(); + binderConfig.getConfiguration().put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SSL"); + ClassPathResource ts = new ClassPathResource("test.truststore.ks"); + binderConfig.getConfiguration().put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, ts.getFile().getAbsolutePath()); + binderConfig.setBrokers("localhost:1234"); + KafkaTopicProvisioner provisioner = new KafkaTopicProvisioner(binderConfig, bootConfig); + AdminClient adminClient = KafkaTestUtils.getPropertyValue(provisioner, "adminClient", AdminClient.class); + assertThat(KafkaTestUtils.getPropertyValue(adminClient, "client.selector.channelBuilder")).isInstanceOf(SslChannelBuilder.class); + Map configs = KafkaTestUtils.getPropertyValue(adminClient, "client.selector.channelBuilder.configs", Map.class); + assertThat(((List) configs.get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG)).get(0)).isEqualTo("localhost:1234"); + } + + @SuppressWarnings("rawtypes") + @Test + public void brokersInvalid() throws Exception { + KafkaProperties bootConfig = new KafkaProperties(); + KafkaBinderConfigurationProperties binderConfig = new KafkaBinderConfigurationProperties(); + binderConfig.getConfiguration().put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "localhost:1234"); + try { + new KafkaTopicProvisioner(binderConfig, bootConfig); + fail("Expected illegal state"); + } + catch (IllegalStateException e) { + assertThat(e.getMessage()) + .isEqualTo("Set binder bootstrap servers via the 'brokers' property, not 'configuration'"); + } + } + +} diff --git a/spring-cloud-stream-binder-kafka-core/src/test/resources/test.truststore.ks b/spring-cloud-stream-binder-kafka-core/src/test/resources/test.truststore.ks new file mode 100644 index 000000000..24ead4bc6 Binary files /dev/null and b/spring-cloud-stream-binder-kafka-core/src/test/resources/test.truststore.ks differ