(kafka) branch trunk updated: MINOR: Refactor awaitNonEmptyRecords to remove dead code and improve clarity (#19868)

[fr...@apache.org]

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

frankvicky 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 8f1a1cd8387 MINOR: Refactor awaitNonEmptyRecords to remove dead code and improve clarity (#19868)
8f1a1cd8387 is described below

commit 8f1a1cd8387188be90d6baee8ed678706dd24bc4
Author: Jing-Jia Hung <32...@users.noreply.github.com>
AuthorDate: Mon Jun 2 12:46:56 2025 +0800

    MINOR: Refactor awaitNonEmptyRecords to remove dead code and improve clarity (#19868)
    
    This refactor improves the implementation of `awaitNonEmptyRecords` by:
    
    - Removing the unreachable `throw new IllegalStateException` statement,
    which was dead code due to `pollRecordsUntilTrue` throwing exceptions on
    timeout.
    - Eliminating the use of `return` inside the lambda, which relies on
    non-local returns that can be confusing and error-prone in Scala.
    
    Reviewers: Yung <yu...@gmail.com>, Ken Huang
     <s7...@gmail.com>, TengYao Chi <fr...@apache.org>
    
    ---------
    
    Co-authored-by: Jing-Jia Hung <ji...@Jing-JiadeiMac.local>
---
 .../scala/integration/kafka/api/AbstractConsumerTest.scala     | 10 ++++++----
 1 file changed, 6 insertions(+), 4 deletions(-)

diff --git a/core/src/test/scala/integration/kafka/api/AbstractConsumerTest.scala b/core/src/test/scala/integration/kafka/api/AbstractConsumerTest.scala
index 517614d84a1..6a60621308b 100644
--- a/core/src/test/scala/integration/kafka/api/AbstractConsumerTest.scala
+++ b/core/src/test/scala/integration/kafka/api/AbstractConsumerTest.scala
@@ -94,12 +94,14 @@ abstract class AbstractConsumerTest extends BaseRequestTest {
   def awaitNonEmptyRecords[K, V](consumer: Consumer[K, V],
                                  partition: TopicPartition,
                                  pollTimeoutMs: Long = 100): ConsumerRecords[K, V] = {
+    var result: ConsumerRecords[K, V] = null
+
     TestUtils.pollRecordsUntilTrue(consumer, (polledRecords: ConsumerRecords[K, V]) => {
-      if (polledRecords.records(partition).asScala.nonEmpty)
-        return polledRecords
-      false
+      val hasRecords = !polledRecords.records(partition).isEmpty
+      if (hasRecords) result = polledRecords
+      hasRecords
     }, s"Consumer did not consume any messages for partition $partition before timeout.", JTestUtils.DEFAULT_MAX_WAIT_MS, pollTimeoutMs)
-    throw new IllegalStateException("Should have timed out before reaching here")
+    result
   }
 
   /**