GH-10379: Optimize Observability API usage (#10380)

Fixes: #10379

The `ObservationRegistry.NOOP` still brings some performance overhead.

* Fix `MessageProducerSupport` & `SourcePollingChannelAdapter` to check for `isObserved()`
before using the `IntegrationObservation`

(cherry picked from commit 36559ca)

Author: artembilan

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

@@ -253,12 +253,17 @@ protected void sendMessage(Message<?> message) {
 		}
 
 		try {
-			IntegrationObservation.HANDLER.observation(
-							this.observationConvention,
-							DefaultMessageReceiverObservationConvention.INSTANCE,
-							() -> new MessageReceiverContext(message, getComponentName(), "message-producer"),
-							this.observationRegistry)
-					.observe(() -> this.messagingTemplate.send(getRequiredOutputChannel(), trackMessageIfAny(message)));
+			if (isObserved()) {
+				IntegrationObservation.HANDLER.observation(
+								this.observationConvention,
+								DefaultMessageReceiverObservationConvention.INSTANCE,
+								() -> new MessageReceiverContext(message, getComponentName(), "message-producer"),
+								this.observationRegistry)
+						.observe(() -> sendMessageWithTracking(message));
+			}
+			else {
+				sendMessageWithTracking(message);
+			}
 		}
 		catch (RuntimeException ex) {
 			if (!sendErrorMessageIfNecessary(message, ex)) {
@@ -267,6 +272,10 @@ protected void sendMessage(Message<?> message) {
 		}
 	}
 
+	private void sendMessageWithTracking(Message<?> message) {
+		this.messagingTemplate.send(getRequiredOutputChannel(), trackMessageIfAny(message));
+	}
+
 	protected void subscribeToPublisher(Publisher<? extends Message<?>> publisher) {
 		MessageChannel channelForSubscription = getRequiredOutputChannel();
 

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

@@ -257,13 +257,15 @@ protected Message<?> receiveMessage() {
 	 */
 	@Override
 	protected void messageReceived(@Nullable IntegrationResourceHolder holder, Message<?> message) {
-		Observation observation =
-				IntegrationObservation.HANDLER.observation(this.observationConvention,
-						DefaultMessageReceiverObservationConvention.INSTANCE,
-						() -> new MessageReceiverContext(message, getComponentName(), "message-source"),
-						this.observationRegistry);
-
-		observation.start().openScope();
+		if (isObserved()) {
+			Observation observation =
+					IntegrationObservation.HANDLER.observation(this.observationConvention,
+							DefaultMessageReceiverObservationConvention.INSTANCE,
+							() -> new MessageReceiverContext(message, getComponentName(), "message-source"),
+							this.observationRegistry);
+
+			observation.start().openScope();
+		}
 		super.messageReceived(holder, message);
 	}