(kafka) branch trunk updated: KAFKA-20786: Bound IntegrationTestUtils.readRecords on real elapsed time (#22790)

[lianetm]

This is an automated email from the ASF dual-hosted git repository.

lianetm pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 0dcc2f103e9 KAFKA-20786: Bound IntegrationTestUtils.readRecords on 
real elapsed time (#22790)
0dcc2f103e9 is described below

commit 0dcc2f103e9b1235fc25fb03f52ecf762b66f3a2
Author: Matthias J. Sax <matth...@confluent.io>
AuthorDate: Thu Jul 9 06:29:19 2026 -0700

    KAFKA-20786: Bound IntegrationTestUtils.readRecords on real elapsed time 
(#22790)
    
    readRecords() polls up to `waitTime` by counting a fixed 100ms per
    iteration rather than measuring the wall clock. A single consumer.poll()
    can block longer than the requested 100ms interval, and thus
    readReacord() can exceed the specified `waitTimeMs` by orders of
    magnitude.
    
    This PR bounds the loop on real elapsed time using a fixed deadline.
---
 .../apache/kafka/streams/integration/utils/IntegrationTestUtils.java | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git 
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java
 
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java
index 1fb35121a92..381e8e95629 100644
--- 
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java
+++ 
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java
@@ -1212,10 +1212,9 @@ public class IntegrationTestUtils {
         consumer.subscribe(singletonList(topic));
         final int pollIntervalMs = 100;
         consumerRecords = new ArrayList<>();
-        int totalPollTimeMs = 0;
-        while (totalPollTimeMs < waitTime &&
+        final long deadline = System.currentTimeMillis() + waitTime;
+        while (System.currentTimeMillis() < deadline &&
             continueConsuming(consumerRecords.size(), maxMessages)) {
-            totalPollTimeMs += pollIntervalMs;
             final ConsumerRecords<K, V> records = 
consumer.poll(Duration.ofMillis(pollIntervalMs));
 
             for (final ConsumerRecord<K, V> record : records) {