GH-3151: Include error handler into a listener observation

Fixes: #3151

After fixing #3049 we are missing an `ErrorHandler` part within an observation.
This even cause a retryable topic logic ot be out of an observation scope.

* Restore the previous behavior and add `observation.error(e)` when it is not re-thrown in case of `this.commonErrorHandler` presence

(cherry picked from commit c24575c)

Author: artembilan

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

@@ -2794,37 +2794,38 @@ private RuntimeException doInvokeRecordListener(final ConsumerRecord<K, V> cReco
 					DefaultKafkaListenerObservationConvention.INSTANCE,
 					() -> new KafkaRecordReceiverContext(cRecord, getListenerId(), this::clusterId),
 					this.observationRegistry);
-			try {
-				observation.observe(() -> {
+			return observation.observe(() -> {
+				try {
 					invokeOnMessage(cRecord);
 					successTimer(sample, cRecord);
 					recordInterceptAfter(cRecord, null);
-				});
-			}
-			catch (RuntimeException e) {
-				failureTimer(sample, cRecord);
-				recordInterceptAfter(cRecord, e);
-				if (this.commonErrorHandler == null) {
-					throw e;
-				}
-				try {
-					invokeErrorHandler(cRecord, iterator, e);
-					commitOffsetsIfNeededAfterHandlingError(cRecord);
-				}
-				catch (KafkaException ke) {
-					ke.selfLog(ERROR_HANDLER_THREW_AN_EXCEPTION, this.logger);
-					return ke;
-				}
-				catch (RuntimeException ee) {
-					this.logger.error(ee, ERROR_HANDLER_THREW_AN_EXCEPTION);
-					return ee;
 				}
-				catch (Error er) { // NOSONAR
-					this.logger.error(er, "Error handler threw an error");
-					throw er;
+				catch (RuntimeException e) {
+					failureTimer(sample, cRecord);
+					recordInterceptAfter(cRecord, e);
+					if (this.commonErrorHandler == null) {
+						throw e;
+					}
+					observation.error(e);
+					try {
+						invokeErrorHandler(cRecord, iterator, e);
+						commitOffsetsIfNeededAfterHandlingError(cRecord);
+					}
+					catch (KafkaException ke) {
+						ke.selfLog(ERROR_HANDLER_THREW_AN_EXCEPTION, this.logger);
+						return ke;
+					}
+					catch (RuntimeException ee) {
+						this.logger.error(ee, ERROR_HANDLER_THREW_AN_EXCEPTION);
+						return ee;
+					}
+					catch (Error er) { // NOSONAR
+						this.logger.error(er, "Error handler threw an error");
+						throw er;
+					}
 				}
-			}
-			return null;
+				return null;
+			});
 		}
 
 		private void commitOffsetsIfNeededAfterHandlingError(final ConsumerRecord<K, V> cRecord) {