From b09892904864f1fe120a7d33ab3deb3a4d581063 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 18 Jan 2024 15:46:17 -0500 Subject: [PATCH] Fix generics for `BinderCustomizer` contract The expectation is like this `Binder` so, the `BinderCustomizer` must be in extension bounds as well. * Move Javadocs from the `BinderCustomizer.customize()` to the class level * Add `BinderCustomizer` verification to the `KafkaConfigCustomizationTests` --- .../KafkaConfigCustomizationTests.java | 46 +++++++++++++++++-- .../cloud/stream/binder/BinderCustomizer.java | 29 +++++++----- 2 files changed, 58 insertions(+), 17 deletions(-) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java index 8921d7495..1af790b5b 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2023 the original author or authors. + * Copyright 2020-2024 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. @@ -35,11 +35,16 @@ import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.stream.binder.BinderCustomizer; +import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder; +import org.springframework.cloud.stream.binder.kafka.config.ClientFactoryCustomizer; import org.springframework.cloud.stream.binder.kafka.support.ConsumerConfigCustomizer; import org.springframework.cloud.stream.binder.kafka.support.ProducerConfigCustomizer; import org.springframework.context.annotation.Bean; +import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.test.EmbeddedKafkaBroker; import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; @@ -49,22 +54,24 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Soby Chacko + * @author Artem Bilan * * Based on: https://github.com/spring-projects/spring-kafka/issues/897#issuecomment-466060097 */ -@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = {"spring.cloud.function.definition=process", +@SpringBootTest(properties = {"spring.cloud.function.definition=process", "spring.cloud.stream.bindings.process-in-0.group=KafkaConfigCustomizationTests.group"}) @DirtiesContext -@EmbeddedKafka(bootstrapServersProperty = "spring.kafka.bootstrap-servers") +@EmbeddedKafka class KafkaConfigCustomizationTests { - private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers"; - static final CountDownLatch countDownLatch = new CountDownLatch(2); @Autowired EmbeddedKafkaBroker embeddedKafkaBroker; + @Autowired + ConfigCustomizerTestConfig configCustomizerTestConfig; + @Test void bothConsumerAndProducerConfigsCanBeCustomized() throws InterruptedException { Map producerProps = KafkaTestUtils @@ -74,6 +81,9 @@ class KafkaConfigCustomizationTests { template.send("process-in-0", "test-foo"); template.flush(); assertThat(countDownLatch.await(10, TimeUnit.SECONDS)).isTrue(); + + assertThat(this.configCustomizerTestConfig.producerFactoryCustomized).isTrue(); + assertThat(this.configCustomizerTestConfig.consumerFactoryCustomized).isTrue(); } @SpringBootApplication @@ -108,6 +118,30 @@ class KafkaConfigCustomizationTests { public Foo foo() { return new Foo(); } + + private boolean producerFactoryCustomized; + + private boolean consumerFactoryCustomized; + + @Bean + BinderCustomizer binderCustomizer() { + return (binder,binderName) -> { + if (binder instanceof KafkaMessageChannelBinder kafkaMessageChannelBinder) { + kafkaMessageChannelBinder.addClientFactoryCustomizer(new ClientFactoryCustomizer() { + @Override + public void configure(ConsumerFactory cf) { + consumerFactoryCustomized = true; + } + + @Override + public void configure(ProducerFactory pf) { + producerFactoryCustomized = true; + } + }); + } + }; + } + } public static class Foo { @@ -166,5 +200,7 @@ class KafkaConfigCustomizationTests { @Override public void close() { } + } + } diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderCustomizer.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderCustomizer.java index 5e15de867..9ed688364 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderCustomizer.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderCustomizer.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2020 the original author or authors. + * Copyright 2015-2024 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. @@ -17,24 +17,29 @@ package org.springframework.cloud.stream.binder; /** + * The binder customization strategy. + *

+ * When customization beans are present in an application that uses a single binder, + * those beans are detected by the binder. However, this is not the case in a multi-binder + * scenario, since various binders live in different application contexts. This + * customizer enables the application to properly apply customizations in all the + * binders. By providing an implementation of this interface, the binders, although + * reside in different application contexts, will receive the customization. + * Spring Cloud Stream ensures that the customizations take place before the binders are accessed. + *

+ * In the case of a single binder, the use of this customizer is redundant. + * * @author Soby Chacko + * @author Artme Bilan * @since 3.1.0 */ public interface BinderCustomizer { /** - * When customization beans are present in an application that uses a single binder, - * those beans are detected by the binder. However, this is not the case in a multi-binder - * scenario, since various binders live in different application contexts. This customizer - * enables the application to properly apply customizations in all the binders. - * By providing an implementation of this interface, the binders, although reside - * in different application contexts, will receive the customization. - * Spring Cloud Stream ensures that the customizations take place before the binders are - * accessed. The user must check for the binder type and then apply the necessary customizations. - * In the case of a single binder, the use of this customizer is redundant. - * + * The user must check for the binder type and then apply the necessary customizations. * @param binder to be customized * @param binderName binder name to distinguish between multiple instances of the same binder type */ - void customize(Binder binder, String binderName); + void customize(Binder binder, String binderName); + }