You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by yh...@apache.org on 2024/02/11 14:37:47 UTC
(beam) branch master updated: Fix two flaky tests (#30278)
This is an automated email from the ASF dual-hosted git repository.
yhu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 6d10e0f40fd Fix two flaky tests (#30278)
6d10e0f40fd is described below
commit 6d10e0f40fd2411b9c4102a7c219535fe7a46066
Author: Shunping Huang <sh...@google.com>
AuthorDate: Sun Feb 11 09:37:40 2024 -0500
Fix two flaky tests (#30278)
* Fix a flaky test caused by race condition
* Use notify and wait per reviewer request
* Increase assert threshold to allow more room for extra threads
* Move synchronized block outside of assertThrows
* Replace this with isReady when defining synchronized block
---
.../beam/sdk/fn/data/BeamFnDataInboundObserverTest.java | 11 +++++++++++
.../beam/sdk/util/UnboundedScheduledExecutorServiceTest.java | 2 +-
2 files changed, 12 insertions(+), 1 deletion(-)
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/fn/data/BeamFnDataInboundObserverTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/fn/data/BeamFnDataInboundObserverTest.java
index bd279cee208..91556cf2ac1 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/fn/data/BeamFnDataInboundObserverTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/fn/data/BeamFnDataInboundObserverTest.java
@@ -28,6 +28,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.beam.model.fnexecution.v1.BeamFnApi;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
@@ -146,10 +147,15 @@ public class BeamFnDataInboundObserverTest {
Arrays.asList(DataEndpoint.create(TRANSFORM_ID, CODER, (value) -> {})),
Collections.emptyList());
+ AtomicBoolean isReady = new AtomicBoolean(false);
Future<?> future =
executor.submit(
() -> {
observer.accept(dataWith("ABC"));
+ synchronized (isReady) {
+ isReady.set(true);
+ isReady.notify();
+ }
assertThrows(
BeamFnDataInboundObserver.CloseException.class,
() -> {
@@ -165,6 +171,11 @@ public class BeamFnDataInboundObserverTest {
Future<?> future2 =
executor.submit(
() -> {
+ synchronized (isReady) {
+ while (!isReady.get()) {
+ isReady.wait();
+ }
+ }
observer.close();
return null;
});
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/UnboundedScheduledExecutorServiceTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/UnboundedScheduledExecutorServiceTest.java
index b8efa292bd2..4b5bb55439a 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/UnboundedScheduledExecutorServiceTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/UnboundedScheduledExecutorServiceTest.java
@@ -553,7 +553,7 @@ public class UnboundedScheduledExecutorServiceTest {
LOG.info("Created {} threads to execute at most 100 parallel tasks", largestPool);
// Ideally we would never create more than 100, however with contention it is still possible
// some extra threads will be created.
- assertTrue(largestPool <= 104);
+ assertTrue(largestPool <= 110);
executorService.shutdown();
}
}