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 5b2a6b8b8..a2385a502 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 @@ -24,10 +24,10 @@ 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.boot.context.properties.DeprecatedConfigurationProperty; +import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -46,13 +46,25 @@ public class KafkaBinderConfigurationProperties { private final Transaction transaction = new Transaction(); - @Autowired - private KafkaProperties kafkaProperties; + private final KafkaProperties kafkaProperties; private String[] zkNodes = new String[] { "localhost" }; + /** + * Arbitrary kafka properties that apply to both producers and consumers. + */ private Map configuration = new HashMap<>(); + /** + * Arbitrary kafka consumer properties. + */ + private Map consumerProperties = new HashMap<>(); + + /** + * Arbitrary kafka producer properties. + */ + private Map producerProperties = new HashMap<>(); + private String defaultZkPort = "2181"; private String[] brokers = new String[] { "localhost" }; @@ -107,6 +119,12 @@ public class KafkaBinderConfigurationProperties { */ private String headerMapperBeanName; + + public KafkaBinderConfigurationProperties(KafkaProperties kafkaProperties) { + Assert.notNull(kafkaProperties, "'kafkaProperties' cannot be null"); + this.kafkaProperties = kafkaProperties; + } + public Transaction getTransaction() { return this.transaction; } @@ -461,18 +479,40 @@ public class KafkaBinderConfigurationProperties { this.configuration = configuration; } - public Map getConsumerConfiguration() { + public Map getConsumerProperties() { + return this.consumerProperties; + } + + public void setConsumerProperties(Map consumerProperties) { + Assert.notNull(consumerProperties, "'consumerProperties' cannot be null"); + this.consumerProperties = consumerProperties; + } + + public Map getProducerProperties() { + return this.producerProperties; + } + + public void setProducerProperties(Map producerProperties) { + Assert.notNull(producerProperties, "'producerProperties' cannot be null"); + this.producerProperties = producerProperties; + } + + /** + * Merge boot consumer properties, general properties from + * {@link #setConfiguration(Map)} that apply to consumers, properties from + * {@link #setConsumerProperties(Map)}, in that order. + * @return the merged properties. + */ + public Map mergedConsumerConfiguration() { 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 + consumerConfiguration.putAll(this.kafkaProperties.buildConsumerProperties()); + // Copy configured binder properties that apply to consumers for (Map.Entry configurationEntry : this.configuration.entrySet()) { if (ConsumerConfig.configNames().contains(configurationEntry.getKey())) { consumerConfiguration.put(configurationEntry.getKey(), configurationEntry.getValue()); } } + consumerConfiguration.putAll(this.consumerProperties); // 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))) { @@ -492,18 +532,22 @@ public class KafkaBinderConfigurationProperties { return Collections.unmodifiableMap(consumerConfiguration); } - public Map getProducerConfiguration() { + /** + * Merge boot producer properties, general properties from + * {@link #setConfiguration(Map)} that apply to producers, properties from + * {@link #setProducerProperties(Map)}, in that order. + * @return the merged properties. + */ + public Map mergedProducerConfiguration() { 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 + producerConfiguration.putAll(this.kafkaProperties.buildProducerProperties()); + // Copy configured binder properties that apply to producers for (Map.Entry configurationEntry : configuration.entrySet()) { if (ProducerConfig.configNames().contains(configurationEntry.getKey())) { producerConfiguration.put(configurationEntry.getKey(), configurationEntry.getValue()); } } + producerConfiguration.putAll(this.producerProperties); // 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))) { 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 index ba8568bab..335c9c31d 100644 --- 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 @@ -49,7 +49,7 @@ public class KafkaTopicProvisionerTests { KafkaProperties bootConfig = new KafkaProperties(); bootConfig.getProperties().put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT"); bootConfig.setBootstrapServers(Collections.singletonList("localhost:1234")); - KafkaBinderConfigurationProperties binderConfig = new KafkaBinderConfigurationProperties(); + KafkaBinderConfigurationProperties binderConfig = new KafkaBinderConfigurationProperties(bootConfig); 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()); @@ -68,7 +68,7 @@ public class KafkaTopicProvisionerTests { KafkaProperties bootConfig = new KafkaProperties(); bootConfig.getProperties().put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT"); bootConfig.setBootstrapServers(Collections.singletonList("localhost:9092")); - KafkaBinderConfigurationProperties binderConfig = new KafkaBinderConfigurationProperties(); + KafkaBinderConfigurationProperties binderConfig = new KafkaBinderConfigurationProperties(bootConfig); 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()); @@ -84,7 +84,7 @@ public class KafkaTopicProvisionerTests { @Test public void brokersInvalid() throws Exception { KafkaProperties bootConfig = new KafkaProperties(); - KafkaBinderConfigurationProperties binderConfig = new KafkaBinderConfigurationProperties(); + KafkaBinderConfigurationProperties binderConfig = new KafkaBinderConfigurationProperties(bootConfig); binderConfig.getConfiguration().put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "localhost:1234"); try { new KafkaTopicProvisioner(binderConfig, bootConfig); diff --git a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc index 8d7f4db1d..668908df9 100644 --- a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc +++ b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc @@ -63,6 +63,12 @@ Default: `9092`. spring.cloud.stream.kafka.binder.configuration:: Key/Value map of client properties (both producers and consumer) passed to all clients created by the binder. Due to the fact that these properties are used by both producers and consumers, usage should be restricted to common properties -- for example, security settings. +Properties here supersede any properties set in boot. ++ +Default: Empty map. +spring.cloud.stream.kafka.binder.consumerProperties:: +Key/Value map of arbitrary Kafka client consumer properties. +Properties here supersede any properties set in boot and in the `configuration` property above. + Default: Empty map. spring.cloud.stream.kafka.binder.headers:: @@ -86,6 +92,11 @@ The global minimum number of partitions that the binder configures on topics on It can be superseded by the `partitionCount` setting of the producer or by the value of `instanceCount * concurrency` settings of the producer (if either is larger). + Default: `1`. +spring.cloud.stream.kafka.binder.producerProperties:: +Key/Value map of arbitrary Kafka client producer properties. +Properties here supersede any properties set in boot and in the `configuration` property above. ++ +Default: Empty map. spring.cloud.stream.kafka.binder.replicationFactor:: The replication factor of auto-created topics if `autoCreateTopics` is active. Can be overridden on each binding. diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index 70859f056..95d0ee290 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -27,6 +27,7 @@ import org.apache.kafka.streams.errors.LogAndFailExceptionHandler; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; @@ -41,6 +42,7 @@ import org.springframework.util.ObjectUtils; /** * @author Marius Bogoevici * @author Soby Chacko + * @author Gary Russell */ @EnableConfigurationProperties(KafkaStreamsExtendedBindingProperties.class) @ConditionalOnBean(BindingService.class) @@ -48,8 +50,8 @@ public class KafkaStreamsBinderSupportAutoConfiguration { @Bean @ConfigurationProperties(prefix = "spring.cloud.stream.kafka.streams.binder") - public KafkaStreamsBinderConfigurationProperties binderConfigurationProperties() { - return new KafkaStreamsBinderConfigurationProperties(); + public KafkaStreamsBinderConfigurationProperties binderConfigurationProperties(KafkaProperties kafkaProperties) { + return new KafkaStreamsBinderConfigurationProperties(kafkaProperties); } @Bean("streamConfigGlobalProperties") diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsDlqDispatch.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsDlqDispatch.java index 4c7db8b75..94e2a07ba 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsDlqDispatch.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsDlqDispatch.java @@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.kafka.streams; /** * @author Soby Chacko * @author Rafal Zukowski + * @author Gary Russell */ import java.util.HashMap; import java.util.Map; @@ -105,8 +106,9 @@ class KafkaStreamsDlqDispatch { props.put(ProducerConfig.RETRIES_CONFIG, 0); props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); props.put(ProducerConfig.ACKS_CONFIG, configurationProperties.getRequiredAcks()); - if (!ObjectUtils.isEmpty(configurationProperties.getProducerConfiguration())) { - props.putAll(configurationProperties.getProducerConfiguration()); + Map mergedConfig = configurationProperties.mergedProducerConfiguration(); + if (!ObjectUtils.isEmpty(mergedConfig)) { + props.putAll(mergedConfig); } if (ObjectUtils.isEmpty(props.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) { props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, configurationProperties.getKafkaConnectionString()); diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java index 795138e3f..4c51bcbbb 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java @@ -16,13 +16,19 @@ package org.springframework.cloud.stream.binder.kafka.streams.properties; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; /** * @author Soby Chacko + * @author Gary Russell */ public class KafkaStreamsBinderConfigurationProperties extends KafkaBinderConfigurationProperties { + public KafkaStreamsBinderConfigurationProperties(KafkaProperties kafkaProperties) { + super(kafkaProperties); + } + public enum SerdeError { logAndContinue, logAndFail, 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 6c15693ed..057c1d3ba 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 @@ -134,8 +134,9 @@ public class KafkaBinderMetrics implements MeterBinder, ApplicationListener 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.getConsumerConfiguration())) { - props.putAll(binderConfigurationProperties.getConsumerConfiguration()); + Map mergedConfig = this.binderConfigurationProperties.mergedConsumerConfiguration(); + if (!ObjectUtils.isEmpty(mergedConfig)) { + props.putAll(mergedConfig); } 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 fc9f42b27..fb1e0420b 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 @@ -286,8 +286,9 @@ 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.getProducerConfiguration())) { - props.putAll(configurationProperties.getProducerConfiguration()); + Map mergedConfig = this.configurationProperties.mergedProducerConfiguration(); + if (!ObjectUtils.isEmpty(mergedConfig)) { + props.putAll(mergedConfig); } if (ObjectUtils.isEmpty(props.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) { props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString()); @@ -723,8 +724,9 @@ 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.getConsumerConfiguration())) { - props.putAll(configurationProperties.getConsumerConfiguration()); + Map mergedConfig = configurationProperties.mergedConsumerConfiguration(); + if (!ObjectUtils.isEmpty(mergedConfig)) { + props.putAll(mergedConfig); } 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 b24cc761a..625bfddea 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 @@ -70,8 +70,8 @@ public class KafkaBinderConfiguration { private KafkaProperties kafkaProperties; @Bean - KafkaBinderConfigurationProperties configurationProperties() { - return new KafkaBinderConfigurationProperties(); + KafkaBinderConfigurationProperties configurationProperties(KafkaProperties kafkaProperties) { + return new KafkaBinderConfigurationProperties(kafkaProperties); } @Bean diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderHealthIndicatorConfiguration.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderHealthIndicatorConfiguration.java index 646c47d5f..ef1cb9f55 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderHealthIndicatorConfiguration.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderHealthIndicatorConfiguration.java @@ -33,7 +33,7 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.util.ObjectUtils; /** - * + * * @author Oleg Zhurakousky * */ @@ -48,8 +48,9 @@ class KafkaBinderHealthIndicatorConfiguration { 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.getConsumerConfiguration())) { - props.putAll(configurationProperties.getConsumerConfiguration()); + Map mergedConfig = configurationProperties.mergedConsumerConfiguration(); + if (!ObjectUtils.isEmpty(mergedConfig)) { + props.putAll(mergedConfig); } if (!props.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) { props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, configurationProperties.getKafkaConnectionString()); 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 56c5c51c7..8bc3f1ba9 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 @@ -63,7 +63,6 @@ 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; @@ -182,7 +181,7 @@ public class KafkaBinderTests extends protected KafkaTestBinder getBinder() { if (binder == null) { KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); - KafkaTopicProvisioner kafkaTopicProvisioner = new KafkaTopicProvisioner(binderConfiguration, new KafkaProperties()); + KafkaTopicProvisioner kafkaTopicProvisioner = new KafkaTopicProvisioner(binderConfiguration, new TestKafkaProperties()); try { kafkaTopicProvisioner.afterPropertiesSet(); } @@ -196,7 +195,7 @@ public class KafkaBinderTests extends private Binder getBinder(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) { KafkaTopicProvisioner provisioningProvider = - new KafkaTopicProvisioner(kafkaBinderConfigurationProperties, new KafkaProperties()); + new KafkaTopicProvisioner(kafkaBinderConfigurationProperties, new TestKafkaProperties()); try { provisioningProvider.afterPropertiesSet(); } @@ -207,7 +206,8 @@ public class KafkaBinderTests extends } private KafkaBinderConfigurationProperties createConfigurationProperties() { - KafkaBinderConfigurationProperties binderConfiguration = new KafkaBinderConfigurationProperties(); + KafkaBinderConfigurationProperties binderConfiguration = new KafkaBinderConfigurationProperties( + new TestKafkaProperties()); BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses(); List bAddresses = new ArrayList<>(); for (BrokerAddress bAddress : brokerAddresses) { @@ -215,7 +215,6 @@ public class KafkaBinderTests extends } String[] foo = new String[bAddresses.size()]; binderConfiguration.setBrokers(bAddresses.toArray(foo)); - binderConfiguration.setZkNodes(embeddedKafka.getZookeeperConnectionString()); return binderConfiguration; } 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 dfc5a6eb7..48f3e6b4f 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 @@ -30,6 +30,7 @@ import java.util.stream.Collectors; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; import org.junit.Test; @@ -67,8 +68,10 @@ public class KafkaBinderUnitTests { @Test public void testPropertyOverrides() throws Exception { - KafkaBinderConfigurationProperties binderConfigurationProperties = new KafkaBinderConfigurationProperties(); - KafkaTopicProvisioner provisioningProvider = new KafkaTopicProvisioner(binderConfigurationProperties, new KafkaProperties()); + KafkaProperties kafkaProperties = new TestKafkaProperties(); + KafkaBinderConfigurationProperties binderConfigurationProperties = + new KafkaBinderConfigurationProperties(kafkaProperties); + KafkaTopicProvisioner provisioningProvider = new KafkaTopicProvisioner(binderConfigurationProperties, kafkaProperties); KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(binderConfigurationProperties, provisioningProvider); KafkaConsumerProperties consumerProps = new KafkaConsumerProperties(); @@ -102,6 +105,30 @@ public class KafkaBinderUnitTests { assertThat(configs.get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)).isEqualTo("earliest"); } + @Test + public void testMergedConsumerProperties() { + KafkaProperties bootProps = new TestKafkaProperties(); + bootProps.getConsumer().getProperties().put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "bar"); + KafkaBinderConfigurationProperties props = new KafkaBinderConfigurationProperties(bootProps); + assertThat(props.mergedConsumerConfiguration().get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)).isEqualTo("bar"); + props.getConfiguration().put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "baz"); + assertThat(props.mergedConsumerConfiguration().get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)).isEqualTo("baz"); + props.getConsumerProperties().put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "qux"); + assertThat(props.mergedConsumerConfiguration().get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)).isEqualTo("qux"); + } + + @Test + public void testMergedProducerProperties() { + KafkaProperties bootProps = new TestKafkaProperties(); + bootProps.getProducer().getProperties().put(ProducerConfig.RETRIES_CONFIG, "bar"); + KafkaBinderConfigurationProperties props = new KafkaBinderConfigurationProperties(bootProps); + assertThat(props.mergedProducerConfiguration().get(ProducerConfig.RETRIES_CONFIG)).isEqualTo("bar"); + props.getConfiguration().put(ProducerConfig.RETRIES_CONFIG, "baz"); + assertThat(props.mergedProducerConfiguration().get(ProducerConfig.RETRIES_CONFIG)).isEqualTo("baz"); + props.getProducerProperties().put(ProducerConfig.RETRIES_CONFIG, "qux"); + assertThat(props.mergedProducerConfiguration().get(ProducerConfig.RETRIES_CONFIG)).isEqualTo("qux"); + } + @Test public void testOffsetResetWithGroupManagementEarliest() throws Exception { testOffsetResetWithGroupManagement(true, true); @@ -126,7 +153,8 @@ public class KafkaBinderUnitTests { final List partitions = new ArrayList<>(); partitions.add(new TopicPartition("foo", 0)); partitions.add(new TopicPartition("foo", 1)); - KafkaBinderConfigurationProperties configurationProperties = new KafkaBinderConfigurationProperties(); + KafkaBinderConfigurationProperties configurationProperties = new KafkaBinderConfigurationProperties( + new TestKafkaProperties()); KafkaTopicProvisioner provisioningProvider = mock(KafkaTopicProvisioner.class); ConsumerDestination dest = mock(ConsumerDestination.class); given(dest.getName()).willReturn("foo"); diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTransactionTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTransactionTests.java index 8b81ee4e9..762b3110b 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTransactionTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTransactionTests.java @@ -58,9 +58,10 @@ public class KafkaTransactionTests { @SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testProducerRunsInTx() { - KafkaProperties kafkaProperties = new KafkaProperties(); + KafkaProperties kafkaProperties = new TestKafkaProperties(); kafkaProperties.setBootstrapServers(Collections.singletonList(embeddedKafka.getBrokersAsString())); - KafkaBinderConfigurationProperties configurationProperties = new KafkaBinderConfigurationProperties(); + KafkaBinderConfigurationProperties configurationProperties = + new KafkaBinderConfigurationProperties(kafkaProperties); configurationProperties.getTransaction().setTransactionIdPrefix("foo-"); KafkaTopicProvisioner provisioningProvider = new KafkaTopicProvisioner(configurationProperties, kafkaProperties); provisioningProvider.setMetadataRetryOperations(new RetryTemplate()); diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/TestKafkaProperties.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/TestKafkaProperties.java new file mode 100644 index 000000000..cf52d42e8 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/TestKafkaProperties.java @@ -0,0 +1,41 @@ +/* + * 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; + +import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.kafka.common.serialization.ByteArraySerializer; + +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; + +/** + * Test {@link KafkaProperties} initialized in the same way as the + * {@code KafkaBinderEnvironmentPostProcessor} initializes the properties. + * + * @author Gary Russell + * @since 2.1 + * + */ +public class TestKafkaProperties extends KafkaProperties { + + public TestKafkaProperties() { + getConsumer().setKeyDeserializer(ByteArrayDeserializer.class); + getConsumer().setValueDeserializer(ByteArrayDeserializer.class); + getProducer().setKeySerializer(ByteArraySerializer.class); + getProducer().setValueSerializer(ByteArraySerializer.class); + } + +}