diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index 6af99f560..2d6b208d7 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -735,3 +735,17 @@ public interface KafkaBindingRebalanceListener { ==== You cannot set the `resetOffsets` consumer property to `true` when you provide a rebalance listener. + +[[consumer-producer-config-customizer]] +=== Customizing Consumer and Producer configuration + +If you want advanced customization of consumer and producer configuration that is used for creating `ConsumerFactory` and `ProducerFactory` in Kafka, +you can implement the following customizers. + +* ConsusumerConfigCustomizer +* ProducerConfigCustomizer + +Both of these interfaces provide a way to configure the config map used for consumer and producer properties. +For example, if you want to gain access to a bean that is defined at the application level, you can inject that in the implementation of the `configure` method. +When the binder discovers that these customizers are available as beans, it will invoke the `configure` method right before creating the consumer and producer factories. + 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 e209aa6f4..caeaf722c 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 @@ -64,6 +64,8 @@ import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; import org.springframework.cloud.stream.binder.HeaderMode; import org.springframework.cloud.stream.binder.MessageValues; import org.springframework.cloud.stream.binder.kafka.config.ClientFactoryCustomizer; +import org.springframework.cloud.stream.binder.kafka.config.ConsumerConfigCustomizer; +import org.springframework.cloud.stream.binder.kafka.config.ProducerConfigCustomizer; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties.StandardHeaders; @@ -221,6 +223,10 @@ public class KafkaMessageChannelBinder extends private ClientFactoryCustomizer clientFactoryCustomizer; + private ProducerConfigCustomizer producerConfigCustomizer; + + private ConsumerConfigCustomizer consumerConfigCustomizer; + public KafkaMessageChannelBinder( KafkaBinderConfigurationProperties configurationProperties, KafkaTopicProvisioner provisioningProvider) { @@ -529,6 +535,9 @@ public class KafkaMessageChannelBinder extends if (!ObjectUtils.isEmpty(kafkaProducerProperties.getConfiguration())) { props.putAll(kafkaProducerProperties.getConfiguration()); } + if (this.producerConfigCustomizer != null) { + this.producerConfigCustomizer.configure(props); + } DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( props); if (transactionIdPrefix != null) { @@ -1340,6 +1349,9 @@ public class KafkaMessageChannelBinder extends consumerProperties.getExtension().getStartOffset().name()); } + if (this.consumerConfigCustomizer != null) { + this.consumerConfigCustomizer.configure(props); + } DefaultKafkaConsumerFactory factory = new DefaultKafkaConsumerFactory<>(props); factory.setBeanName(beanName); if (this.clientFactoryCustomizer != null) { @@ -1392,6 +1404,14 @@ public class KafkaMessageChannelBinder extends return stringWriter.getBuffer().toString(); } + public void setConsumerConfigCustomizer(ConsumerConfigCustomizer consumerConfigCustomizer) { + this.consumerConfigCustomizer = consumerConfigCustomizer; + } + + public void setProducerConfigCustomizer(ProducerConfigCustomizer producerConfigCustomizer) { + this.producerConfigCustomizer = producerConfigCustomizer; + } + private final class ProducerConfigurationMessageHandler extends KafkaProducerMessageHandler { diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ConsumerConfigCustomizer.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ConsumerConfigCustomizer.java new file mode 100644 index 000000000..b1c5bd650 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ConsumerConfigCustomizer.java @@ -0,0 +1,32 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.config; + +import java.util.Map; + +/** + * This customizer is called by the binder to customize consumer configuration in + * Kafka Consumer factory. + * + * @author Soby Chacko + * @since 3.0.9 + */ +@FunctionalInterface +public interface ConsumerConfigCustomizer { + + void configure(Map consumerProperties); +} 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 18f044161..d3ab95088 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 @@ -118,7 +118,10 @@ public class KafkaBinderConfiguration { @Nullable ConsumerEndpointCustomizer> consumerCustomizer, ObjectProvider rebalanceListener, ObjectProvider dlqPartitionFunction, - ObjectProvider clientFactoryCustomizer) { + ObjectProvider clientFactoryCustomizer, + ObjectProvider consumerConfigCustomizer, + ObjectProvider producerConfigCustomizer + ) { KafkaMessageChannelBinder kafkaMessageChannelBinder = new KafkaMessageChannelBinder( configurationProperties, provisioningProvider, @@ -130,6 +133,8 @@ public class KafkaBinderConfiguration { kafkaMessageChannelBinder.setProducerMessageHandlerCustomizer(messageHandlerCustomizer); kafkaMessageChannelBinder.setConsumerEndpointCustomizer(consumerCustomizer); kafkaMessageChannelBinder.setClientFactoryCustomizer(clientFactoryCustomizer.getIfUnique()); + kafkaMessageChannelBinder.setConsumerConfigCustomizer(consumerConfigCustomizer.getIfUnique()); + kafkaMessageChannelBinder.setProducerConfigCustomizer(producerConfigCustomizer.getIfUnique()); return kafkaMessageChannelBinder; } diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ProducerConfigCustomizer.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ProducerConfigCustomizer.java new file mode 100644 index 000000000..c452332d2 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ProducerConfigCustomizer.java @@ -0,0 +1,32 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.config; + +import java.util.Map; + +/** + * This customizer is called by the binder to customize producer configuration in + * Kafka Producer factory. + * + * @author Soby Chacko + * @since 3.0.9 + */ +@FunctionalInterface +public interface ProducerConfigCustomizer { + + void configure(Map consumerProperties); +} diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java new file mode 100644 index 000000000..c69477d47 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java @@ -0,0 +1,180 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.integration; + +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.function.Function; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerInterceptor; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerInterceptor; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.TopicPartition; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.stream.binder.kafka.config.ConsumerConfigCustomizer; +import org.springframework.cloud.stream.binder.kafka.config.ProducerConfigCustomizer; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Soby Chacko + * + * Based on: https://github.com/spring-projects/spring-kafka/issues/897#issuecomment-466060097 + */ +@RunWith(SpringRunner.class) +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = {"spring.cloud.function.definition=process", + "spring.cloud.stream.bindings.process-in-0.group=KafkaConfigCustomizationTests.group"}) +@DirtiesContext +public class KafkaConfigCustomizationTests { + + private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers"; + + @ClassRule + public static EmbeddedKafkaRule kafkaEmbedded = new EmbeddedKafkaRule(1, true); + + static final CountDownLatch countDownLatch = new CountDownLatch(2); + + @BeforeClass + public static void setup() { + System.setProperty(KAFKA_BROKERS_PROPERTY, + kafkaEmbedded.getEmbeddedKafka().getBrokersAsString()); + } + + @AfterClass + public static void clean() { + System.clearProperty(KAFKA_BROKERS_PROPERTY); + } + + @Test + public void testBothConsumerAndProducerConfigsCanBeCustomized() throws InterruptedException { + Map producerProps = KafkaTestUtils + .producerProps(kafkaEmbedded.getEmbeddedKafka()); + KafkaTemplate template = new KafkaTemplate<>( + new DefaultKafkaProducerFactory<>(producerProps)); + template.send("process-in-0", "test-foo"); + template.flush(); + assertThat(countDownLatch.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @SpringBootApplication + public static class ConfigCustomizerTestConfig { + + @Bean + public Function process() { + return payload -> payload; + } + + @Bean + public ConsumerConfigCustomizer consumerConfigCustomizer() { + return consumerProperties -> { + consumerProperties.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, MyConsumerInterceptor.class.getName()); + consumerProperties.put("foo.bean", foo()); + }; + } + + @Bean + public ProducerConfigCustomizer producerConfigCustomizer() { + return producerProperties -> { + producerProperties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MyProducerInterceptor.class.getName()); + producerProperties.put("foo.bean", foo()); + }; + } + + @Bean + public Foo foo() { + return new Foo(); + } + } + + public static class Foo { + + public void foo(String what) { + KafkaConfigCustomizationTests.countDownLatch.countDown(); + } + + } + + public static class MyConsumerInterceptor implements ConsumerInterceptor { + + private Foo foo; + + @Override + public void configure(Map configs) { + this.foo = (Foo) configs.get("foo.bean"); + } + + @Override + public ConsumerRecords onConsume(ConsumerRecords records) { + this.foo.foo("consumer interceptor"); + return records; + } + + @Override + public void onCommit(Map offsets) { + + } + + @Override + public void close() { + } + + } + + public static class MyProducerInterceptor implements ProducerInterceptor { + + private Foo foo; + + @Override + public void configure(Map configs) { + this.foo = (Foo) configs.get("foo.bean"); + } + + @Override + public ProducerRecord onSend(ProducerRecord record) { + this.foo.foo("producer interceptor"); + return record; + } + + @Override + public void onAcknowledgement(RecordMetadata metadata, Exception exception) { + } + + @Override + public void close() { + } + } +}