Batch listener does not support observation (#3047)

Fixes: #3047

* because of observation does not support batch listener, add properties `observationEnabled` and replace
  `this.containerProperties.isObservationEnabled()` for `KafkaMessageListenerContainer.ListenerConsumer`, when
  batch listener set false to `observationEnabled`.
* add notice in micrometer.adoc
* add observationEnabled and replace containerProperties.isObservationEnabled() in
  KafkaMessageListenerContainer.ListenerConsumer

**Auto-cherry-pick to `3.1.x`**

Author: Wzy19930507

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/micrometer.adoc

@@ -95,6 +95,8 @@ Using Micrometer for observation is now supported, since version 3.0, for the `K
 
 Set `observationEnabled` to `true` on the `KafkaTemplate` and `ContainerProperties` to enable observation; this will disable xref:kafka/micrometer.adoc[Micrometer Timers] because the timers will now be managed with each observation.
 
+IMPORTANT: Micrometer Observation does not support batch listener; this will enable Micrometer Timers
+
 Refer to https://micrometer.io/docs/tracing[Micrometer Tracing] for more information.
 
 To add tags to timers/traces, configure a custom `KafkaTemplateObservationConvention` or `KafkaListenerObservationConvention` to the template or listener container, respectively.

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

@@ -751,6 +751,8 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private final MicrometerHolder micrometerHolder;
 
+		private final boolean observationEnabled;
+
 		private final AtomicBoolean polling = new AtomicBoolean();
 
 		private final boolean subBatchPerPartition;
@@ -902,13 +904,15 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 				this.isBatchListener = true;
 				this.wantsFullRecords = this.batchListener.wantsPollResult();
 				this.pollThreadStateProcessor = setUpPollProcessor(true);
+				this.observationEnabled = false;
 			}
 			else if (listener instanceof MessageListener) {
 				this.listener = (MessageListener<K, V>) listener;
 				this.batchListener = null;
 				this.isBatchListener = false;
 				this.wantsFullRecords = false;
 				this.pollThreadStateProcessor = setUpPollProcessor(false);
+				this.observationEnabled = this.containerProperties.isObservationEnabled();
 			}
 			else {
 				throw new IllegalArgumentException("Listener must be one of 'MessageListener', "
@@ -990,7 +994,7 @@ private Object determineBootstrapServers(Properties consumerProperties) {
 		@Nullable
 		private KafkaAdmin obtainAdmin() {
 			KafkaAdmin customAdmin = KafkaMessageListenerContainer.this.thisOrParentContainer.getKafkaAdmin();
-			if (customAdmin == null && this.containerProperties.isObservationEnabled()) {
+			if (customAdmin == null && this.observationEnabled) {
 				ApplicationContext applicationContext = getApplicationContext();
 				if (applicationContext != null) {
 					KafkaAdmin admin = applicationContext.getBeanProvider(KafkaAdmin.class).getIfUnique();
@@ -1270,7 +1274,7 @@ private MicrometerHolder obtainMicrometerHolder() {
 			MicrometerHolder holder = null;
 			try {
 				if (KafkaUtils.MICROMETER_PRESENT && this.containerProperties.isMicrometerEnabled()
-						&& !this.containerProperties.isObservationEnabled()) {
+						&& !this.observationEnabled) {
 
 					Function<Object, Map<String, String>> mergedProvider =
 							cr -> this.containerProperties.getMicrometerTags();