Fix generics for BinderCustomizer contract

The expectation is like this `Binder<T, C extends ConsumerProperties, P extends ProducerProperties>`
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`

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<String, Object> 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() {
 		}
+
 	}
+
 }

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.
+ * <p>
+ * 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.
+ * <p>
+ * 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<?, ConsumerProperties, ProducerProperties> binder, String binderName);
+	void customize(Binder<?, ? extends ConsumerProperties, ? extends ProducerProperties> binder, String binderName);
+
 }