GH-3033: Register ObservationRegistry for Dynamic MessageChannels

Fixes: gh-3033

* ensure `ObservationRegistry` is registered on dynamically created `MessageChannel` instances in `StreamBridge`

binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderObservationTests.java

@@ -100,7 +100,7 @@ public class ReactorKafkaBinderObservationTests {
 		streamBridge.send("rkbot-in-topic", MessageBuilder.withPayload("data")
 			.build());
 
-		await().timeout(Duration.ofSeconds(10)).untilAsserted(() -> assertThat(SPANS.spans()).hasSize(3));
+		await().timeout(Duration.ofSeconds(10)).untilAsserted(() -> assertThat(SPANS.spans()).hasSize(4));
 		SpansAssert.assertThat(SPANS.spans().stream().map(BraveFinishedSpan::fromBrave).collect(Collectors.toList()))
 			.haveSameTraceId();
 	}

core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java

@@ -856,6 +856,21 @@ class StreamBridgeTests {
 		assertThat(new String(message.getPayload())).isEqualTo("JOHN DOE");
 	}
 
+	@Test
+	void test_3033() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+			TestChannelBinderConfiguration.getCompleteConfiguration(
+				EmptyConfiguration.class)).web(WebApplicationType.NONE).run(
+			"--spring.cloud.stream.source=outputA",
+			"--spring.jmx.enabled=false")) {
+			StreamBridge streamBridge = context.getBean(StreamBridge.class);
+			streamBridge.send("outputA", MessageBuilder.withPayload("A").build());
+
+			OutputDestination output = context.getBean(OutputDestination.class);
+			assertThat(output.receive(1000, "outputA").getHeaders().containsKey("traceparent")).isTrue();
+		}
+	}
+
 	@EnableAutoConfiguration
 	public static class DynamicProducerDestinationConfig {
 		@Bean

core/spring-cloud-stream-test-binder/pom.xml

@@ -24,6 +24,14 @@
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-test</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-actuator</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>io.micrometer</groupId>
+			<artifactId>micrometer-tracing-bridge-brave</artifactId>
+		</dependency>
 	</dependencies>
 
 </project>

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

@@ -20,6 +20,8 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
+import io.micrometer.observation.ObservationRegistry;
+
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.test.context.runner.ApplicationContextRunner;
 import org.springframework.cloud.stream.binder.Binder;
@@ -107,4 +109,8 @@ public class TestChannelBinderConfiguration<T> {
 		return new TestChannelBinderProvisioner();
 	}
 
+	@Bean
+	public ObservationRegistry observationRegistry() {
+		return ObservationRegistry.create();
+	}
 }

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java

@@ -36,6 +36,7 @@ import java.util.function.Supplier;
 import java.util.stream.StreamSupport;
 
 import io.micrometer.context.ContextSnapshotFactory;
+import io.micrometer.observation.ObservationRegistry;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.reactivestreams.Publisher;
@@ -46,6 +47,7 @@ import reactor.util.function.Tuples;
 
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
@@ -151,8 +153,10 @@ public class FunctionConfiguration {
 	@Bean
 	public StreamBridge streamBridgeUtils(FunctionCatalog functionCatalog,
 			BindingServiceProperties bindingServiceProperties, ConfigurableApplicationContext applicationContext,
-			@Nullable NewDestinationBindingCallback callback) {
-		return new StreamBridge(functionCatalog, bindingServiceProperties, applicationContext, callback);
+			@Nullable NewDestinationBindingCallback callback,
+			ObjectProvider<ObservationRegistry> observationRegistries) {
+		return new StreamBridge(functionCatalog, bindingServiceProperties, applicationContext, callback,
+				observationRegistries);
 	}
 
 	@Bean

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java

@@ -29,10 +29,12 @@ import java.util.function.Function;
 
 import io.micrometer.context.ContextExecutorService;
 import io.micrometer.context.ContextSnapshotFactory;
+import io.micrometer.observation.ObservationRegistry;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.SmartInitializingSingleton;
 import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.cloud.function.context.FunctionRegistration;
@@ -126,6 +128,7 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
 
 	private static final ReentrantLock lock = new ReentrantLock();
 
+	private ObservationRegistry observationRegistry = ObservationRegistry.NOOP;
 	/**
 	 *
 	 * @param functionCatalog instance of {@link FunctionCatalog}
@@ -133,7 +136,7 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
 	 * @param applicationContext instance of {@link ConfigurableApplicationContext}
 	 */
 	StreamBridge(FunctionCatalog functionCatalog, BindingServiceProperties bindingServiceProperties,
-		ConfigurableApplicationContext applicationContext, @Nullable NewDestinationBindingCallback destinationBindingCallback) {
+		ConfigurableApplicationContext applicationContext, @Nullable NewDestinationBindingCallback destinationBindingCallback, ObjectProvider<ObservationRegistry> observationRegistries) {
 		this.executorService = Executors.newCachedThreadPool();
 		Assert.notNull(functionCatalog, "'functionCatalog' must not be null");
 		Assert.notNull(applicationContext, "'applicationContext' must not be null");
@@ -158,6 +161,7 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
 		};
 		this.functionInvocationHelper = applicationContext.getBean(FunctionInvocationHelper.class);
 		this.streamBridgeFunctionCache = new HashMap<>();
+		observationRegistries.ifAvailable(registry -> this.observationRegistry = registry);
 	}
 
 	@Override
@@ -288,8 +292,12 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
 					messageChannel = this.isAsync() ? new ExecutorChannel(this.executorService) : new DirectWithAttributesChannel();
 					((AbstractSubscribableChannel) messageChannel).setApplicationContext(applicationContext);
 					((AbstractSubscribableChannel) messageChannel).setComponentName(destinationName);
+//<<<<<<< HEAD
 
 					BinderWrapper binderWrapper = bindingService.createBinderWrapper(binderName, destinationName, messageChannel.getClass());
+//=======
+					((AbstractSubscribableChannel) messageChannel).registerObservationRegistry(observationRegistry);
+//>>>>>>> a1418283c (GH-3033: Register ObservationRegistry for Dynamic MessageChannels)
 					if (this.destinationBindingCallback != null) {
 						Object extendedProducerProperties = this.bindingService
 								.getExtendedProducerProperties(binderWrapper.binder(), destinationName);