From 6a31e9c94f604f7054dd4f50f905da4d8edd5c78 Mon Sep 17 00:00:00 2001 From: Ilayaperumal Gopinathan Date: Tue, 21 Feb 2017 19:26:22 +0530 Subject: [PATCH] Support KafkaProperties from Spring Boot autoconfiguration - If KafkaProperties is available from KafkaAutoConfiguration, retrieve the kafka properties and set as `KafkaBinderConfigurationProperties`' configuration property. This way, these properties are available at the top level and per-binding properties could still override when setting producer/consumer properties during binding operation - Add tests to verify the scenarios Resolves #73 Fix deprecation warning Remove unused constant Polishing --- .../KafkaBinderConfigurationProperties.java | 6 +- .../KafkaBinderEnvironmentPostProcessor.java | 37 ++++++-- .../kafka/KafkaMessageChannelBinder.java | 84 ++++++++-------- .../config/KafkaBinderConfiguration.java | 53 ++++++++++- ...BinderAutoConfigurationPropertiesTest.java | 95 +++++++++++++++++++ ...afkaBinderConfigurationPropertiesTest.java | 89 +++++++++++++++++ .../kafka/KafkaBinderConfigurationTest.java | 6 +- ...afkaBinderJaasInitializerListenerTest.java | 1 + .../stream/binder/kafka/KafkaBinderTests.java | 2 +- .../bootstrap/KafkaBinderBootstrapTest.java | 47 +++++++++ .../binder-config-autoconfig.properties | 7 ++ .../test/resources/binder-config.properties | 1 + 12 files changed, 375 insertions(+), 53 deletions(-) create mode 100644 spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderAutoConfigurationPropertiesTest.java create mode 100644 spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationPropertiesTest.java create mode 100644 spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderBootstrapTest.java create mode 100644 spring-cloud-stream-binder-kafka/src/test/resources/binder-config-autoconfig.properties create mode 100644 spring-cloud-stream-binder-kafka/src/test/resources/binder-config.properties 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 4bb30d952..40cebc2bd 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 @@ -33,7 +33,7 @@ public class KafkaBinderConfigurationProperties { private String[] zkNodes = new String[] {"localhost"}; - private Map configuration = new HashMap<>(); + private Map configuration = new HashMap<>(); private String defaultZkPort = "2181"; @@ -249,11 +249,11 @@ 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; } diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderEnvironmentPostProcessor.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderEnvironmentPostProcessor.java index 6f59c5f0e..0e24318b1 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderEnvironmentPostProcessor.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderEnvironmentPostProcessor.java @@ -19,6 +19,9 @@ package org.springframework.cloud.stream.binder.kafka; import java.util.HashMap; import java.util.Map; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.kafka.common.serialization.ByteArraySerializer; + import org.springframework.boot.SpringApplication; import org.springframework.boot.env.EnvironmentPostProcessor; import org.springframework.core.env.ConfigurableEnvironment; @@ -32,13 +35,35 @@ import org.springframework.core.env.MapPropertySource; */ public class KafkaBinderEnvironmentPostProcessor implements EnvironmentPostProcessor { + public final static String SPRING_KAFKA = "spring.kafka"; + + public final static String SPRING_KAFKA_PRODUCER = SPRING_KAFKA + ".producer"; + + public final static String SPRING_KAFKA_CONSUMER = SPRING_KAFKA + ".consumer"; + + public final static String SPRING_KAFKA_PRODUCER_KEY_SERIALIZER = SPRING_KAFKA_PRODUCER + "." + "keySerializer"; + + public final static String SPRING_KAFKA_PRODUCER_VALUE_SERIALIZER = SPRING_KAFKA_PRODUCER + "." + "valueSerializer"; + + public final static String SPRING_KAFKA_CONSUMER_KEY_DESERIALIZER = SPRING_KAFKA_CONSUMER + "." + "keyDeserializer"; + + public final static String SPRING_KAFKA_CONSUMER_VALUE_DESERIALIZER = SPRING_KAFKA_CONSUMER + "." + "valueDeserializer"; + + private static final String KAFKA_BINDER_DEFAULT_PROPERTIES = "kafkaBinderDefaultProperties"; + @Override public void postProcessEnvironment(ConfigurableEnvironment environment, SpringApplication application) { - Map propertiesToAdd = new HashMap<>(); - propertiesToAdd.put("logging.pattern.console", "%d{ISO8601} %5p %t %c{2}:%L - %m%n"); - propertiesToAdd.put("logging.level.org.I0Itec.zkclient", "ERROR"); - propertiesToAdd.put("logging.level.kafka.server.KafkaConfig", "ERROR"); - propertiesToAdd.put("logging.level.kafka.admin.AdminClient.AdminConfig", "ERROR"); - environment.getPropertySources().addLast(new MapPropertySource("kafkaBinderLogConfig", propertiesToAdd)); + if (!environment.getPropertySources().contains(KAFKA_BINDER_DEFAULT_PROPERTIES)) { + Map kafkaBinderDefaultProperties = new HashMap<>(); + kafkaBinderDefaultProperties.put("logging.pattern.console", "%d{ISO8601} %5p %t %c{2}:%L - %m%n"); + kafkaBinderDefaultProperties.put("logging.level.org.I0Itec.zkclient", "ERROR"); + kafkaBinderDefaultProperties.put("logging.level.kafka.server.KafkaConfig", "ERROR"); + kafkaBinderDefaultProperties.put("logging.level.kafka.admin.AdminClient.AdminConfig", "ERROR"); + kafkaBinderDefaultProperties.put(SPRING_KAFKA_PRODUCER_KEY_SERIALIZER, ByteArraySerializer.class.getName()); + kafkaBinderDefaultProperties.put(SPRING_KAFKA_PRODUCER_VALUE_SERIALIZER, ByteArraySerializer.class.getName()); + kafkaBinderDefaultProperties.put(SPRING_KAFKA_CONSUMER_KEY_DESERIALIZER, ByteArrayDeserializer.class.getName()); + kafkaBinderDefaultProperties.put(SPRING_KAFKA_CONSUMER_VALUE_DESERIALIZER, ByteArrayDeserializer.class.getName()); + environment.getPropertySources().addLast(new MapPropertySource(KAFKA_BINDER_DEFAULT_PROPERTIES, kafkaBinderDefaultProperties)); + } } } 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 ae54100be..84955306d 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 @@ -98,7 +98,7 @@ public class KafkaMessageChannelBinder extends private final Map> topicsInUse = new HashMap<>(); public KafkaMessageChannelBinder(KafkaBinderConfigurationProperties configurationProperties, - KafkaTopicProvisioner provisioningProvider) { + KafkaTopicProvisioner provisioningProvider) { super(false, headersToMap(configurationProperties), provisioningProvider); this.configurationProperties = configurationProperties; } @@ -143,7 +143,7 @@ public class KafkaMessageChannelBinder extends @Override protected MessageHandler createProducerMessageHandler(final ProducerDestination destination, - ExtendedProducerProperties producerProperties) throws Exception { + ExtendedProducerProperties producerProperties) throws Exception { final DefaultKafkaProducerFactory producerFB = getProducerFactory(producerProperties); Collection partitions = provisioningProvider.getPartitionsForTopic(producerProperties.getPartitionCount(), new Callable>() { @@ -171,20 +171,27 @@ public class KafkaMessageChannelBinder extends private DefaultKafkaProducerFactory getProducerFactory( ExtendedProducerProperties producerProperties) { Map props = new HashMap<>(); - if (!ObjectUtils.isEmpty(configurationProperties.getConfiguration())) { - props.putAll(configurationProperties.getConfiguration()); - } - props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString()); props.put(ProducerConfig.RETRIES_CONFIG, 0); - props.put(ProducerConfig.BATCH_SIZE_CONFIG, String.valueOf(producerProperties.getExtension().getBufferSize())); props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); props.put(ProducerConfig.ACKS_CONFIG, String.valueOf(this.configurationProperties.getRequiredAcks())); - props.put(ProducerConfig.LINGER_MS_CONFIG, - String.valueOf(producerProperties.getExtension().getBatchTimeout())); - props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, - producerProperties.getExtension().getCompressionType().toString()); + if (!ObjectUtils.isEmpty(configurationProperties.getConfiguration())) { + props.putAll(configurationProperties.getConfiguration()); + } + if (ObjectUtils.isEmpty(props.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) { + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString()); + } + if (ObjectUtils.isEmpty(props.get(ProducerConfig.BATCH_SIZE_CONFIG))) { + props.put(ProducerConfig.BATCH_SIZE_CONFIG, String.valueOf(producerProperties.getExtension().getBufferSize())); + } + if (ObjectUtils.isEmpty(props.get(ProducerConfig.LINGER_MS_CONFIG))) { + props.put(ProducerConfig.LINGER_MS_CONFIG, String.valueOf(producerProperties.getExtension().getBatchTimeout())); + } + if (ObjectUtils.isEmpty(props.get(ProducerConfig.COMPRESSION_TYPE_CONFIG))) { + props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, + producerProperties.getExtension().getCompressionType().toString()); + } if (!ObjectUtils.isEmpty(producerProperties.getExtension().getConfiguration())) { props.putAll(producerProperties.getExtension().getConfiguration()); } @@ -194,18 +201,14 @@ public class KafkaMessageChannelBinder extends @Override @SuppressWarnings("unchecked") protected MessageProducer createConsumerEndpoint(final ConsumerDestination destination, final String group, - ExtendedConsumerProperties properties) { + ExtendedConsumerProperties extendedConsumerProperties) { boolean anonymous = !StringUtils.hasText(group); - Assert.isTrue(!anonymous || !properties.getExtension().isEnableDlq(), + Assert.isTrue(!anonymous || !extendedConsumerProperties.getExtension().isEnableDlq(), "DLQ support is not available for anonymous subscriptions"); String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString() : group; - Map props = getConsumerConfig(anonymous, consumerGroup); - if (!ObjectUtils.isEmpty(properties.getExtension().getConfiguration())) { - props.putAll(properties.getExtension().getConfiguration()); - } - final ConsumerFactory consumerFactory = new DefaultKafkaConsumerFactory<>(props); - int partitionCount = properties.getInstanceCount() * properties.getConcurrency(); + final ConsumerFactory consumerFactory = createKafkaConsumerFactory(anonymous, consumerGroup, extendedConsumerProperties); + int partitionCount = extendedConsumerProperties.getInstanceCount() * extendedConsumerProperties.getConcurrency(); Collection allPartitions = provisioningProvider.getPartitionsForTopic(partitionCount, new Callable>() { @@ -217,15 +220,15 @@ public class KafkaMessageChannelBinder extends Collection listenedPartitions; - if (properties.getExtension().isAutoRebalanceEnabled() || - properties.getInstanceCount() == 1) { + if (extendedConsumerProperties.getExtension().isAutoRebalanceEnabled() || + extendedConsumerProperties.getInstanceCount() == 1) { listenedPartitions = allPartitions; } else { listenedPartitions = new ArrayList<>(); for (PartitionInfo partition : allPartitions) { // divide partitions across modules - if ((partition.partition() % properties.getInstanceCount()) == properties.getInstanceIndex()) { + if ((partition.partition() % extendedConsumerProperties.getInstanceCount()) == extendedConsumerProperties.getInstanceIndex()) { listenedPartitions.add(partition); } } @@ -236,9 +239,9 @@ public class KafkaMessageChannelBinder extends final TopicPartitionInitialOffset[] topicPartitionInitialOffsets = getTopicPartitionInitialOffsets( listenedPartitions); final ContainerProperties containerProperties = - anonymous || properties.getExtension().isAutoRebalanceEnabled() ? new ContainerProperties(destination.getName()) + anonymous || extendedConsumerProperties.getExtension().isAutoRebalanceEnabled() ? new ContainerProperties(destination.getName()) : new ContainerProperties(topicPartitionInitialOffsets); - int concurrency = Math.min(properties.getConcurrency(), listenedPartitions.size()); + int concurrency = Math.min(extendedConsumerProperties.getConcurrency(), listenedPartitions.size()); final ConcurrentMessageListenerContainer messageListenerContainer = new ConcurrentMessageListenerContainer( consumerFactory, containerProperties) { @@ -249,8 +252,8 @@ public class KafkaMessageChannelBinder extends } }; messageListenerContainer.setConcurrency(concurrency); - messageListenerContainer.getContainerProperties().setAckOnError(isAutoCommitOnError(properties)); - if (!properties.getExtension().isAutoCommitOffset()) { + messageListenerContainer.getContainerProperties().setAckOnError(isAutoCommitOnError(extendedConsumerProperties)); + if (!extendedConsumerProperties.getExtension().isAutoCommitOffset()) { messageListenerContainer.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL); } if (this.logger.isDebugEnabled()) { @@ -265,9 +268,9 @@ public class KafkaMessageChannelBinder extends new KafkaMessageDrivenChannelAdapter<>( messageListenerContainer); kafkaMessageDrivenChannelAdapter.setBeanFactory(this.getBeanFactory()); - final RetryTemplate retryTemplate = buildRetryTemplate(properties); + final RetryTemplate retryTemplate = buildRetryTemplate(extendedConsumerProperties); kafkaMessageDrivenChannelAdapter.setRetryTemplate(retryTemplate); - if (properties.getExtension().isEnableDlq()) { + if (extendedConsumerProperties.getExtension().isEnableDlq()) { DefaultKafkaProducerFactory producerFactory = getProducerFactory(new ExtendedProducerProperties<>(new KafkaProducerProperties())); final KafkaTemplate kafkaTemplate = new KafkaTemplate<>(producerFactory); messageListenerContainer.getContainerProperties().setErrorHandler(new ErrorHandler() { @@ -308,20 +311,25 @@ public class KafkaMessageChannelBinder extends return kafkaMessageDrivenChannelAdapter; } - private Map getConsumerConfig(boolean anonymous, String consumerGroup) { + private ConsumerFactory createKafkaConsumerFactory(boolean anonymous, String consumerGroup, + ExtendedConsumerProperties consumerProperties) { Map props = new HashMap<>(); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); + props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); + props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, anonymous ? "latest" : "earliest"); + props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100); if (!ObjectUtils.isEmpty(configurationProperties.getConfiguration())) { props.putAll(configurationProperties.getConfiguration()); } - props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString()); - props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); - props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup); - props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, - anonymous ? "latest" : "earliest"); - props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100); - return props; + if (ObjectUtils.isEmpty(props.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG))) { + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString()); + } + if (!ObjectUtils.isEmpty(consumerProperties.getExtension().getConfiguration())) { + props.putAll(consumerProperties.getExtension().getConfiguration()); + } + return new DefaultKafkaConsumerFactory<>(props); } private boolean isAutoCommitOnError(ExtendedConsumerProperties properties) { @@ -358,8 +366,8 @@ public class KafkaMessageChannelBinder extends private final DefaultKafkaProducerFactory producerFactory; private ProducerConfigurationMessageHandler(KafkaTemplate kafkaTemplate, String topic, - ExtendedProducerProperties producerProperties, - DefaultKafkaProducerFactory producerFactory) { + ExtendedProducerProperties producerProperties, + DefaultKafkaProducerFactory producerFactory) { super(kafkaTemplate); setTopicExpression(new LiteralExpression(topic)); setBeanFactory(KafkaMessageChannelBinder.this.getBeanFactory()); 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 fa7c25d4e..a01465534 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 @@ -17,15 +17,20 @@ package org.springframework.cloud.stream.binder.kafka.config; import java.io.IOException; +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.producer.ProducerConfig; import org.apache.kafka.common.utils.AppInfoParser; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration; 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; @@ -51,6 +56,7 @@ import org.springframework.core.type.AnnotatedTypeMetadata; import org.springframework.integration.codec.Codec; import org.springframework.kafka.support.LoggingProducerListener; import org.springframework.kafka.support.ProducerListener; +import org.springframework.util.ObjectUtils; /** * @author David Turanski @@ -61,7 +67,7 @@ import org.springframework.kafka.support.ProducerListener; */ @Configuration @ConditionalOnMissingBean(Binder.class) -@Import({KryoCodecAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class}) +@Import({KryoCodecAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class, KafkaBinderConfiguration.KafkaPropertiesConfiguration.class}) @EnableConfigurationProperties({KafkaBinderConfigurationProperties.class, KafkaExtendedBindingProperties.class}) public class KafkaBinderConfiguration { @@ -154,4 +160,47 @@ 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 new file mode 100644 index 000000000..9ec3b4a53 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderAutoConfigurationPropertiesTest.java @@ -0,0 +1,95 @@ +/* + * Copyright 2016-2017 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.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +import org.apache.kafka.common.serialization.LongDeserializer; +import org.apache.kafka.common.serialization.LongSerializer; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.test.context.TestPropertySource; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.ReflectionUtils; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +/** + * @author Ilayaperumal Gopinathan + */ +@RunWith(SpringJUnit4ClassRunner.class) +@SpringBootTest(classes = {KafkaBinderAutoConfigurationPropertiesTest.KafkaBinderConfigProperties.class, KafkaBinderConfiguration.class}) +@TestPropertySource(locations = "classpath:binder-config-autoconfig.properties") +public class KafkaBinderAutoConfigurationPropertiesTest { + + @Autowired + private KafkaMessageChannelBinder kafkaMessageChannelBinder; + + @Test + public void testKafkaBinderConfigurationWithKafkaProperties() throws Exception { + assertNotNull(this.kafkaMessageChannelBinder); + ExtendedProducerProperties producerProperties = new ExtendedProducerProperties<>(new KafkaProducerProperties()); + Method getProducerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod("getProducerFactory", ExtendedProducerProperties.class); + getProducerFactoryMethod.setAccessible(true); + DefaultKafkaProducerFactory producerFactory = (DefaultKafkaProducerFactory) getProducerFactoryMethod.invoke(this.kafkaMessageChannelBinder, producerProperties); + Field producerFactoryConfigField = ReflectionUtils.findField(DefaultKafkaProducerFactory.class, "configs", Map.class); + ReflectionUtils.makeAccessible(producerFactoryConfigField); + Map producerConfigs = (Map) ReflectionUtils.getField(producerFactoryConfigField, producerFactory); + assertTrue(producerConfigs.get("batch.size").equals(10)); + assertTrue(producerConfigs.get("key.serializer").equals(LongSerializer.class)); + assertTrue(producerConfigs.get("value.serializer").equals(LongSerializer.class)); + assertTrue(producerConfigs.get("compression.type").equals("snappy")); + List bootstrapServers = new ArrayList<>(); + bootstrapServers.add("10.98.09.199:9092"); + bootstrapServers.add("10.98.09.196:9092"); + assertTrue((((List) producerConfigs.get("bootstrap.servers")).containsAll(bootstrapServers))); + Method createKafkaConsumerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod("createKafkaConsumerFactory", boolean.class, String.class, ExtendedConsumerProperties.class); + createKafkaConsumerFactoryMethod.setAccessible(true); + ExtendedConsumerProperties consumerProperties = new ExtendedConsumerProperties<>(new KafkaConsumerProperties()); + DefaultKafkaConsumerFactory consumerFactory = (DefaultKafkaConsumerFactory) createKafkaConsumerFactoryMethod.invoke(this.kafkaMessageChannelBinder, true, "test", consumerProperties); + Field consumerFactoryConfigField = ReflectionUtils.findField(DefaultKafkaConsumerFactory.class, "configs", Map.class); + ReflectionUtils.makeAccessible(consumerFactoryConfigField); + Map consumerConfigs = (Map) ReflectionUtils.getField(consumerFactoryConfigField, consumerFactory); + assertTrue(consumerConfigs.get("key.deserializer").equals(LongDeserializer.class)); + assertTrue(consumerConfigs.get("value.deserializer").equals(LongDeserializer.class)); + assertTrue((((List) consumerConfigs.get("bootstrap.servers")).containsAll(bootstrapServers))); + } + + public static class KafkaBinderConfigProperties { + + @Bean + KafkaProperties kafkaProperties() { + return new KafkaProperties(); + } + } +} 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 new file mode 100644 index 000000000..f3521f7d7 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationPropertiesTest.java @@ -0,0 +1,89 @@ +/* + * Copyright 2016-2017 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.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.kafka.common.serialization.ByteArraySerializer; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.test.context.TestPropertySource; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.ReflectionUtils; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +/** + * @author Ilayaperumal Gopinathan + */ +@RunWith(SpringJUnit4ClassRunner.class) +@SpringBootTest(classes = {KafkaBinderConfiguration.class}) +@TestPropertySource(locations = "classpath:binder-config.properties") +public class KafkaBinderConfigurationPropertiesTest { + + @Autowired + private KafkaMessageChannelBinder kafkaMessageChannelBinder; + + @Test + public void testKafkaBinderConfigurationProperties() throws Exception { + assertNotNull(this.kafkaMessageChannelBinder); + KafkaProducerProperties kafkaProducerProperties = new KafkaProducerProperties(); + kafkaProducerProperties.setBufferSize(12345); + kafkaProducerProperties.setBatchTimeout(100); + kafkaProducerProperties.setCompressionType(KafkaProducerProperties.CompressionType.gzip); + ExtendedProducerProperties producerProperties = new ExtendedProducerProperties<>(kafkaProducerProperties); + Method getProducerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod("getProducerFactory", ExtendedProducerProperties.class); + getProducerFactoryMethod.setAccessible(true); + DefaultKafkaProducerFactory producerFactory = (DefaultKafkaProducerFactory) getProducerFactoryMethod.invoke(this.kafkaMessageChannelBinder, producerProperties); + Field producerFactoryConfigField = ReflectionUtils.findField(DefaultKafkaProducerFactory.class, "configs", Map.class); + ReflectionUtils.makeAccessible(producerFactoryConfigField); + Map producerConfigs = (Map) ReflectionUtils.getField(producerFactoryConfigField, producerFactory); + assertTrue(producerConfigs.get("batch.size").equals("12345")); + assertTrue(producerConfigs.get("linger.ms").equals("100")); + assertTrue(producerConfigs.get("key.serializer").equals(ByteArraySerializer.class)); + assertTrue(producerConfigs.get("value.serializer").equals(ByteArraySerializer.class)); + assertTrue(producerConfigs.get("compression.type").equals("gzip")); + List bootstrapServers = new ArrayList<>(); + bootstrapServers.add("10.98.09.199:9082"); + assertTrue((((String) producerConfigs.get("bootstrap.servers")).contains("10.98.09.199:9082"))); + Method createKafkaConsumerFactoryMethod = KafkaMessageChannelBinder.class.getDeclaredMethod("createKafkaConsumerFactory", boolean.class, String.class, ExtendedConsumerProperties.class); + createKafkaConsumerFactoryMethod.setAccessible(true); + ExtendedConsumerProperties consumerProperties = new ExtendedConsumerProperties<>(new KafkaConsumerProperties()); + DefaultKafkaConsumerFactory consumerFactory = (DefaultKafkaConsumerFactory) createKafkaConsumerFactoryMethod.invoke(this.kafkaMessageChannelBinder, true, "test", consumerProperties); + Field consumerFactoryConfigField = ReflectionUtils.findField(DefaultKafkaConsumerFactory.class, "configs", Map.class); + ReflectionUtils.makeAccessible(consumerFactoryConfigField); + Map consumerConfigs = (Map) ReflectionUtils.getField(consumerFactoryConfigField, consumerFactory); + assertTrue(consumerConfigs.get("key.deserializer").equals(ByteArrayDeserializer.class)); + assertTrue(consumerConfigs.get("value.deserializer").equals(ByteArrayDeserializer.class)); + assertTrue((((String) consumerConfigs.get("bootstrap.servers")).contains("10.98.09.199:9082"))); + } +} 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 2922ce6ef..1f05de3a1 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 @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -15,8 +15,6 @@ */ package org.springframework.cloud.stream.binder.kafka; -import static org.junit.Assert.assertNotNull; - import java.lang.reflect.Field; import org.junit.Test; @@ -29,6 +27,8 @@ import org.springframework.kafka.support.ProducerListener; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.util.ReflectionUtils; +import static org.junit.Assert.assertNotNull; + /** * @author Ilayaperumal Gopinathan */ diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListenerTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListenerTest.java index e833b6616..4cbe8dc02 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListenerTest.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListenerTest.java @@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.kafka; import javax.security.auth.login.AppConfigurationEntry; import com.sun.security.auth.login.ConfigFile; + import org.apache.kafka.common.security.JaasUtils; import org.junit.Test; 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 0c08b7c49..3aa45074e 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 @@ -1224,7 +1224,7 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests 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/bootstrap/KafkaBinderBootstrapTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderBootstrapTest.java new file mode 100644 index 000000000..ad672c892 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderBootstrapTest.java @@ -0,0 +1,47 @@ +/* + * Copyright 2017 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.bootstrap; + +import org.junit.ClassRule; +import org.junit.Test; + +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.kafka.test.rule.KafkaEmbedded; + +/** + * @author Marius Bogoevici + */ +public class KafkaBinderBootstrapTest { + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 10); + + @Test + public void testKafkaBinderConfiguration() throws Exception { + ConfigurableApplicationContext applicationContext = new SpringApplicationBuilder(SimpleApplication.class) + .web(false) + .run("--spring.cloud.stream.kafka.binder.brokers=" + embeddedKafka.getBrokersAsString(), + "--spring.cloud.stream.kafka.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); + applicationContext.close(); + } + + @SpringBootApplication + static class SimpleApplication { + } + +} diff --git a/spring-cloud-stream-binder-kafka/src/test/resources/binder-config-autoconfig.properties b/spring-cloud-stream-binder-kafka/src/test/resources/binder-config-autoconfig.properties new file mode 100644 index 000000000..8ac39c0e5 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/resources/binder-config-autoconfig.properties @@ -0,0 +1,7 @@ +spring.kafka.producer.keySerializer=org.apache.kafka.common.serialization.LongSerializer +spring.kafka.producer.valueSerializer=org.apache.kafka.common.serialization.LongSerializer +spring.kafka.consumer.keyDeserializer=org.apache.kafka.common.serialization.LongDeserializer +spring.kafka.consumer.valueDeserializer=org.apache.kafka.common.serialization.LongDeserializer +spring.kafka.producer.batchSize=10 +spring.kafka.bootstrapServers=10.98.09.199:9092,10.98.09.196:9092 +spring.kafka.producer.compressionType=snappy diff --git a/spring-cloud-stream-binder-kafka/src/test/resources/binder-config.properties b/spring-cloud-stream-binder-kafka/src/test/resources/binder-config.properties new file mode 100644 index 000000000..5a1096324 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/resources/binder-config.properties @@ -0,0 +1 @@ +spring.cloud.stream.kafka.binder.brokers=10.98.09.199:9082