Internal ObservationRegistry in the ErrorHandlingTaskExecutor

There is no need to configure an external `ObservationRegistry` on the `ErrorHandlingTaskExecutor`.
We only use it for closing currently in scope `Observation` after error handling on the task.

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java

@@ -44,7 +44,6 @@
 import org.springframework.integration.support.management.observation.MessageReceiverContext;
 import org.springframework.integration.support.management.observation.MessageReceiverObservationConvention;
 import org.springframework.integration.transaction.IntegrationResourceHolder;
-import org.springframework.integration.util.ErrorHandlingTaskExecutor;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessagingException;
@@ -142,12 +141,6 @@ public void setShouldTrack(boolean shouldTrack) {
 	@Override
 	public void registerObservationRegistry(ObservationRegistry observationRegistry) {
 		this.observationRegistry = observationRegistry;
-		if (isObserved()) {
-			ErrorHandlingTaskExecutor taskExecutor = (ErrorHandlingTaskExecutor) getTaskExecutor();
-			if (taskExecutor.getObservationRegistry() == null) {
-				taskExecutor.setObservationRegistry(observationRegistry);
-			}
-		}
 	}
 
 	/**
@@ -287,6 +280,7 @@ protected void messageReceived(@Nullable IntegrationResourceHolder holder, Messa
 	/**
 	 * Stop an observation (and close its scope) previously started
 	 * from the {@link #messageReceived(IntegrationResourceHolder, Message)}.
+	 * @param message the received message. Can be {@code null}; ignored in this implementation.
 	 */
 	@Override
 	protected void donePollingTask(@Nullable Message<?> message) {

spring-integration-core/src/main/java/org/springframework/integration/util/ErrorHandlingTaskExecutor.java

@@ -20,7 +20,6 @@
 
 import io.micrometer.observation.Observation;
 import io.micrometer.observation.ObservationRegistry;
-import org.jspecify.annotations.Nullable;
 
 import org.springframework.core.task.SyncTaskExecutor;
 import org.springframework.core.task.TaskExecutor;
@@ -32,8 +31,7 @@
  * instance in order to catch any exceptions. If an exception is thrown, it
  * will be handled by the provided {@link ErrorHandler}.
  * <p>
- * If an {@link ObservationRegistry} is provided, the current observation in scope
- * will be closed after error handling.
+ * After invoking {@link ErrorHandler}, the current observation scope will be closed.
  *
  * @author Jonas Partner
  * @author Mark Fisher
@@ -42,32 +40,19 @@
  */
 public class ErrorHandlingTaskExecutor implements TaskExecutor {
 
+	private static final ObservationRegistry FOR_SCOPES = ObservationRegistry.create();
+
 	private final Executor executor;
 
 	private final ErrorHandler errorHandler;
 
-	private @Nullable ObservationRegistry observationRegistry;
-
 	public ErrorHandlingTaskExecutor(Executor executor, ErrorHandler errorHandler) {
 		Assert.notNull(executor, "executor must not be null");
 		Assert.notNull(errorHandler, "errorHandler must not be null");
 		this.executor = executor;
 		this.errorHandler = errorHandler;
 	}
 
-	/**
-	 * Set an {@link ObservationRegistry} to close current observation in scope after error handling.
-	 * @param observationRegistry the {@link ObservationRegistry} to use.
-	 * @since 6.5.3
-	 */
-	public void setObservationRegistry(ObservationRegistry observationRegistry) {
-		this.observationRegistry = observationRegistry;
-	}
-
-	public @Nullable ObservationRegistry getObservationRegistry() {
-		return this.observationRegistry;
-	}
-
 	public boolean isSyncExecutor() {
 		return this.executor instanceof SyncTaskExecutor;
 	}
@@ -83,14 +68,12 @@ public void execute(final Runnable task) {
 					ErrorHandlingTaskExecutor.this.errorHandler.handleError(throwable);
 				}
 				finally {
-					if (this.observationRegistry != null) {
-						var	currentObservationScope = this.observationRegistry.getCurrentObservationScope();
-						if (currentObservationScope != null) {
-							currentObservationScope.close();
-							Observation currentObservation = currentObservationScope.getCurrentObservation();
-							currentObservation.error(throwable);
-							currentObservation.stop();
-						}
+					var currentObservationScope = FOR_SCOPES.getCurrentObservationScope();
+					if (currentObservationScope != null) {
+						currentObservationScope.close();
+						Observation currentObservation = currentObservationScope.getCurrentObservation();
+						currentObservation.error(throwable);
+						currentObservation.stop();
 					}
 				}
 			}

spring-integration-core/src/test/java/org/springframework/integration/support/management/observation/SourcePollingChannelAdapterErrorObservationTests.java

@@ -65,25 +65,26 @@ public SampleTestRunnerConsumer yourCode() {
 
 			SpansAssert.assertThat(bb.getFinishedSpans())
 					.haveSameTraceId()
-					.hasASpanWithName("dataMessageSource receive", spanAssert -> spanAssert
-							.hasTag(IntegrationObservation.GatewayTags.COMPONENT_TYPE.asString(), "message-source")
-							.hasKindEqualTo(Span.Kind.CONSUMER))
-					.hasASpanWithName("inputChannel send", spanAssert -> spanAssert
-							.hasTag(IntegrationObservation.ProducerTags.COMPONENT_NAME.asString(), "inputChannel")
-							.hasTag(IntegrationObservation.ProducerTags.COMPONENT_TYPE.asString(), "producer")
-							.hasKindEqualTo(Span.Kind.PRODUCER))
-					.hasASpanWithName("dataHandler receive", spanAssert -> spanAssert
-							.hasTag(IntegrationObservation.HandlerTags.COMPONENT_NAME.asString(), "dataHandler")
-							.hasTag(IntegrationObservation.HandlerTags.COMPONENT_TYPE.asString(), "handler")
-							.hasKindEqualTo(Span.Kind.CONSUMER))
-					.hasASpanWithName("errorChannel send", spanAssert -> spanAssert
-							.hasTag(IntegrationObservation.ProducerTags.COMPONENT_NAME.asString(), "errorChannel")
-							.hasTag(IntegrationObservation.ProducerTags.COMPONENT_TYPE.asString(), "producer")
-							.hasKindEqualTo(Span.Kind.PRODUCER))
-					.hasASpanWithName("errorChannel.bridgeTo receive", spanAssert -> spanAssert
-							.hasTag(IntegrationObservation.ProducerTags.COMPONENT_NAME.asString(), "errorChannel.bridgeTo")
-							.hasTag(IntegrationObservation.ProducerTags.COMPONENT_TYPE.asString(), "handler")
-							.hasKindEqualTo(Span.Kind.CONSUMER));
+					.hasASpanWithName("dataMessageSource receive",
+							spanAssert -> spanAssert
+									.hasTag(IntegrationObservation.GatewayTags.COMPONENT_TYPE.asString(), "message-source")
+									.hasKindEqualTo(Span.Kind.CONSUMER))
+					.hasASpanWithName("inputChannel send",
+							spanAssert -> spanAssert
+									.hasTag(IntegrationObservation.ProducerTags.COMPONENT_TYPE.asString(), "producer")
+									.hasKindEqualTo(Span.Kind.PRODUCER))
+					.hasASpanWithName("dataHandler receive",
+							spanAssert -> spanAssert
+									.hasTag(IntegrationObservation.HandlerTags.COMPONENT_TYPE.asString(), "handler")
+									.hasKindEqualTo(Span.Kind.CONSUMER))
+					.hasASpanWithName("errorChannel send",
+							spanAssert -> spanAssert
+									.hasTag(IntegrationObservation.ProducerTags.COMPONENT_TYPE.asString(), "producer")
+									.hasKindEqualTo(Span.Kind.PRODUCER))
+					.hasASpanWithName("errorChannel.bridgeTo receive",
+							spanAssert -> spanAssert
+									.hasTag(IntegrationObservation.ProducerTags.COMPONENT_TYPE.asString(), "handler")
+									.hasKindEqualTo(Span.Kind.CONSUMER));
 		};
 	}