Improve CI test logging; stabilize share consumer and batch ITs

Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>

Author: sobychacko

build.gradle

@@ -157,14 +157,48 @@ configure(javaProjects) { subproject ->
 	[compileJava, compileTestJava]*.options*.compilerArgs = ['-Xlint:all,-options,-processing', '-parameters']
 
 	test {
+		// Default timeout for every testable method (unless @Timeout). Stops hung tests instead of wedging CI.
+		// Override: ./gradlew test -Djunit.jupiter.execution.timeout.default=PT20M  or  -PjunitTimeoutDefault=PT20M
+		def junitTimeoutDefault = System.getProperty('junit.jupiter.execution.timeout.default')
+				?: findProperty('junitTimeoutDefault')
+				?: 'PT10M'
+		systemProperty 'junit.jupiter.execution.timeout.default', junitTimeoutDefault
+
 		testLogging {
-			events 'skipped', 'failed'
+			// STARTED: last line before a hang usually identifies the stuck method. Class summary in afterSuite.
+			events 'started', 'failed'
 			showStandardStreams = project.hasProperty('showStandardStreams') ?: false
 			showExceptions = true
 			showStackTraces = true
 			exceptionFormat = 'full'
 		}
 
+		afterSuite { desc, result ->
+			// Top-level class only (skip engine root, nested classes, and empty suites)
+			if (desc.className == null || result.testCount == 0 || desc.parent?.className != null) {
+				return
+			}
+			long classMs = result.endTime - result.startTime
+			String duration = classMs >= 60_000
+					? String.format("%.2f min", classMs / 60_000.0)
+					: classMs >= 10_000
+							? String.format("%.2fs", classMs / 1000.0)
+							: "${classMs}ms"
+			String fqn = desc.className
+			String simpleName = fqn.contains('.') ? fqn.substring(fqn.lastIndexOf('.') + 1) : fqn
+			StringBuilder summary = new StringBuilder()
+			summary.append(simpleName).append(" finished in ").append(duration)
+					.append(" (").append(result.testCount).append(" tests")
+			if (result.failedTestCount > 0) {
+				summary.append(", ").append(result.failedTestCount).append(" failed")
+			}
+			if (result.skippedTestCount > 0) {
+				summary.append(", ").append(result.skippedTestCount).append(" skipped")
+			}
+			summary.append(')')
+			logger.lifecycle(summary.toString())
+		}
+
 		maxHeapSize = '1536m'
 		maxParallelForks = Runtime.runtime.availableProcessors().intdiv(2) ?: 1
 		useJUnitPlatform()

spring-kafka/src/test/java/org/springframework/kafka/annotation/BatchListenerConversion2Tests.java

@@ -75,6 +75,7 @@ public void testBatchOfPojosWithABadOne() throws Exception {
 		this.template.send(topic, "{\"bar\":\"baz\"}");
 		this.template.send(topic, "junk");
 		this.template.send(topic, "{\"bar\":\"baz\"}");
+		this.template.flush();
 		assertThat(listener.latch1.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(listener.badFoo).isInstanceOf(BadFoo.class);
 		assertThat(listener.receivedFoos).isEqualTo(2);

spring-kafka/src/test/java/org/springframework/kafka/core/DefaultShareConsumerFactoryTests.java

@@ -151,17 +151,19 @@ void integrationTestDefaultShareConsumerFactory(EmbeddedKafkaBroker broker) thro
 		final String groupId = "testGroup";
 		var bootstrapServers = broker.getBrokersAsString();
 
+		// Apply group config before producing so a joining share consumer consistently uses earliest.
+		setShareAutoOffsetResetEarliest(bootstrapServers, groupId);
+
 		var producerProps = new java.util.Properties();
 		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);
 
 		try (var producer = new KafkaProducer<String, String>(producerProps)) {
 			producer.send(new ProducerRecord<>(topic, "key", "integration-test-value")).get();
+			producer.flush();
 		}
 
-		setShareAutoOffsetResetEarliest(bootstrapServers, groupId);
-
 		var consumerProps = new HashMap<String, Object>();
 		consumerProps.put("bootstrap.servers", bootstrapServers);
 		consumerProps.put("key.deserializer", StringDeserializer.class);
@@ -172,11 +174,21 @@ void integrationTestDefaultShareConsumerFactory(EmbeddedKafkaBroker broker) thro
 		var consumer = factory.createShareConsumer(groupId, "myapp-client-id");
 		consumer.subscribe(Collections.singletonList(topic));
 
-		var records = consumer.poll(Duration.ofSeconds(10));
-		assertThat(records.count())
+		// First poll(s) may be empty (assignment / share coordinator); avoid a single-poll race on CI.
+		int total = 0;
+		String value = null;
+		long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(30);
+		while (total == 0 && System.currentTimeMillis() < deadline) {
+			var records = consumer.poll(Duration.ofSeconds(1));
+			total += records.count();
+			if (!records.isEmpty()) {
+				value = records.iterator().next().value();
+			}
+		}
+		assertThat(total)
 				.as("Should have received at least one record")
 				.isGreaterThan(0);
-		assertThat(records.iterator().next().value())
+		assertThat(value)
 				.as("Record value should match")
 				.isEqualTo("integration-test-value");
 		consumer.close();
@@ -213,6 +225,8 @@ private static List<String> runSharedConsumerTest(String topic, String groupId,
 			List<String> consumerIds, int recordCount, EmbeddedKafkaBroker broker) throws Exception {
 		var bootstrapServers = broker.getBrokersAsString();
 
+		setShareAutoOffsetResetEarliest(bootstrapServers, groupId);
+
 		var producerProps = new java.util.Properties();
 		producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
 		producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
@@ -224,8 +238,6 @@ private static List<String> runSharedConsumerTest(String topic, String groupId,
 			producer.flush();
 		}
 
-		setShareAutoOffsetResetEarliest(bootstrapServers, groupId);
-
 		List<String> allReceived = Collections.synchronizedList(new ArrayList<>());
 		var latch = new java.util.concurrent.CountDownLatch(recordCount);
 		ExecutorService executor = Executors.newCachedThreadPool();
@@ -255,11 +267,11 @@ private static List<String> runSharedConsumerTest(String topic, String groupId,
 			});
 		}
 
-		assertThat(latch.await(10, TimeUnit.SECONDS))
+		assertThat(latch.await(30, TimeUnit.SECONDS))
 			.as("All records should be received within timeout")
 			.isTrue();
 		executor.shutdown();
-		assertThat(executor.awaitTermination(10, TimeUnit.SECONDS))
+		assertThat(executor.awaitTermination(30, TimeUnit.SECONDS))
 			.as("Executor should terminate after shutdown")
 			.isTrue();
 		return allReceived;

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

@@ -36,14 +36,17 @@
 import org.apache.kafka.common.config.ConfigResource;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
+import org.awaitility.Awaitility;
 import org.jspecify.annotations.Nullable;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.annotation.EnableKafka;
 import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
 import org.springframework.kafka.config.ShareKafkaListenerContainerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.DefaultShareConsumerFactory;
@@ -75,6 +78,7 @@
 		"share-listener-error-handling-test",
 		"share-listener-factory-props-test"
 },
+		partitions = 1,
 		brokerProperties = {
 				"share.coordinator.state.topic.replication.factor=1",
 				"share.coordinator.state.topic.min.isr=1"
@@ -87,6 +91,9 @@ class ShareKafkaListenerIntegrationTests {
 	@Autowired
 	KafkaTemplate<String, String> kafkaTemplate;
 
+	@Autowired
+	KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;
+
 	@Test
 	void shouldSupportBasicShareKafkaListener() throws Exception {
 		final String topic = "share-listener-basic-test";
@@ -95,6 +102,7 @@ void shouldSupportBasicShareKafkaListener() throws Exception {
 
 		// Send test message
 		kafkaTemplate.send(topic, "basic-test-message");
+		kafkaTemplate.flush();
 
 		// Wait for processing
 		assertThat(BasicTestListener.latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -111,6 +119,7 @@ void shouldSupportExplicitAcknowledgmentWithShareAcknowledgment() throws Excepti
 		kafkaTemplate.send(topic, "accept", "accept-message");
 		kafkaTemplate.send(topic, "release", "release-message");
 		kafkaTemplate.send(topic, "reject", "reject-message");
+		kafkaTemplate.flush();
 
 		// Wait for processing
 		assertThat(ExplicitAckTestListener.latch.await(15, TimeUnit.SECONDS)).isTrue();
@@ -134,6 +143,7 @@ void shouldSupportShareConsumerAwareListener() throws Exception {
 
 		// Send test message
 		kafkaTemplate.send(topic, "consumer-aware-message");
+		kafkaTemplate.flush();
 
 		// Wait for processing
 		assertThat(ShareConsumerAwareTestListener.latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -147,8 +157,12 @@ void shouldSupportAcknowledgingShareConsumerAwareListener() throws Exception {
 		final String groupId = "share-ack-consumer-aware-group";
 		setShareAutoOffsetResetEarliest(this.broker.getBrokersAsString(), groupId);
 
+		// This test can run first (JUnit 6); wait for containers before producing so share consumers are subscribed.
+		awaitRunningShareListenerContainers();
+
 		// Send test message
 		kafkaTemplate.send(topic, "ack-consumer-aware-message");
+		kafkaTemplate.flush();
 
 		// Wait for processing
 		assertThat(AckShareConsumerAwareTestListener.latch.await(30, TimeUnit.SECONDS)).isTrue();
@@ -168,6 +182,7 @@ void shouldHandleMixedAcknowledgmentScenarios() throws Exception {
 		kafkaTemplate.send(topic, "success1", "success-message-1");
 		kafkaTemplate.send(topic, "success2", "success-message-2");
 		kafkaTemplate.send(topic, "retry", "retry-message");
+		kafkaTemplate.flush();
 
 		// Wait for processing
 		assertThat(MixedAckTestListener.processedLatch.await(15, TimeUnit.SECONDS)).isTrue();
@@ -189,6 +204,7 @@ void shouldHandleProcessingErrorsCorrectly() throws Exception {
 		kafkaTemplate.send(topic, "success", "success-message");
 		kafkaTemplate.send(topic, "error", "error-message");
 		kafkaTemplate.send(topic, "success2", "success-message-2");
+		kafkaTemplate.flush();
 
 		// Wait for processing
 		assertThat(ErrorHandlingTestListener.latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -206,6 +222,7 @@ void shouldSupportExplicitAcknowledgmentViaFactoryContainerProperties() throws E
 
 		// Send test message
 		kafkaTemplate.send(topic, "factory-test", "factory-props-message");
+		kafkaTemplate.flush();
 
 		// Wait for processing
 		assertThat(FactoryPropsTestListener.latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -243,6 +260,20 @@ private boolean isAcknowledgedInternal(ShareAcknowledgment ack) {
 		}
 	}
 
+	/**
+	 * Wait until all listener containers have started (needed when a test runs before others).
+	 */
+	private void awaitRunningShareListenerContainers() {
+		Awaitility.await()
+				.atMost(30, TimeUnit.SECONDS)
+				.pollInterval(100, TimeUnit.MILLISECONDS)
+				.untilAsserted(() -> {
+					assertThat(this.kafkaListenerEndpointRegistry.getListenerContainerIds()).isNotEmpty();
+					this.kafkaListenerEndpointRegistry.getListenerContainers().forEach(container ->
+							assertThat(container.isRunning()).isTrue());
+				});
+	}
+
 	@Configuration
 	@EnableKafka
 	static class TestConfig {
@@ -262,7 +293,7 @@ public ShareConsumerFactory<String, String> explicitShareConsumerFactory(Embedde
 			configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBrokersAsString());
 			configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
 			configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
-			configs.put("share.acknowledgement.mode", "explicit");
+			configs.put(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG, "explicit");
 			return new DefaultShareConsumerFactory<>(configs);
 		}
 
@@ -280,10 +311,10 @@ public ShareKafkaListenerContainerFactory<String, String> explicitShareKafkaList
 
 		@Bean
 		public ShareKafkaListenerContainerFactory<String, String> factoryPropsShareKafkaListenerContainerFactory(
-				ShareConsumerFactory<String, String> shareConsumerFactory) {
+				@Qualifier("explicitShareConsumerFactory") ShareConsumerFactory<String, String> explicitShareConsumerFactory) {
 			ShareKafkaListenerContainerFactory<String, String> factory =
-					new ShareKafkaListenerContainerFactory<>(shareConsumerFactory);
-			// Configure explicit acknowledgment via factory's container properties
+					new ShareKafkaListenerContainerFactory<>(explicitShareConsumerFactory);
+			// Configure explicit acknowledgment via factory's container properties (consumer must use explicit mode too)
 			factory.getContainerProperties().setExplicitShareAcknowledgment(true);
 			return factory;
 		}