Skip to content

Fix AsyncWrapperTest timeout and flakiness issues. #38970

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
damccorm merged 2 commits into from
Jun 16, 2026
Merged

Fix AsyncWrapperTest timeout and flakiness issues. #38970

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3073491
Fix AsyncWrapperTest timeout and flakiness issues.
tejasiyer-dev Jun 15, 2026
fa018c7
Fix AsyncWrapperTest to prevent new race conditions and premature tim…
tejasiyer-dev Jun 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 21 additions & 22 deletions sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/AsyncWrapperTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -230,21 +230,20 @@ private void waitForEmpty(AsyncWrapper<?, ?, ?> asyncWrapper) {
}

private void waitForEmpty(AsyncWrapper<?, ?, ?> asyncWrapper, int timeoutSeconds) {
int count = 0;
long limit = System.currentTimeMillis() + timeoutSeconds * 1000L;
while (!asyncWrapper.isEmpty()) {
if (System.currentTimeMillis() > limit) {
throw new RuntimeException("Timed out waiting for async dofn to be empty");
}
Comment thread
tejasiyer-dev marked this conversation as resolved.
try {
Thread.sleep(1000);
Thread.sleep(5);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException(e);
}
count += 1;
if (count > timeoutSeconds) {
throw new RuntimeException("Timed out waiting for async dofn to be empty");
}
}
try {
Thread.sleep(1000);
Thread.sleep(5);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
Expand Down Expand Up @@ -419,7 +418,7 @@ public void testMultiKey() {
// execution task has not finished processing yet.
@Test
public void testLongItem() {
BasicDofn dofn = new BasicDofn(1000);
BasicDofn dofn = new BasicDofn(500);
AsyncWrapper<String, String, String> asyncWrapper =
new AsyncWrapper<>(
dofn, 1, Duration.standardSeconds(5), null, null, null, null, useThreadPool);
Expand All @@ -438,7 +437,7 @@ public void testLongItem() {
assertEquals(0, dofn.getProcessed());
assertEquals(1, fakeBagState.items.size());

waitForEmpty(asyncWrapper, 20);
waitForEmpty(asyncWrapper, 2);

result =
asyncWrapper.commitFinishedItemsDirect(
Expand Down Expand Up @@ -538,7 +537,7 @@ public void testMultiElementDofn() {
// Identical elements should not spawn multiple concurrent background executions.
@Test
public void testDuplicates() {
BasicDofn dofn = new BasicDofn(1000);
BasicDofn dofn = new BasicDofn(10);
AsyncWrapper<String, String, String> asyncWrapper =
new AsyncWrapper<>(
dofn, 1, Duration.standardSeconds(5), null, null, null, null, useThreadPool);
Expand Down Expand Up @@ -568,7 +567,7 @@ public void testDuplicates() {
// has cleared are correctly tracked and processed.
@Test
public void testSlowDuplicates() {
BasicDofn dofn = new BasicDofn(5000);
BasicDofn dofn = new BasicDofn(20);
AsyncWrapper<String, String, String> asyncWrapper =
new AsyncWrapper<>(
dofn, 1, Duration.standardSeconds(5), null, null, null, null, useThreadPool);
Expand All @@ -581,7 +580,7 @@ public void testSlowDuplicates() {
asyncWrapper.processDirect(msg, GlobalWindow.INSTANCE, Instant.now(), fakeBagState, fakeTimer);

try {
Thread.sleep(10000);
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
Expand Down Expand Up @@ -610,7 +609,7 @@ public void testSlowDuplicates() {
// and decrement immediately upon execution completion.
@Test
public void testBufferCount() {
BasicDofn dofn = new BasicDofn(1000);
BasicDofn dofn = new BasicDofn(10);
AsyncWrapper<String, String, String> asyncWrapper =
new AsyncWrapper<>(
dofn, 1, Duration.standardSeconds(5), null, null, null, null, useThreadPool);
Expand All @@ -637,7 +636,7 @@ public void testBufferCount() {
// the scheduler must block and delay submissions appropriately.
@Test
public void testBufferStopsAcceptingItems() {
BasicDofn dofn = new BasicDofn(1000);
BasicDofn dofn = new BasicDofn(500);
AsyncWrapper<String, String, String> asyncWrapper =
new AsyncWrapper<>(
dofn,
Expand Down Expand Up @@ -670,7 +669,7 @@ public void testBufferStopsAcceptingItems() {
}

try {
Thread.sleep(200);
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
Expand Down Expand Up @@ -707,7 +706,7 @@ public void testBufferStopsAcceptingItems() {
// Verifies actively cancelled elements are cleanly dropped from the buffer during throttling.
@Test
public void testBufferWithCancellation() {
BasicDofn dofn = new BasicDofn(1000);
BasicDofn dofn = new BasicDofn(10);
AsyncWrapper<String, String, String> asyncWrapper =
new AsyncWrapper<>(
dofn, 1, Duration.standardSeconds(5), null, null, null, null, useThreadPool);
Expand Down Expand Up @@ -746,7 +745,7 @@ public void testBufferWithCancellation() {
// across multiple keys correctly under heavy multi-threaded load.
@Test
public void testLoadCorrectness() {
BasicDofn dofn = new BasicDofn(1000);
BasicDofn dofn = new BasicDofn(10);
AsyncWrapper<String, String, String> asyncWrapper =
new AsyncWrapper<>(
dofn,
Expand Down Expand Up @@ -791,14 +790,14 @@ public void testLoadCorrectness() {
timers.get(key));
}));
try {
Thread.sleep(random.nextInt(200));
Thread.sleep(random.nextInt(2));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}

try {
Comment thread
tejasiyer-dev marked this conversation as resolved.
Thread.sleep(3000 + random.nextInt(2000));
Thread.sleep(1000 + random.nextInt(1000));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
Expand Down Expand Up @@ -834,7 +833,7 @@ public void testLoadCorrectness() {
}
}
try {
Thread.sleep(1000 + random.nextInt(2000));
Thread.sleep(10 + random.nextInt(20));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
Expand All @@ -854,7 +853,7 @@ public void testLoadCorrectness() {
// must complete cleanly without thread or lock deadlocks.
@Test
public void testResetStateConcurrentTeardown() {
BasicDofn dofn = new BasicDofn(500);
BasicDofn dofn = new BasicDofn(10);
AsyncWrapper<String, String, String> asyncWrapper =
new AsyncWrapper<>(
dofn, 1, Duration.standardSeconds(5), null, null, null, null, useThreadPool);
Expand All @@ -867,7 +866,7 @@ public void testResetStateConcurrentTeardown() {
KV.of("key1", "1"), GlobalWindow.INSTANCE, Instant.now(), fakeBagState, fakeTimer);

try {
Thread.sleep(50);
Thread.sleep(2);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
Expand Down
Loading