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 24320173a..fc2965d37 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 @@ -25,6 +25,8 @@ import javax.validation.constraints.AssertTrue; import javax.validation.constraints.Min; import javax.validation.constraints.NotNull; +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; @@ -56,6 +58,8 @@ public class KafkaBinderConfigurationProperties { private static final String DEFAULT_KAFKA_CONNECTION_STRING = "localhost:9092"; + private final Log logger = LogFactory.getLog(getClass()); + private final Transaction transaction = new Transaction(); private final KafkaProperties kafkaProperties; @@ -529,6 +533,7 @@ public class KafkaBinderConfigurationProperties { } } consumerConfiguration.putAll(this.consumerProperties); + filterStreamManagedConfiguration(consumerConfiguration); // Override Spring Boot bootstrap server setting if left to default with the value // configured in the binder return getConfigurationWithBootstrapServer(consumerConfiguration, @@ -559,6 +564,25 @@ public class KafkaBinderConfigurationProperties { ProducerConfig.BOOTSTRAP_SERVERS_CONFIG); } + private void filterStreamManagedConfiguration(Map configuration) { + if (configuration.containsKey(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG) + && configuration.get(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG).equals(true)) { + logger.warn(constructIgnoredConfigMessage(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG) + + ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG + "=true is not supported by the Kafka binder"); + configuration.remove(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG); + } + if (configuration.containsKey(ConsumerConfig.GROUP_ID_CONFIG)) { + logger.warn(constructIgnoredConfigMessage(ConsumerConfig.GROUP_ID_CONFIG) + + "Use spring.cloud.stream.default.group or spring.cloud.stream.binding..group to specify " + + "the group instead of " + ConsumerConfig.GROUP_ID_CONFIG); + configuration.remove(ConsumerConfig.GROUP_ID_CONFIG); + } + } + + private String constructIgnoredConfigMessage(String config) { + return String.format("Ignoring provided value(s) for '%s'. ", config); + } + private Map getConfigurationWithBootstrapServer( Map configuration, String bootstrapServersConfig) { if (ObjectUtils.isEmpty(configuration.get(bootstrapServersConfig))) { diff --git a/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationPropertiesTest.java b/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationPropertiesTest.java new file mode 100644 index 000000000..9305d58fb --- /dev/null +++ b/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationPropertiesTest.java @@ -0,0 +1,108 @@ +/* + * Copyright 2018-2019 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.properties; + +import java.util.Collections; +import java.util.Map; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.junit.Test; + +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; + +import static org.assertj.core.api.Assertions.assertThat; + +public class KafkaBinderConfigurationPropertiesTest { + + @Test + public void mergedConsumerConfigurationFiltersGroupIdFromKafkaProperties() { + KafkaProperties kafkaProperties = new KafkaProperties(); + kafkaProperties.getConsumer().setGroupId("group1"); + KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties = + new KafkaBinderConfigurationProperties(kafkaProperties); + + Map mergedConsumerConfiguration = + kafkaBinderConfigurationProperties.mergedConsumerConfiguration(); + + assertThat(mergedConsumerConfiguration).doesNotContainKeys(ConsumerConfig.GROUP_ID_CONFIG); + } + + @Test + public void mergedConsumerConfigurationFiltersEnableAutoCommitFromKafkaProperties() { + KafkaProperties kafkaProperties = new KafkaProperties(); + kafkaProperties.getConsumer().setEnableAutoCommit(true); + KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties = + new KafkaBinderConfigurationProperties(kafkaProperties); + + Map mergedConsumerConfiguration = + kafkaBinderConfigurationProperties.mergedConsumerConfiguration(); + + assertThat(mergedConsumerConfiguration).doesNotContainKeys(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG); + } + + @Test + public void mergedConsumerConfigurationFiltersGroupIdFromKafkaBinderConfigurationPropertiesConfiguration() { + KafkaProperties kafkaProperties = new KafkaProperties(); + KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties = + new KafkaBinderConfigurationProperties(kafkaProperties); + kafkaBinderConfigurationProperties + .setConfiguration(Collections.singletonMap(ConsumerConfig.GROUP_ID_CONFIG, "group1")); + + Map mergedConsumerConfiguration = kafkaBinderConfigurationProperties.mergedConsumerConfiguration(); + + assertThat(mergedConsumerConfiguration).doesNotContainKeys(ConsumerConfig.GROUP_ID_CONFIG); + } + + @Test + public void mergedConsumerConfigurationFiltersEnableAutoCommitFromKafkaBinderConfigurationPropertiesConfiguration() { + KafkaProperties kafkaProperties = new KafkaProperties(); + KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties = + new KafkaBinderConfigurationProperties(kafkaProperties); + kafkaBinderConfigurationProperties + .setConfiguration(Collections.singletonMap(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")); + + Map mergedConsumerConfiguration = kafkaBinderConfigurationProperties.mergedConsumerConfiguration(); + + assertThat(mergedConsumerConfiguration).doesNotContainKeys(ConsumerConfig.GROUP_ID_CONFIG); + } + + @Test + public void mergedConsumerConfigurationFiltersGroupIdFromKafkaBinderConfigurationPropertiesConsumerProperties() { + KafkaProperties kafkaProperties = new KafkaProperties(); + KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties = + new KafkaBinderConfigurationProperties(kafkaProperties); + kafkaBinderConfigurationProperties + .setConsumerProperties(Collections.singletonMap(ConsumerConfig.GROUP_ID_CONFIG, "group1")); + + Map mergedConsumerConfiguration = kafkaBinderConfigurationProperties.mergedConsumerConfiguration(); + + assertThat(mergedConsumerConfiguration).doesNotContainKeys(ConsumerConfig.GROUP_ID_CONFIG); + } + + @Test + public void mergedConsumerConfigurationFiltersEnableAutoCommitFromKafkaBinderConfigurationPropertiesConsumerProps() { + KafkaProperties kafkaProperties = new KafkaProperties(); + KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties = + new KafkaBinderConfigurationProperties(kafkaProperties); + kafkaBinderConfigurationProperties + .setConsumerProperties(Collections.singletonMap(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")); + + Map mergedConsumerConfiguration = kafkaBinderConfigurationProperties.mergedConsumerConfiguration(); + + assertThat(mergedConsumerConfiguration).doesNotContainKeys(ConsumerConfig.GROUP_ID_CONFIG); + } +} 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 110b7fb34..84ee1c71d 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 @@ -105,7 +105,7 @@ public class KafkaBinderAutoConfigurationPropertiesTest { assertThat(consumerConfigs.get("value.deserializer")) .isEqualTo(LongDeserializer.class); assertThat(consumerConfigs.get("value.serialized")).isNull(); - assertThat(consumerConfigs.get("group.id")).isEqualTo("groupIdFromBootConfig"); + assertThat(consumerConfigs.get("group.id")).isEqualTo("test"); assertThat(consumerConfigs.get("auto.offset.reset")).isEqualTo("earliest"); assertThat((((List) consumerConfigs.get("bootstrap.servers")) .containsAll(bootstrapServers))).isTrue();