Addressing PR review

Author: sobychacko

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

@@ -21,7 +21,6 @@
 
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.jspecify.annotations.NonNull;
 import org.jspecify.annotations.Nullable;
 
 import org.springframework.beans.BeanUtils;
@@ -48,6 +47,7 @@
  * @param <V> the value type
  *
  * @author Soby Chacko
+ * @since 4.0
  */
 public abstract class AbstractShareKafkaMessageListenerContainer<K, V>
 		implements GenericMessageListenerContainer<K, V>, BeanNameAware, ApplicationEventPublisherAware,
@@ -58,7 +58,6 @@ public abstract class AbstractShareKafkaMessageListenerContainer<K, V>
 	 */
 	public static final int DEFAULT_PHASE = Integer.MAX_VALUE - 100;
 
-	@NonNull
 	protected final ShareConsumerFactory<K, V> shareConsumerFactory;
 
 	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass()));
@@ -67,7 +66,6 @@ public abstract class AbstractShareKafkaMessageListenerContainer<K, V>
 
 	protected final ReentrantLock lifecycleLock = new ReentrantLock();
 
-	@NonNull
 	private String beanName = "noBeanNameSet";
 
 	@Nullable

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

@@ -16,6 +16,7 @@
 
 package org.springframework.kafka.listener;
 
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
@@ -49,10 +50,13 @@
  * @param <V> the value type
  *
  * @author Soby Chacko
+ * @since 4.0
  */
 public class ShareKafkaMessageListenerContainer<K, V>
 		extends AbstractShareKafkaMessageListenerContainer<K, V> {
 
+	private static final int POLL_TIMEOUT = 1000;
+
 	@Nullable
 	private String clientId;
 
@@ -166,11 +170,12 @@ private class ShareListenerConsumer implements Runnable {
 			this.consumer = ShareKafkaMessageListenerContainer.this.shareConsumerFactory.createShareConsumer(
 				ShareKafkaMessageListenerContainer.this.getGroupId(),
 				ShareKafkaMessageListenerContainer.this.getClientId());
+
 			this.genericListener = listener;
 			this.clientId = ShareKafkaMessageListenerContainer.this.getClientId();
 			// Subscribe to topics, just like in the test
 			ContainerProperties containerProperties = getContainerProperties();
-			this.consumer.subscribe(java.util.Arrays.asList(containerProperties.getTopics()));
+			this.consumer.subscribe(Arrays.asList(containerProperties.getTopics()));
 		}
 
 		@Nullable
@@ -184,7 +189,7 @@ public void run() {
 			Throwable exitThrowable = null;
 			while (isRunning()) {
 				try {
-					var records = this.consumer.poll(java.time.Duration.ofMillis(1000));
+					var records = this.consumer.poll(java.time.Duration.ofMillis(POLL_TIMEOUT));
 					if (records != null && records.count() > 0) {
 						for (var record : records) {
 							@SuppressWarnings("unchecked")
@@ -199,6 +204,7 @@ public void run() {
 				}
 				catch (Error e) {
 					this.logger.error(e, "Stopping share consumer due to an Error");
+					wrapUp();
 					throw e;
 				}
 				catch (Exception e) {
@@ -213,15 +219,19 @@ public void run() {
 			if (exitThrowable != null) {
 				this.logger.error(exitThrowable, "ShareListenerConsumer exiting due to error");
 			}
-			this.consumer.close();
-			this.logger.info(() -> this.consumerGroupId + ": Consumer stopped");
+			wrapUp();
 		}
 
 		protected void initialize() {
 			publishConsumerStartingEvent();
 			publishConsumerStartedEvent();
 		}
 
+		private void wrapUp(){
+			this.consumer.close();
+			this.logger.info(() -> this.consumerGroupId + ": Consumer stopped");
+		}
+
 		@Override
 		public String toString() {
 			return "ShareKafkaMessageListenerContainer.ShareListenerConsumer ["