Add start/stop count metrics to AbstractMessageListenerContainer

Adds two Micrometer counters to AbstractMessageListenerContainer to
track listener container lifecycle events:

- spring.kafka.container.start.count: incremented each time a
  container is successfully started via doStart()
- spring.kafka.container.stop.count: incremented each time a
  container is stopped, covering both wait=true and wait=false paths

Closes #4277

Signed-off-by: Vineeth Yelagandula <111960524+Yelagandula@users.noreply.github.com>

Author: Yelagandula

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

@@ -32,6 +32,9 @@
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Metrics;
+
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AdminClientConfig;
@@ -145,6 +148,14 @@ public abstract class AbstractMessageListenerContainer<K, V>
 	@Nullable
 	private KafkaAdmin kafkaAdmin;
 
+	private final Counter startCounter = Counter.builder("spring.kafka.container.start.count")
+			.description("The number of times this listener container has been started")
+			.register(Metrics.globalRegistry);
+
+	private final Counter stopCounter = Counter.builder("spring.kafka.container.stop.count")
+			.description("The number of times this listener container has been stopped")
+			.register(Metrics.globalRegistry);
+
 	/**
 	 * Construct an instance with the provided factory and properties.
 	 * @param consumerFactory the factory.
@@ -530,6 +541,8 @@ 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();
+				this.startCounter.increment();
+
 			}
 		}
 		finally {
@@ -633,6 +646,7 @@ public final void stop(boolean wait) {
 				try {
 					latch.await(this.containerProperties.getShutdownTimeout(), TimeUnit.MILLISECONDS); // NOSONAR
 					publishContainerStoppedEvent();
+					this.stopCounter.increment();
 				}
 				catch (@SuppressWarnings("unused") InterruptedException e) {
 					Thread.currentThread().interrupt();
@@ -642,6 +656,7 @@ public final void stop(boolean wait) {
 				this.lifecycleLock.lock();
 				try {
 					doStop(this::publishContainerStoppedEvent);
+					this.stopCounter.increment();
 				}
 				finally {
 					this.lifecycleLock.unlock();