GH-3686: Fix observation scope closure in the `KafkaMessageListenerCo…

…ntainer`

Fixes: #3686

According to our investigation around the `try-with-resource`,
it looks like the resource is already closed when we reach the `catch` block.

* Rework  `KafkaMessageListenerContainer.ListenerConsumer.doInvokeRecordListener()`
to `observation.openScope()` before the `try` and close it manually in the `finally` block

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -2750,7 +2750,6 @@ private void pauseForNackSleep() {
 		 * @throws Error an error.
 		 */
 		@Nullable
-		@SuppressWarnings("try")
 		private RuntimeException doInvokeRecordListener(final ConsumerRecord<K, V> cRecord, // NOSONAR
 				Iterator<ConsumerRecord<K, V>> iterator) {
 
@@ -2763,7 +2762,9 @@ private RuntimeException doInvokeRecordListener(final ConsumerRecord<K, V> cReco
 					this.observationRegistry);
 
 			observation.start();
-			try (Observation.Scope ignored = observation.openScope()) {
+			Observation.Scope observationScope = observation.openScope();
+			// We cannot use 'try-with-resource' because the resource is closed just before catch block
+			try {
 				invokeOnMessage(cRecord);
 				successTimer(sample, cRecord);
 				recordInterceptAfter(cRecord, null);
@@ -2802,6 +2803,7 @@ private RuntimeException doInvokeRecordListener(final ConsumerRecord<K, V> cReco
 				if (!(this.listener instanceof RecordMessagingMessageListenerAdapter<K, V>)) {
 					observation.stop();
 				}
+				observationScope.close();
 			}
 			return null;
 		}
@@ -4020,6 +4022,6 @@ private static class StopAfterFenceException extends KafkaException {
 
 	}
 
-	private record FailedRecordTuple<K, V>(ConsumerRecord<K, V> record, RuntimeException ex) { };
+	private record FailedRecordTuple<K, V>(ConsumerRecord<K, V> record, RuntimeException ex) { }
 
 }