This is an automated email from the ASF dual-hosted git repository. MartijnVisser pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 06a929fbd6f9ba08b8397f30d26c67af8d69e5bf Author: Martijn Visser <[email protected]> AuthorDate: Tue Jun 16 15:12:10 2026 -0700 [FLINK-37663][connector-base] Add test reproducing FutureCompletingBlockingQueue lost wakeup A freed slot signals a putter that wakeUpPuttingThread() already released instead of the genuinely waiting putter; the test reproduces the stall and fails without the fix. Its @VisibleForTesting accessor is recorded in the architecture violation store, like the existing take(). Generated-by: Claude Code (Claude Opus 4.8) --- .../f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e | 1 + .../FutureCompletingBlockingQueue.java | 10 ++++ .../FutureCompletingBlockingQueueTest.java | 54 ++++++++++++++++++++++ 3 files changed, 65 insertions(+) diff --git a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e index caac78113b2..a6052759e03 100644 --- a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e +++ b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e @@ -191,6 +191,7 @@ Method <org.apache.flink.connector.base.source.hybrid.HybridSource$HybridSourceB Method <org.apache.flink.connector.base.source.hybrid.HybridSourceSplitEnumerator.checkAndSignalNoMoreSplits(org.apache.flink.api.connector.source.SplitEnumeratorContext, int, int, int)> calls method <org.apache.flink.api.connector.source.SupportsIntermediateNoMoreSplits.signalIntermediateNoMoreSplits(int)> in (HybridSourceSplitEnumerator.java:450) Method <org.apache.flink.connector.base.source.hybrid.HybridSourceSplitEnumerator.checkAndSignalNoMoreSplits(org.apache.flink.api.connector.source.SplitEnumeratorContext, int, int, int)> checks instanceof <org.apache.flink.api.connector.source.SupportsIntermediateNoMoreSplits> in (HybridSourceSplitEnumerator.java:442) Method <org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.getNumAliveFetchers()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (SplitFetcherManager.java:0) +Method <org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue.getNumberOfQueuedPutters()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (FutureCompletingBlockingQueue.java:0) Method <org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue.take()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (FutureCompletingBlockingQueue.java:0) Method <org.apache.flink.connector.datagen.functions.FromElementsGeneratorFunction.getSerializer()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (FromElementsGeneratorFunction.java:0) Method <org.apache.flink.connector.datagen.functions.IndexLookupGeneratorFunction.getSerializer()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (IndexLookupGeneratorFunction.java:0) diff --git a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/synchronization/FutureCompletingBlockingQueue.java b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/synchronization/FutureCompletingBlockingQueue.java index a341c52b9d2..aaf90f4b5ee 100644 --- a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/synchronization/FutureCompletingBlockingQueue.java +++ b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/synchronization/FutureCompletingBlockingQueue.java @@ -307,6 +307,16 @@ public class FutureCompletingBlockingQueue<T> { } } + @VisibleForTesting + int getNumberOfQueuedPutters() { + lock.lock(); + try { + return notFull.size(); + } finally { + lock.unlock(); + } + } + /** * Gracefully wakes up the thread with the given {@code threadIndex} if it is blocked in adding * an element. to the queue. If the thread is blocked in {@link #put(int, Object)} it will diff --git a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/synchronization/FutureCompletingBlockingQueueTest.java b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/synchronization/FutureCompletingBlockingQueueTest.java index a19e07346e0..65e7ff38b92 100644 --- a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/synchronization/FutureCompletingBlockingQueueTest.java +++ b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/synchronization/FutureCompletingBlockingQueueTest.java @@ -20,14 +20,18 @@ package org.apache.flink.connector.base.source.reader.synchronization; import org.apache.flink.connector.base.source.reader.SourceReaderOptions; import org.apache.flink.runtime.io.AvailabilityProvider; +import org.apache.flink.runtime.testutils.CommonTestUtils; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; import java.util.ArrayList; import java.util.BitSet; import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; @@ -102,6 +106,56 @@ class FutureCompletingBlockingQueueTest { assertThat(latch.getCount()).isEqualTo(0); } + /** + * A putter gracefully woken via {@link FutureCompletingBlockingQueue#wakeUpPuttingThread(int)} + * must not prevent another, genuinely waiting putter from being signalled once a slot frees up. + * See FLINK-37663. + */ + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void testWakeUpDoesNotStrandAnotherPutter() throws Exception { + final FutureCompletingBlockingQueue<Integer> queue = new FutureCompletingBlockingQueue<>(1); + + queue.put(2, 0); + + final AtomicBoolean wokenPutterResult = new AtomicBoolean(true); + final Thread wokenPutter = + new Thread(() -> wokenPutterResult.set(putUnchecked(queue, 0, 1)), "wokenPutter"); + final CountDownLatch genuinePutterDone = new CountDownLatch(1); + final Thread genuinePutter = + new Thread( + () -> { + putUnchecked(queue, 1, 2); + genuinePutterDone.countDown(); + }, + "genuinePutter"); + + // Block putter 0 before putter 1 so putter 0 is at the head of the wait queue. + wokenPutter.start(); + CommonTestUtils.waitUntilCondition(() -> queue.getNumberOfQueuedPutters() == 1); + genuinePutter.start(); + CommonTestUtils.waitUntilCondition(() -> queue.getNumberOfQueuedPutters() == 2); + + queue.wakeUpPuttingThread(0); + wokenPutter.join(); + assertThat(wokenPutterResult).isFalse(); + + queue.poll(); + + assertThat(genuinePutterDone.await(10, TimeUnit.SECONDS)) + .as("A still-waiting putter must be signalled when a slot frees up") + .isTrue(); + } + + private static boolean putUnchecked( + FutureCompletingBlockingQueue<Integer> queue, int threadIndex, int value) { + try { + return queue.put(threadIndex, value); + } catch (InterruptedException e) { + return fail("Putting thread interrupted unexpectedly."); + } + } + @Test void testConcurrency() throws InterruptedException { FutureCompletingBlockingQueue<Integer> queue = new FutureCompletingBlockingQueue<>(5);
