Add container lifecycle metrics for start/stop counts

Introduce protected recordContainerStarted() and recordContainerStopped()
lifecycle hooks in AbstractMessageListenerContainer. Add
ContainerLifecycleMicrometerHolder to handle counter registration.
Override lifecycle methods in KafkaMessageListenerContainer following
the obtainMicrometerHolder() pattern to avoid tight Micrometer coupling
in the abstract base class.

Closes #4277

Signed-off-by: Vineeth <vineeth>

Author: Vineeth

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

@@ -530,6 +530,7 @@ public final void start() {
 						() -> "A " + GenericMessageListener.class.getName() + " implementation must be provided");
 				Assert.state(!this.fenced, "Container Fenced. It is not allowed to start.");
 				doStart();
+				recordContainerStarted();
 			}
 		}
 		finally {
@@ -609,6 +610,18 @@ public void checkGroupId() {
 
 	protected abstract void doStart();
 
+	/**
+	 * Called when the container is started. Subclasses may override to record metrics.
+	 */
+	protected void recordContainerStarted() {
+	}
+
+	/**
+	 * Called when the container is stopped. Subclasses may override to record metrics.
+	 */
+	protected void recordContainerStopped() {
+	}
+
 	@Override
 	public final void stop() {
 		stop(true);
@@ -689,6 +702,7 @@ public void stopAbnormally(Runnable callback) {
 	}
 
 	protected void doStop(Runnable callback) {
+		recordContainerStopped();
 		doStop(callback, true);
 		publishContainerStoppedEvent();
 	}

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

@@ -122,6 +122,7 @@
 import org.springframework.kafka.support.micrometer.KafkaListenerObservation;
 import org.springframework.kafka.support.micrometer.KafkaListenerObservation.DefaultKafkaListenerObservationConvention;
 import org.springframework.kafka.support.micrometer.KafkaRecordReceiverContext;
+import org.springframework.kafka.support.micrometer.ContainerLifecycleMicrometerHolder;
 import org.springframework.kafka.support.micrometer.MicrometerHolder;
 import org.springframework.kafka.support.serializer.DeserializationException;
 import org.springframework.kafka.support.serializer.SerializationUtils;
@@ -195,6 +196,9 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 
 	private @Nullable String clientIdSuffix;
 
+	@Nullable
+	private ContainerLifecycleMicrometerHolder lifecycleMicrometerHolder;
+
 	private Runnable emergencyStop = () -> stopAbnormally(() -> {
 	});
 
@@ -365,6 +369,43 @@ private void consumerWakeIfNecessary() {
 		return Collections.emptyMap();
 	}
 
+	@Override
+	protected void recordContainerStarted() {
+		if (this.lifecycleMicrometerHolder == null) {
+			this.lifecycleMicrometerHolder = obtainLifecycleMicrometerHolder();
+		}
+		if (this.lifecycleMicrometerHolder != null) {
+			this.lifecycleMicrometerHolder.recordStart();
+		}
+	}
+
+	@Override
+	protected void recordContainerStopped() {
+		if (this.lifecycleMicrometerHolder != null) {
+			this.lifecycleMicrometerHolder.recordStop();
+		}
+	}
+
+	@Nullable
+	private ContainerLifecycleMicrometerHolder obtainLifecycleMicrometerHolder() {
+		try {
+			if (KafkaUtils.MICROMETER_PRESENT) {
+				org.springframework.context.ApplicationContext ctx = getApplicationContext();
+				if (ctx != null) {
+					io.micrometer.core.instrument.MeterRegistry registry =
+							ctx.getBeanProvider(io.micrometer.core.instrument.MeterRegistry.class).getIfUnique();
+					if (registry != null) {
+						return new ContainerLifecycleMicrometerHolder(registry, getBeanName());
+					}
+				}
+			}
+		}
+		catch (Exception ex) {
+			// no micrometer
+		}
+		return null;
+	}
+
 	@Override
 	protected void doStart() {
 		if (isRunning()) {

spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/ContainerLifecycleMicrometerHolder.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2016-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.support.micrometer;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.MeterRegistry;
+
+/**
+ * Micrometer holder for container lifecycle metrics (start/stop counts).
+ *
+ * @author Vineeth Yelagandula
+ * @since 3.4
+ */
+public class ContainerLifecycleMicrometerHolder {
+
+	private final Counter startCounter;
+
+	private final Counter stopCounter;
+
+	public ContainerLifecycleMicrometerHolder(MeterRegistry registry, String containerName) {
+		this.startCounter = Counter.builder("spring.kafka.container.start.count")
+				.description("Number of times this listener container has been started")
+				.tag("name", containerName)
+				.register(registry);
+		this.stopCounter = Counter.builder("spring.kafka.container.stop.count")
+				.description("Number of times this listener container has been stopped")
+				.tag("name", containerName)
+				.register(registry);
+	}
+
+	/**
+	 * Increment the start counter.
+	 */
+	public void recordStart() {
+		this.startCounter.increment();
+	}
+
+	/**
+	 * Increment the stop counter.
+	 */
+	public void recordStop() {
+		this.stopCounter.increment();
+	}
+
+}