SourcePollingChannelAdapterErrorObservationTests: race condition

The `try-with-resource` on the `ApplicationContext` may stop this context
before the message is handled as expected in the flow logic.
The timing race condition is due to a scheduling nature of the `SourcePollingChannelAdapter`

Fix the `SourcePollingChannelAdapterErrorObservationTests` with a `CountDownLatch`
to wait in the `try-with-resource` block until it is fulfilled in the `afterSendCompletion()`
of the `ChannelInterceptor` on the `errorChannel`

(cherry picked from commit d60fc95)

Author: artembilan

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

@@ -16,6 +16,9 @@
 
 package org.springframework.integration.support.management.observation;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
 import io.micrometer.observation.ObservationRegistry;
 import io.micrometer.tracing.Span;
 import io.micrometer.tracing.test.SampleTestRunner;
@@ -34,6 +37,9 @@
 import org.springframework.integration.config.EnableIntegrationManagement;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.integration.test.util.OnlyOnceTrigger;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.support.ChannelInterceptor;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
@@ -59,6 +65,8 @@ public SampleTestRunnerConsumer yourCode() {
 				applicationContext.registerBean(ObservationRegistry.class, () -> observationRegistry);
 				applicationContext.register(ObservationIntegrationTestConfiguration.class);
 				applicationContext.refresh();
+				var testConfiguration = applicationContext.getBean(ObservationIntegrationTestConfiguration.class);
+				assertThat(testConfiguration.errorHandledLatch.await(10, TimeUnit.SECONDS)).isTrue();
 			}
 
 			await().untilAsserted(() -> assertThat(bb.getFinishedSpans()).hasSize(5));
@@ -93,6 +101,8 @@ public SampleTestRunnerConsumer yourCode() {
 	@EnableIntegrationManagement(observationPatterns = "*")
 	static class ObservationIntegrationTestConfiguration {
 
+		CountDownLatch errorHandledLatch = new CountDownLatch(1);
+
 		@Bean
 		PollerMetadata pollerMetadata() {
 			PollerMetadata pollerMetadata = new PollerMetadata();
@@ -115,7 +125,17 @@ void processData(String data) {
 		@Bean
 		@BridgeTo("nullChannel")
 		DirectChannel errorChannel() {
-			return new DirectChannel();
+			DirectChannel directChannel = new DirectChannel();
+			directChannel.addInterceptor(new ChannelInterceptor() {
+
+				@Override
+				public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex) {
+					// Ensure that the error is handled before the application context is closed in the test above.
+					ObservationIntegrationTestConfiguration.this.errorHandledLatch.countDown();
+				}
+
+			});
+			return directChannel;
 		}
 
 	}