GH-2806: Pulsar binder defaults properties config

- Ensure that the Pulsar binder default properties can be properly
   expressed via spring.cloud.stream.pulsar.default property prefix.
 - Add the binder child context bean with the name binderName_binderProducingContext
   into the parent application context so that individual beans from the binder context
   can be easily queried.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2806

binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/config/ExtendedBindingHandlerMappingsProviderConfiguration.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2023-2023 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.pulsar.config;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.context.properties.source.ConfigurationPropertyName;
+import org.springframework.cloud.stream.config.BindingHandlerAdvise;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+
+/**
+ * {@link EnableAutoConfiguration Auto-configuration} for extended binding metadata for the Pulsar binder.
+ *
+ * @author Soby Chacko
+ * @since 4.1.0
+ */
+@Configuration(proxyBeanMethods = false)
+public class ExtendedBindingHandlerMappingsProviderConfiguration {
+
+	@Bean
+	public BindingHandlerAdvise.MappingsProvider pulsarExtendedPropertiesDefaultMappingsProvider() {
+		return () -> {
+			Map<ConfigurationPropertyName, ConfigurationPropertyName> mappings = new HashMap<>();
+			mappings.put(
+				ConfigurationPropertyName.of("spring.cloud.stream.pulsar.bindings"),
+				ConfigurationPropertyName.of("spring.cloud.stream.pulsar.default"));
+			return mappings;
+		};
+	}
+}

binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -0,0 +1 @@
+org.springframework.cloud.stream.binder.pulsar.config.ExtendedBindingHandlerMappingsProviderConfiguration

binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarExtendedBindingDefaultPropertiesTests.java

@@ -0,0 +1,114 @@
+/*
+ * Copyright 2023-2023 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.pulsar;
+
+import java.util.function.Function;
+
+import org.apache.pulsar.common.schema.SchemaType;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.cloud.stream.binder.pulsar.config.ExtendedBindingHandlerMappingsProviderConfiguration;
+import org.springframework.cloud.stream.binder.pulsar.properties.PulsarExtendedBindingProperties;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link ExtendedBindingHandlerMappingsProviderConfiguration}.
+ *
+ * @author Soby Chacko
+ */
+class PulsarExtendedBindingDefaultPropertiesTests implements PulsarTestContainerSupport {
+
+	private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+		.withUserConfiguration(DefaultPropertiesTestApp.class)
+		.withPropertyValues(
+			"spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+			"spring.pulsar.admin.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
+			"spring.cloud.stream.pulsar.binder.partitionCount: 1",
+			"spring.cloud.stream.pulsar.default.consumer.schema-type: JSON",
+			"spring.cloud.stream.pulsar.default.consumer.receiverQueueSize: 5000",
+			"spring.cloud.stream.pulsar.default.consumer.startPaused: true",
+			"spring.cloud.stream.pulsar.default.consumer.subscription.name: my-subscription",
+			"spring.cloud.stream.pulsar.default.producer.schema-type: JSON",
+			"spring.cloud.stream.pulsar.default.producer.blockIfQueueFull: true",
+			"spring.cloud.stream.pulsar.default.producer.maxPendingMessages: 200",
+			"spring.cloud.stream.pulsar.default.producer.name: my-producer");
+
+	@Test
+	void defaultsUsedWhenNoCustomBindingProperties() {
+		this.contextRunner.run((context) -> {
+			assertThat(context)
+				.hasNotFailed()
+				.hasBean("pulsar_binderProducingContext");
+			ConfigurableApplicationContext pulsarBinderProducingContext =
+				context.getBean("pulsar_binderProducingContext", ConfigurableApplicationContext.class);
+			PulsarExtendedBindingProperties extendedBindingProperties = pulsarBinderProducingContext.getBean(PulsarExtendedBindingProperties.class);
+			assertThat(extendedBindingProperties.getExtendedConsumerProperties("process-in-0"))
+				.hasFieldOrPropertyWithValue("schemaType", SchemaType.JSON)
+				.hasFieldOrPropertyWithValue("receiverQueueSize", 5000)
+				.hasFieldOrPropertyWithValue("subscription.name", "my-subscription")
+				.hasFieldOrPropertyWithValue("startPaused", true);
+			assertThat(extendedBindingProperties.getExtendedProducerProperties("process-out-0"))
+				.hasFieldOrPropertyWithValue("schemaType", SchemaType.JSON)
+				.hasFieldOrPropertyWithValue("blockIfQueueFull", true)
+				.hasFieldOrPropertyWithValue("maxPendingMessages", 200)
+				.hasFieldOrPropertyWithValue("name", "my-producer");
+		});
+	}
+
+	@Test
+	void defaultsRespectedWhenCustomBindingProperties() {
+		this.contextRunner
+			.withPropertyValues(
+				"spring.cloud.stream.pulsar.bindings.process-in-0.consumer.receiverQueueSize: 8000",
+				"spring.cloud.stream.pulsar.bindings.process-out-0.producer.blockIfQueueFull: false",
+				"spring.cloud.stream.pulsar.bindings.process-out-0.producer.maxPendingMessages: 400")
+			.run((context) -> {
+				assertThat(context)
+					.hasNotFailed()
+					.hasBean("pulsar_binderProducingContext");
+				ConfigurableApplicationContext pulsarBinderProducingContext =
+					context.getBean("pulsar_binderProducingContext", ConfigurableApplicationContext.class);
+				PulsarExtendedBindingProperties extendedBindingProperties = pulsarBinderProducingContext.getBean(PulsarExtendedBindingProperties.class);
+				assertThat(extendedBindingProperties.getExtendedConsumerProperties("process-in-0"))
+					.hasFieldOrPropertyWithValue("schemaType", SchemaType.JSON)
+					.hasFieldOrPropertyWithValue("receiverQueueSize", 8000)
+					.hasFieldOrPropertyWithValue("subscription.name", "my-subscription")
+					.hasFieldOrPropertyWithValue("startPaused", true);
+				assertThat(extendedBindingProperties.getExtendedProducerProperties("process-out-0"))
+					.hasFieldOrPropertyWithValue("schemaType", SchemaType.JSON)
+					.hasFieldOrPropertyWithValue("blockIfQueueFull", false)
+					.hasFieldOrPropertyWithValue("maxPendingMessages", 400)
+					.hasFieldOrPropertyWithValue("name", "my-producer");
+			});
+	}
+
+
+	@EnableAutoConfiguration
+	static class DefaultPropertiesTestApp {
+
+		@Bean
+		public Function<String, String> process() {
+			return s -> s;
+		}
+	}
+}

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java

@@ -482,6 +482,10 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
 
 		if (refresh) {
 			binderProducingContext.refresh();
+			if (this.context != null) {
+				this.context.getBeanFactory().registerSingleton(configurationName + "_binderProducingContext",
+					binderProducingContext);
+			}
 			if (!useApplicationContextAsParent || "integration".equals(binderType.getDefaultName())) {
 				this.propagateSharedBeans((GenericApplicationContext) this.context, binderProducingContext);
 			}