Improve logging and test reliability for share consumer implementation

  - Replace Thread.sleep() with CountDownLatch assertions in tests for
    reliable synchronization and faster test execution
  - Use LogMessage.format() instead of Supplier with String.format()
    for consistent logging patterns
  - Fix lambda expression to use effectively final variable in error logging
  - Change shareConsumer field from nullable Boolean to primitive boolean
    in AbstractKafkaListenerEndpoint for cleaner null handling
  - Make utility methods static where appropriate for better code organization

Signed-off-by: Soby Chacko <[email protected]>

Author: sobychacko

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java

@@ -98,7 +98,7 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 
 	private @Nullable Boolean batchListener;
 
-	private @Nullable Boolean shareConsumer;
+	private boolean shareConsumer;
 
 	private @Nullable KafkaTemplate<?, ?> replyTemplate;
 
@@ -303,7 +303,7 @@ public void setShareConsumer(boolean shareConsumer) {
 	 * @since 4.0
 	 */
 	public boolean isShareConsumer() {
-		return this.shareConsumer != null && this.shareConsumer;
+		return this.shareConsumer;
 	}
 
 	/**

spring-kafka/src/main/java/org/springframework/kafka/config/ShareKafkaListenerContainerFactory.java

@@ -50,6 +50,7 @@
  * @param <V> the value type
  *
  * @author Soby Chacko
+ *
  * @since 4.0
  */
 public class ShareKafkaListenerContainerFactory<K, V>
@@ -179,7 +180,7 @@ else if ("implicit".equals(mode)) {
 		return containerProperties.isExplicitShareAcknowledgment();
 	}
 
-	private void validateShareConfiguration(KafkaListenerEndpoint endpoint) {
+	private static void validateShareConfiguration(KafkaListenerEndpoint endpoint) {
 		// Validate that batch listeners aren't used with share consumers
 		if (Boolean.TRUE.equals(endpoint.getBatchListener())) {
 			throw new IllegalArgumentException(

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

@@ -35,6 +35,7 @@
 
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.core.log.LogAccessor;
+import org.springframework.core.log.LogMessage;
 import org.springframework.core.task.AsyncTaskExecutor;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.kafka.core.ShareConsumerFactory;
@@ -71,7 +72,9 @@
  * @param <V> the value type
  *
  * @author Soby Chacko
+ *
  * @since 4.0
+ *
  * @see ShareConsumer
  * @see ShareAcknowledgment
  */
@@ -247,7 +250,7 @@ private class ShareListenerConsumer implements Runnable {
 			if (this.isExplicitMode) {
 				// Apply explicit mode configuration to consumer
 				// Note: This should ideally be done during consumer creation in the factory
-				this.logger.info(() -> "Share consumer configured for explicit acknowledgment mode");
+				this.logger.info("Share consumer configured for explicit acknowledgment mode");
 			}
 
 			this.consumer.subscribe(Arrays.asList(containerProperties.getTopics()));
@@ -409,17 +412,18 @@ void onRecordAcknowledged(ConsumerRecord<K, V> record) {
 		private void processQueuedAcknowledgments() {
 			PendingAcknowledgment<K, V> pendingAck;
 			while ((pendingAck = this.acknowledgmentQueue.poll()) != null) {
+				final PendingAcknowledgment<K, V> ack = pendingAck;
 				try {
-					this.consumer.acknowledge(pendingAck.record, pendingAck.type);
+					this.consumer.acknowledge(ack.record, ack.type);
 					// Find and notify the acknowledgment object
-					ShareConsumerAcknowledgment ack = this.pendingAcknowledgments.get(pendingAck.record);
-					if (ack != null) {
-						ack.notifyAcknowledged(pendingAck.type);
-						onRecordAcknowledged(pendingAck.record);
+					ShareConsumerAcknowledgment acknowledgment = this.pendingAcknowledgments.get(ack.record);
+					if (acknowledgment != null) {
+						acknowledgment.notifyAcknowledged(ack.type);
+						onRecordAcknowledged(ack.record);
 					}
 				}
 				catch (Exception e) {
-					this.logger.error(e, "Failed to process queued acknowledgment for record: " + pendingAck.record);
+					this.logger.error(e, () -> "Failed to process queued acknowledgment for record: " + ack.record);
 				}
 			}
 		}
@@ -437,7 +441,7 @@ private void checkAcknowledgmentTimeouts() {
 				long recordAge = currentTime - entry.getValue();
 				if (recordAge > this.ackTimeoutMs) {
 					ConsumerRecord<K, V> record = entry.getKey();
-					this.logger.warn(() -> String.format(
+					this.logger.warn(LogMessage.format(
 						"Record not acknowledged within timeout (%d seconds). " +
 						"In explicit acknowledgment mode, you must call ack.acknowledge(), ack.release(), " +
 						"or ack.reject() for every record. " +

spring-kafka/src/test/java/org/springframework/kafka/listener/ShareKafkaMessageListenerContainerConstraintTests.java

@@ -122,7 +122,8 @@ public void onShareRecord(ConsumerRecord<String, String> record,
 			produceTestRecords(bootstrapServers, topic, 2);
 
 			// Wait and verify second batch is NOT processed yet
-			Thread.sleep(3000);
+			// Using a latch that should NOT count down to verify blocking behavior
+			assertThat(secondBatchLatch.await(3, TimeUnit.SECONDS)).isFalse();
 			assertThat(totalProcessed.get()).isEqualTo(3);
 			assertThat(secondBatchLatch.getCount()).isEqualTo(2);
 
@@ -198,7 +199,8 @@ public void onShareRecord(ConsumerRecord<String, String> record,
 			produceTestRecords(bootstrapServers, topic, 1);
 
 			// Should not process new records while one is still pending
-			Thread.sleep(3000);
+			// Using a latch that should NOT count down to verify blocking behavior
+			assertThat(nextPollLatch.await(3, TimeUnit.SECONDS)).isFalse();
 			assertThat(totalProcessed.get()).isEqualTo(4);
 
 			// Acknowledge the last pending record

spring-kafka/src/test/java/org/springframework/kafka/listener/ShareKafkaMessageListenerContainerIntegrationTests.java

@@ -26,7 +26,6 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -164,7 +163,7 @@ public void onShareRecord(ConsumerRecord<String, String> record,
 			assertThat(received).hasSize(3);
 			assertThat(acknowledgments).hasSize(3);
 			assertThat(acknowledgments).allMatch(Objects::nonNull);
-			assertThat(acknowledgments).allMatch(this::isAcknowledgedInternal);
+			assertThat(acknowledgments).allMatch(ShareKafkaMessageListenerContainerIntegrationTests::isAcknowledgedInternal);
 			assertThat(acknowledgments).allMatch(ack -> getAcknowledgmentTypeInternal(ack) == AcknowledgeType.ACCEPT);
 		}
 		finally {
@@ -267,18 +266,19 @@ public void onShareRecord(ConsumerRecord<String, String> record,
 			assertThat(firstBatchLatch.await(15, TimeUnit.SECONDS)).isTrue();
 			assertThat(pendingAcks).hasSize(3);
 
-			// Wait a bit to ensure no more records are processed while acknowledgments are pending
-			Thread.sleep(2000);
+			// Produce more records for second batch while first is pending
+			produceTestRecords(bootstrapServers, topic, 3);
+
+			// Verify second batch is NOT processed yet while acknowledgments are pending
+			// Using a latch that should NOT count down to verify blocking behavior
+			assertThat(secondBatchLatch.await(2, TimeUnit.SECONDS)).isFalse();
 			assertThat(processedCount.get()).isEqualTo(3);
 
 			// Acknowledge first batch
 			for (ShareAcknowledgment ack : pendingAcks) {
 				ack.acknowledge();
 			}
 
-			// Produce more records for second batch
-			produceTestRecords(bootstrapServers, topic, 3);
-
 			// Now second batch should be processed
 			assertThat(secondBatchLatch.await(15, TimeUnit.SECONDS)).isTrue();
 			assertThat(processedCount.get()).isEqualTo(6);
@@ -460,12 +460,21 @@ void shouldHandleContainerLifecycle(EmbeddedKafkaBroker broker) throws Exception
 		ContainerProperties containerProps = new ContainerProperties(topic);
 		containerProps.setExplicitShareAcknowledgment(true);
 
-		AtomicBoolean listenerCalled = new AtomicBoolean(false);
+		CountDownLatch firstProcessingLatch = new CountDownLatch(1);
+		CountDownLatch secondProcessingLatch = new CountDownLatch(1);
+		AtomicInteger callCount = new AtomicInteger(0);
+
 		containerProps.setMessageListener(new AcknowledgingShareConsumerAwareMessageListener<String, String>() {
 			@Override
 			public void onShareRecord(ConsumerRecord<String, String> record,
 					@Nullable ShareAcknowledgment acknowledgment, ShareConsumer<?, ?> consumer) {
-				listenerCalled.set(true);
+				int count = callCount.incrementAndGet();
+				if (count == 1) {
+					firstProcessingLatch.countDown();
+				}
+				else if (count == 2) {
+					secondProcessingLatch.countDown();
+				}
 				acknowledgment.acknowledge();
 			}
 		});
@@ -474,35 +483,27 @@ public void onShareRecord(ConsumerRecord<String, String> record,
 				new ShareKafkaMessageListenerContainer<>(factory, containerProps);
 		container.setBeanName("lifecycleTestContainer");
 
-		// Test initial state
 		assertThat(container.isRunning()).isFalse();
 
-		// Test start
 		container.start();
 		assertThat(container.isRunning()).isTrue();
 
-		// Test processing
 		produceTestRecords(bootstrapServers, topic, 1);
-		Thread.sleep(3000); // Give time for processing
-		assertThat(listenerCalled.get()).isTrue();
+		assertThat(firstProcessingLatch.await(10, TimeUnit.SECONDS)).isTrue();
 
-		// Test stop
 		container.stop();
 		assertThat(container.isRunning()).isFalse();
 
-		// Test restart
-		listenerCalled.set(false);
 		container.start();
 		assertThat(container.isRunning()).isTrue();
 
 		produceTestRecords(bootstrapServers, topic, 1);
-		Thread.sleep(3000);
-		assertThat(listenerCalled.get()).isTrue();
+		assertThat(secondProcessingLatch.await(10, TimeUnit.SECONDS)).isTrue();
 
 		container.stop();
 	}
 
-	private void testBasicMessageListener(DefaultShareConsumerFactory<String, String> factory,
+	private static void testBasicMessageListener(DefaultShareConsumerFactory<String, String> factory,
 			String topic, String bootstrapServers, String groupId) throws Exception {
 
 		setShareAutoOffsetResetEarliest(bootstrapServers, groupId);
@@ -529,7 +530,7 @@ private void testBasicMessageListener(DefaultShareConsumerFactory<String, String
 		}
 	}
 
-	private void testShareConsumerAwareListener(DefaultShareConsumerFactory<String, String> factory,
+	private static void testShareConsumerAwareListener(DefaultShareConsumerFactory<String, String> factory,
 			String topic, String bootstrapServers, String groupId) throws Exception {
 
 		setShareAutoOffsetResetEarliest(bootstrapServers, groupId);
@@ -566,7 +567,7 @@ public void onShareRecord(ConsumerRecord<String, String> record,
 		}
 	}
 
-	private void testAckListenerInImplicitMode(DefaultShareConsumerFactory<String, String> factory,
+	private static void testAckListenerInImplicitMode(DefaultShareConsumerFactory<String, String> factory,
 			String topic, String bootstrapServers, String groupId) throws Exception {
 
 		setShareAutoOffsetResetEarliest(bootstrapServers, groupId);
@@ -602,7 +603,7 @@ public void onShareRecord(ConsumerRecord<String, String> record,
 	}
 
 	// Utility methods
-	private Map<String, Object> createConsumerProps(String bootstrapServers, String groupId, boolean explicit) {
+	private static Map<String, Object> createConsumerProps(String bootstrapServers, String groupId, boolean explicit) {
 		Map<String, Object> props = new HashMap<>();
 		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
 		props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
@@ -614,23 +615,23 @@ private Map<String, Object> createConsumerProps(String bootstrapServers, String
 		return props;
 	}
 
-	private void produceTestRecords(String bootstrapServers, String topic, int count) throws Exception {
+	private static void produceTestRecords(String bootstrapServers, String topic, int count) throws Exception {
 		try (var producer = createProducer(bootstrapServers)) {
 			for (int i = 0; i < count; i++) {
 				producer.send(new ProducerRecord<>(topic, "key" + i, "value" + i)).get();
 			}
 		}
 	}
 
-	private KafkaProducer<String, String> createProducer(String bootstrapServers) {
+	private static KafkaProducer<String, String> createProducer(String bootstrapServers) {
 		Map<String, Object> producerProps = new HashMap<>();
 		producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
 		producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
 		producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
 		return new KafkaProducer<>(producerProps);
 	}
 
-	private void setShareAutoOffsetResetEarliest(String bootstrapServers, String groupId) throws Exception {
+	private static void setShareAutoOffsetResetEarliest(String bootstrapServers, String groupId) throws Exception {
 		Map<String, Object> adminProperties = new HashMap<>();
 		adminProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
 		ConfigEntry entry = new ConfigEntry("share.auto.offset.reset", "earliest");
@@ -645,7 +646,7 @@ private void setShareAutoOffsetResetEarliest(String bootstrapServers, String gro
 	/**
 	 * Helper method to access internal acknowledgment state for testing.
 	 */
-	private boolean isAcknowledgedInternal(ShareAcknowledgment ack) {
+	private static boolean isAcknowledgedInternal(ShareAcknowledgment ack) {
 		try {
 			java.lang.reflect.Method method = ack.getClass().getDeclaredMethod("isAcknowledged");
 			method.setAccessible(true);
@@ -659,7 +660,7 @@ private boolean isAcknowledgedInternal(ShareAcknowledgment ack) {
 	/**
 	 * Helper method to access internal acknowledgment type for testing.
 	 */
-	private AcknowledgeType getAcknowledgmentTypeInternal(ShareAcknowledgment ack) {
+	private static AcknowledgeType getAcknowledgmentTypeInternal(ShareAcknowledgment ack) {
 		try {
 			java.lang.reflect.Method method = ack.getClass().getDeclaredMethod("getAcknowledgmentType");
 			method.setAccessible(true);