(kafka) branch trunk updated: KAFKA-20718: Fix flaky test OffsetStorageReaderTest.testClosingOffsetReaderWhenOffsetStoreHangsAndHasIncompleteFutures (#22632)

[vi...@apache.org]

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

viktorsomogyi 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 c1cdb9efcdc KAFKA-20718: Fix flaky test OffsetStorageReaderTest.testClosingOffsetReaderWhenOffsetStoreHangsAndHasIncompleteFutures (#22632)
c1cdb9efcdc is described below

commit c1cdb9efcdc1ef058836a565a2f59b97d9abe342
Author: majialong <ma...@gmail.com>
AuthorDate: Mon Sep 21 20:01:47 2026 +0800

    KAFKA-20718: Fix flaky test OffsetStorageReaderTest.testClosingOffsetReaderWhenOffsetStoreHangsAndHasIncompleteFutures (#22632)
    
    The test had a race condition. It released `latchTask1` inside the
    mocked `offsetBackingStore.get()`, but at that point the worker thread
    may not have returned from `offsetBackingStore.get()` yet.
    
    As a result, the test thread could resume and run the verification
    before `OffsetStorageReaderImpl` had reached `offsetReadFuture.get()`,
    causing the test to fail intermittently.
    
    This fix moves the latch release to the mocked `hangingFuture.get()`, so
    the test only continues after the worker is actually blocked while
    waiting on the future.
    
    Reviewers: Viktor Somogyi-Vass <vi...@gmail.com>
---
 .../java/org/apache/kafka/connect/storage/OffsetStorageReaderTest.java | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/OffsetStorageReaderTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/OffsetStorageReaderTest.java
index 01646c1323f..7fbfa94f465 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/OffsetStorageReaderTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/OffsetStorageReaderTest.java
@@ -107,6 +107,7 @@ public class OffsetStorageReaderTest {
 
         // Mock hanging future
         doAnswer(invocation -> {
+            latchTask1.countDown();
             CompletableFuture<Void> future = new CompletableFuture<>();
             future.get(9999, TimeUnit.SECONDS);
             throw new RuntimeException("Should never get here");
@@ -121,7 +122,6 @@ public class OffsetStorageReaderTest {
                     if (callCount == 0) {
                         callCount += 1;
                         // First connector task
-                        latchTask1.countDown();
                         return hangingFuture;
                     } else {
                         // Second connector task
@@ -144,7 +144,6 @@ public class OffsetStorageReaderTest {
         latchTask1.await();
 
         verify(offsetBackingStore, times(1)).get(any());
-        verify(hangingFuture, times(1)).get();
 
         // Another connector task thread calls `offsets()` --> hangs on offsetBackingStore.get()
         // --> the future is never added to `offsetStorageReaderImpl.offsetReadFutures`