Record unhanlded JMS exceptions as observation errors

Prior to this commit, when using the `DefaultMessageListenerContainer`
as a backing listener container for JMS messages, the instrumentation
would not record JMS exceptions as observation errors when they were not
handled by configured `ErrorHandler`.

This commit ensures that this is the case.

Fixes gh-32458

Author: bclozel

spring-jms/src/main/java/org/springframework/jms/listener/AbstractPollingMessageListenerContainer.java

@@ -339,7 +339,12 @@ protected boolean doReceiveAndExecute(Object invoker, @Nullable Session session,
 						}
 						status.setRollbackOnly();
 					}
-					handleListenerException(ex);
+					try {
+						handleListenerException(ex);
+					}
+					catch (Throwable throwable) {
+						observation.error(throwable);
+					}
 					// Rethrow JMSException to indicate an infrastructure problem
 					// that may have to trigger recovery...
 					if (ex instanceof JMSException jmsException) {

spring-jms/src/test/java/org/springframework/jms/listener/MessageListenerContainerObservationTests.java

@@ -76,6 +76,7 @@ void shouldRecordJmsProcessObservations(AbstractMessageListenerContainer listene
 		assertThat(registry).hasObservationWithNameEqualTo("jms.message.process")
 				.that()
 				.hasHighCardinalityKeyValue("messaging.destination.name", "spring.test.observation");
+		assertThat(registry).hasNumberOfObservationsEqualTo(1);
 		listenerContainer.stop();
 		listenerContainer.shutdown();
 	}
@@ -103,7 +104,37 @@ void shouldHaveObservationScopeInErrorHandler(AbstractMessageListenerContainer l
 		Assertions.assertThat(observationInErrorHandler.get()).isNotNull();
 		assertThat(registry).hasObservationWithNameEqualTo("jms.message.process")
 				.that()
-				.hasHighCardinalityKeyValue("messaging.destination.name", "spring.test.observation");
+				.hasHighCardinalityKeyValue("messaging.destination.name", "spring.test.observation")
+				.hasLowCardinalityKeyValue("exception", "none");
+		assertThat(registry).hasNumberOfObservationsEqualTo(1);
+		listenerContainer.stop();
+		listenerContainer.shutdown();
+	}
+
+	@ParameterizedTest(name = "[{index}] {0}")
+	@MethodSource("listenerContainers")
+	void shouldHaveObservationErrorWhenRethrown(AbstractMessageListenerContainer listenerContainer) throws Exception {
+		JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
+		jmsTemplate.convertAndSend("spring.test.observation", "message content");
+		CountDownLatch latch = new CountDownLatch(1);
+		listenerContainer.setConnectionFactory(connectionFactory);
+		listenerContainer.setObservationRegistry(registry);
+		listenerContainer.setDestinationName("spring.test.observation");
+		listenerContainer.setMessageListener((MessageListener) message -> {
+			throw new IllegalStateException("error");
+		});
+		listenerContainer.setErrorHandler(error -> {
+			latch.countDown();
+			throw new IllegalStateException("not handled");
+		});
+		listenerContainer.afterPropertiesSet();
+		listenerContainer.start();
+		latch.await(2, TimeUnit.SECONDS);
+		assertThat(registry).hasObservationWithNameEqualTo("jms.message.process")
+				.that()
+				.hasHighCardinalityKeyValue("messaging.destination.name", "spring.test.observation")
+				.hasLowCardinalityKeyValue("exception", "IllegalStateException");
+		assertThat(registry).hasNumberOfObservationsEqualTo(1);
 		listenerContainer.stop();
 		listenerContainer.shutdown();
 	}