From 6249c0afefd9d9bfba74e4569db09f9218e69fe4 Mon Sep 17 00:00:00 2001 From: Luke Chen Date: Fri, 17 Jan 2025 16:16:15 +0800 Subject: [PATCH] [FLINK-37159][runtime] Fix the test timeout by yielding Modifier thread --- .../MemoryManagerConcurrentModReleaseTest.java | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/memory/MemoryManagerConcurrentModReleaseTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/memory/MemoryManagerConcurrentModReleaseTest.java index 75e435c56848dd..795e79af5f721a 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/memory/MemoryManagerConcurrentModReleaseTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/memory/MemoryManagerConcurrentModReleaseTest.java @@ -28,6 +28,7 @@ /** Validate memory release under concurrent modification exceptions. */ class MemoryManagerConcurrentModReleaseTest { + private static long testTimeoutMs = 5000; @Test void testConcurrentModificationOnce() throws MemoryAllocationException { @@ -61,7 +62,7 @@ void testConcurrentModificationWhileReleasing() throws Exception { memMan.allocatePages(this, segs, numSegments); // start a thread that performs concurrent modifications - Modifier mod = new Modifier(segs); + Modifier mod = new Modifier(segs, System.currentTimeMillis()); Thread modRunner = new Thread(mod); modRunner.start(); @@ -80,11 +81,14 @@ void testConcurrentModificationWhileReleasing() throws Exception { private static class Modifier implements Runnable { private final ArrayList toModify; + private final long startTimeMs; private volatile boolean running = true; + private volatile boolean timeout = false; - private Modifier(ArrayList toModify) { + private Modifier(ArrayList toModify, long startTimeMs) { this.toModify = toModify; + this.startTimeMs = startTimeMs; } public void cancel() { @@ -97,7 +101,13 @@ public void run() { try { MemorySegment seg = toModify.remove(0); toModify.add(seg); - } catch (IndexOutOfBoundsException e) { + // if the test running time reaches TEST_TIMEOUT_MS, we should lower the + // priority of the Modifier thread to let segment release completes sooner. + if (timeout || System.currentTimeMillis() - startTimeMs > testTimeoutMs) { + timeout = true; + Thread.sleep(1); + } + } catch (IndexOutOfBoundsException | InterruptedException e) { // may happen, just retry } }