From 1cfb63f4bffe74d948c7532d39f2d87ec135b200 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 29 Apr 2020 15:23:00 -0400 Subject: [PATCH] Reject improper settings for bootstrap.servers Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/875 - the `bootstrap.servers` property cannot be overridden at the binding level. --- docs/src/main/asciidoc/overview.adoc | 2 ++ .../AbstractKafkaStreamsBinderProcessor.java | 9 ++++++-- .../streams/KafkaStreamsBinderUtils.java | 9 ++++++-- .../kafka/KafkaMessageChannelBinder.java | 21 +++++++++++++----- .../stream/binder/kafka/KafkaBinderTests.java | 22 +++++++++++++++++++ 5 files changed, 54 insertions(+), 9 deletions(-) diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index 8fd02967a..aeac596aa 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -226,6 +226,7 @@ configuration:: Map with a key/value pair containing generic Kafka consumer properties. In addition to having Kafka consumer properties, other configuration properties can be passed here. For example some properties needed by the application such as `spring.cloud.stream.kafka.bindings.input.consumer.configuration.foo=bar`. +The `bootstrap.servers` property cannot be set here; use multi-binder support if you need to connect to multiple clusters. + Default: Empty map. dlqName:: @@ -354,6 +355,7 @@ For example `!ask,as*` will pass `ash` but not `ask`. Default: `*` (all headers - except the `id` and `timestamp`) configuration:: Map with a key/value pair containing generic Kafka producer properties. +The `bootstrap.servers` property cannot be set here; use multi-binder support if you need to connect to multiple clusters. + Default: Empty map. topic.properties:: diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java index 3f288e9e5..417259246 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java @@ -22,6 +22,7 @@ import java.util.regex.Pattern; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.common.utils.Bytes; @@ -63,6 +64,7 @@ import org.springframework.kafka.core.CleanupConfig; import org.springframework.kafka.streams.RecoveringDeserializationExceptionHandler; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -238,8 +240,11 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application //spring.cloud.stream.kafka.streams.binder.functions.process.configuration.num.threads (assuming that process is the function name). KafkaStreamsConsumerProperties extendedConsumerProperties = this.kafkaStreamsExtendedBindingProperties .getExtendedConsumerProperties(inboundName); - streamConfigGlobalProperties - .putAll(extendedConsumerProperties.getConfiguration()); + Map bindingConfig = extendedConsumerProperties.getConfiguration(); + Assert.state(!bindingConfig.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG), + ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; " + + "use multiple binders instead"); + streamConfigGlobalProperties.putAll(bindingConfig); String bindingLevelApplicationId = extendedConsumerProperties.getApplicationId(); // override application.id if set at the individual binding level. diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java index c24bd8352..0bd397139 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java @@ -43,6 +43,7 @@ import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.DeadLetterPublishingRecoverer; +import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -154,8 +155,12 @@ final class KafkaStreamsBinderUtils { props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, producerProperties.getExtension().getCompressionType().toString()); } - if (!ObjectUtils.isEmpty(producerProperties.getExtension().getConfiguration())) { - props.putAll(producerProperties.getExtension().getConfiguration()); + Map configs = producerProperties.getExtension().getConfiguration(); + Assert.state(!configs.containsKey(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG), + ProducerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; " + + "use multiple binders instead"); + if (!ObjectUtils.isEmpty(configs)) { + props.putAll(configs); } // Always send as byte[] on dlq (the same byte[] that the consumer received) props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); 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 58538c618..f55dea375 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 @@ -216,7 +216,7 @@ public class KafkaMessageChannelBinder extends private KafkaExtendedBindingProperties extendedBindingProperties = new KafkaExtendedBindingProperties(); - private Map ackModeInfo = new ConcurrentHashMap<>(); + private final Map ackModeInfo = new ConcurrentHashMap<>(); public KafkaMessageChannelBinder( KafkaBinderConfigurationProperties configurationProperties, @@ -514,8 +514,12 @@ public class KafkaMessageChannelBinder extends props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, producerProperties.getExtension().getCompressionType().toString()); } - if (!ObjectUtils.isEmpty(producerProperties.getExtension().getConfiguration())) { - props.putAll(producerProperties.getExtension().getConfiguration()); + Map configs = producerProperties.getExtension().getConfiguration(); + Assert.state(!configs.containsKey(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG), + ProducerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; " + + "use multiple binders instead"); + if (!ObjectUtils.isEmpty(configs)) { + props.putAll(configs); } DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( props); @@ -1087,6 +1091,9 @@ public class KafkaMessageChannelBinder extends ? dlqProducerProperties.getConfiguration() : this.configurationProperties.getTransaction() .getProducer().getConfiguration(); + Assert.state(!configs.containsKey(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG), + ProducerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; " + + "use multiple binders instead"); // Finally merge with dlq producer properties or the transaction producer properties. configuration.putAll(configs); if (record.key() != null @@ -1306,8 +1313,12 @@ public class KafkaMessageChannelBinder extends props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString()); } - if (!ObjectUtils.isEmpty(consumerProperties.getExtension().getConfiguration())) { - props.putAll(consumerProperties.getExtension().getConfiguration()); + Map config = consumerProperties.getExtension().getConfiguration(); + if (!ObjectUtils.isEmpty(config)) { + Assert.state(!config.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG), + ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; " + + "use multiple binders instead"); + props.putAll(config); } if (!ObjectUtils.isEmpty(consumerProperties.getExtension().getStartOffset())) { props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, 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 59f925426..4a77decd4 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 @@ -75,6 +75,7 @@ import org.junit.rules.ExpectedException; import org.springframework.beans.DirectFieldAccessor; import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.BinderException; import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.DefaultPollableMessageSource; @@ -146,6 +147,7 @@ import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.SettableListenableFuture; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.assertj.core.api.Assertions.fail; import static org.mockito.Mockito.mock; @@ -3682,6 +3684,26 @@ public class KafkaBinderTests extends consumer.close(); } + @Test + public void testNoBrokerOverride() throws Exception { + Binder binder = getBinder(); + ExtendedProducerProperties producerProperties = createProducerProperties(); + producerProperties.getExtension().getConfiguration().put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "foo"); + BindingProperties outputBindingProperties = createProducerBindingProperties(producerProperties); + DirectChannel moduleOutputChannel = createBindableChannel("output", outputBindingProperties); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().getConfiguration().put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "foo"); + BindingProperties consumerBindingProperties = createConsumerBindingProperties(consumerProperties); + DirectChannel moduleInputChannel = createBindableChannel("input", consumerBindingProperties); + + assertThatExceptionOfType(BinderException.class).isThrownBy(() -> binder.bindProducer("foo.bar", + moduleOutputChannel, outputBindingProperties.getProducer())) + .withCauseExactlyInstanceOf(IllegalStateException.class); + assertThatExceptionOfType(BinderException.class).isThrownBy(() -> binder.bindConsumer("foo.bar", + "testSendAndReceive", moduleInputChannel, consumerProperties)) + .withCauseExactlyInstanceOf(IllegalStateException.class); + } + private final class FailingInvocationCountingMessageHandler implements MessageHandler {