- Revert latch-based negative assertions back to Thread.sleep() in

  constraint and integration tests, as testing for non-occurrence
  (await expecting false) still blocks for full timeout duration

- Use chained assertion style for acknowledgment validations

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

Author: sobychacko

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

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

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

@@ -161,10 +161,10 @@ public void onShareRecord(ConsumerRecord<String, String> record,
 		try {
 			assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue();
 			assertThat(received).hasSize(3);
-			assertThat(acknowledgments).hasSize(3);
-			assertThat(acknowledgments).allMatch(Objects::nonNull);
-			assertThat(acknowledgments).allMatch(ShareKafkaMessageListenerContainerIntegrationTests::isAcknowledgedInternal);
-			assertThat(acknowledgments).allMatch(ack -> getAcknowledgmentTypeInternal(ack) == AcknowledgeType.ACCEPT);
+			assertThat(acknowledgments).hasSize(3)
+					.allMatch(Objects::nonNull)
+					.allMatch(ShareKafkaMessageListenerContainerIntegrationTests::isAcknowledgedInternal)
+					.allMatch(ack -> AcknowledgeType.ACCEPT.equals(getAcknowledgmentTypeInternal(ack)));
 		}
 		finally {
 			container.stop();
@@ -270,8 +270,7 @@ public void onShareRecord(ConsumerRecord<String, String> record,
 			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();
+			Thread.sleep(2000);
 			assertThat(processedCount.get()).isEqualTo(3);
 
 			// Acknowledge first batch