This is an automated email from the ASF dual-hosted git repository.
1996fanrui pushed a commit to branch release-1.20
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.20 by this push:
new 1b5f5767403 [FLINK-40657][connector-base] Release per-fetcher state in
FutureCompletingBlockingQueue (#29389)
1b5f5767403 is described below
commit 1b5f5767403452d7628a86c2f2bd9073722d4ed2
Author: Alexandru Stefanica <[email protected]>
AuthorDate: Sun Oct 11 19:18:25 2026 +0300
[FLINK-40657][connector-base] Release per-fetcher state in
FutureCompletingBlockingQueue (#29389)
Fetcher indexes are never recycled, so the array of ConditionAndFlag grew
without bound. Key the state by producer index instead and release it from
the SplitFetcher shutdown hook.
Generated-by: Claude Code (Claude Opus 5)
(cherry picked from commit 953f843d578572cc74343c1f1a1ddb64d5c5d3e5)
---
.../source/reader/fetcher/SplitFetcherManager.java | 1 +
.../FutureCompletingBlockingQueue.java | 142 ++++++++++++---
.../reader/fetcher/SplitFetcherManagerTest.java | 53 ++++++
.../FutureCompletingBlockingQueueTest.java | 198 +++++++++++++++++++++
.../source/reader/synchronization/QueueProbe.java | 29 +++
5 files changed, 400 insertions(+), 23 deletions(-)
diff --git
a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManager.java
b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManager.java
index 4f7219eb2b8..5eb4ccf4ec0 100644
---
a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManager.java
+++
b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManager.java
@@ -267,6 +267,7 @@ public abstract class SplitFetcherManager<E, SplitT extends
SourceSplit> {
errorHandler,
() -> {
fetchers.remove(fetcherId);
+ elementsQueue.releaseProducer(fetcherId);
// We need this to synchronize status of fetchers
to concurrent partners
// as
// ConcurrentHashMap's aggregate status methods
including size, isEmpty,
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 799fd0c4489..d023b4985ef 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
@@ -27,7 +27,8 @@ import javax.annotation.concurrent.GuardedBy;
import java.lang.reflect.Field;
import java.util.ArrayDeque;
-import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
@@ -37,6 +38,7 @@ import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkState;
/**
* A custom implementation of blocking queue in combination with a {@link
CompletableFuture} that is
@@ -69,6 +71,11 @@ import static
org.apache.flink.util.Preconditions.checkArgument;
* capacity limits, without interrupting the thread. This is done via the
{@link
* #wakeUpPuttingThread(int)} method.
*
+ * <p>The queue keeps a small amount of per-producer state to support this. A
producer that is
+ * permanently done with the queue must be handed to {@link
#releaseProducer(int)}, the lifecycle
+ * counterpart of {@link #wakeUpPuttingThread(int)}, otherwise that state is
retained for the
+ * lifetime of the queue.
+ *
* @param <T> the type of the elements in the queue.
*/
@Internal
@@ -102,9 +109,15 @@ public class FutureCompletingBlockingQueue<T> {
@GuardedBy("lock")
private final Queue<Condition> notFull;
- /** The per-thread conditions and wakeUp flags. */
+ /**
+ * The per-producer conditions and wakeUp flags, keyed by the producer's
thread index.
+ *
+ * <p>Entries are created on demand by {@link #conditionAndFlagFor(int)}
and removed by {@link
+ * #releaseProducer(int)}, so this map is bounded by the peak number of
live or in-flight
+ * producers rather than by the largest producer index ever allocated.
+ */
@GuardedBy("lock")
- private ConditionAndFlag[] putConditionAndFlags;
+ private final Map<Integer, ConditionAndFlag> putConditionAndFlags;
public FutureCompletingBlockingQueue() {
this(SourceReaderOptions.ELEMENT_QUEUE_CAPACITY.defaultValue());
@@ -115,7 +128,7 @@ public class FutureCompletingBlockingQueue<T> {
this.capacity = capacity;
this.queue = new ArrayDeque<>(capacity);
this.lock = new ReentrantLock();
- this.putConditionAndFlags = new ConditionAndFlag[1];
+ this.putConditionAndFlags = new HashMap<>();
this.notFull = new ArrayDeque<>();
// initially the queue is empty and thus unavailable
@@ -317,6 +330,28 @@ public class FutureCompletingBlockingQueue<T> {
}
}
+ int numberOfProducerStates() {
+ lock.lock();
+ try {
+ return putConditionAndFlags.size();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ boolean hasProducerState(int producerIndex) {
+ lock.lock();
+ try {
+ return putConditionAndFlags.containsKey(producerIndex);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ Lock lock() {
+ return lock;
+ }
+
/**
* 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
@@ -330,12 +365,55 @@ public class FutureCompletingBlockingQueue<T> {
public void wakeUpPuttingThread(int threadIndex) {
lock.lock();
try {
- maybeCreateCondition(threadIndex);
- ConditionAndFlag caf = putConditionAndFlags[threadIndex];
- if (caf != null) {
- caf.setWakeUp(true);
- caf.condition().signal();
+ // Creates the entry when absent, deliberately: the flag has to be
sticky, so that a
+ // producer woken before it ever calls put() still observes the
request and returns
+ // immediately instead of parking on a full queue.
+ final ConditionAndFlag caf = conditionAndFlagFor(threadIndex);
+ caf.setWakeUp(true);
+ caf.condition().signal();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Releases the per-producer wakeup state held for {@code threadIndex}.
+ *
+ * <p>Call this once the producer with that index is permanently finished
with this queue.
+ * Without it the queue retains each condition created by a wakeup request
or by a put attempt
+ * made while the queue was full. {@code SplitFetcherManager} hands out a
fresh, never-recycled
+ * index per {@code SplitFetcher}, so for a source whose fetchers are
short-lived that set
+ * otherwise grows without bound for the lifetime of the JVM.
+ *
+ * <p>The call is idempotent and safe for an index that was never used.
+ *
+ * <p><b>The caller must not be inside {@link #put(int, Object)} for this
index.</b> Releasing a
+ * producer that is still running would drop a pending wakeUp flag it has
yet to observe, and it
+ * could then park on a full queue after having been told to stop. {@code
SplitFetcher}
+ * satisfies this by running its shutdown hook only after its run loop has
exited.
+ *
+ * <p>For the same reason this must be the last interaction with the queue
for that index: a
+ * later {@link #wakeUpPuttingThread(int)} would recreate the entry, and
nothing would remove it
+ * again. {@code SplitFetcher} satisfies this in its normal lifecycle
because it clears the
+ * current task before its run loop exits and invokes the shutdown hook
afterward.
+ *
+ * @param threadIndex The number identifying the producer thread, as
passed to {@link #put(int,
+ * Object)}.
+ */
+ public void releaseProducer(int threadIndex) {
+ lock.lock();
+ try {
+ final ConditionAndFlag caf = putConditionAndFlags.get(threadIndex);
+ if (caf == null) {
+ return;
+ }
+ if (caf.hasWaitingPutter()) {
+ // The condition may already have been removed from notFull
after being signalled,
+ // while its putter is still waiting to reacquire the lock.
Dropping the entry in
+ // that interval would discard a wakeUp flag the putter has
yet to observe.
+ return;
}
+ putConditionAndFlags.remove(threadIndex);
} finally {
lock.unlock();
}
@@ -370,15 +448,17 @@ public class FutureCompletingBlockingQueue<T> {
@GuardedBy("lock")
private void waitOnPut(int fetcherIndex) throws InterruptedException {
- maybeCreateCondition(fetcherIndex);
- Condition cond = putConditionAndFlags[fetcherIndex].condition();
- notFull.add(cond);
+ final ConditionAndFlag caf = conditionAndFlagFor(fetcherIndex);
+ final Condition cond = caf.condition();
+ caf.startWaiting();
try {
+ notFull.add(cond);
cond.await();
} finally {
// drop the condition once the thread stops waiting, so a later
signalNextPutter()
// does not signal a putter that is no longer waiting
notFull.remove(cond);
+ caf.stopWaiting();
}
}
@@ -389,22 +469,24 @@ public class FutureCompletingBlockingQueue<T> {
}
}
+ /**
+ * Returns the state for {@code threadIndex}, creating it when absent.
Only {@link
+ * #wakeUpPuttingThread(int)} and {@link #waitOnPut(int)} call this, so a
producer that is never
+ * woken and never blocks costs nothing.
+ */
@GuardedBy("lock")
- private void maybeCreateCondition(int threadIndex) {
- if (putConditionAndFlags.length < threadIndex + 1) {
- putConditionAndFlags = Arrays.copyOf(putConditionAndFlags,
threadIndex + 1);
- }
-
- if (putConditionAndFlags[threadIndex] == null) {
- putConditionAndFlags[threadIndex] = new
ConditionAndFlag(lock.newCondition());
- }
+ private ConditionAndFlag conditionAndFlagFor(int threadIndex) {
+ return putConditionAndFlags.computeIfAbsent(
+ threadIndex, ignored -> new
ConditionAndFlag(lock.newCondition()));
}
@GuardedBy("lock")
private boolean getAndResetWakeUpFlag(int threadIndex) {
- maybeCreateCondition(threadIndex);
- if (putConditionAndFlags[threadIndex].getWakeUp()) {
- putConditionAndFlags[threadIndex].setWakeUp(false);
+ // Deliberately does not create the state: an absent entry cannot
carry a wakeUp flag, and
+ // waitOnPut() creates it a moment later if this producer goes on to
block.
+ final ConditionAndFlag caf = putConditionAndFlags.get(threadIndex);
+ if (caf != null && caf.getWakeUp()) {
+ caf.setWakeUp(false);
return true;
}
return false;
@@ -415,6 +497,7 @@ public class FutureCompletingBlockingQueue<T> {
private static class ConditionAndFlag {
private final Condition cond;
private boolean wakeUp;
+ private int waitingPutters;
private ConditionAndFlag(Condition cond) {
this.cond = cond;
@@ -429,6 +512,19 @@ public class FutureCompletingBlockingQueue<T> {
return wakeUp;
}
+ private void startWaiting() {
+ waitingPutters++;
+ }
+
+ private void stopWaiting() {
+ checkState(waitingPutters > 0, "stopWaiting() without a matching
startWaiting()");
+ waitingPutters--;
+ }
+
+ private boolean hasWaitingPutter() {
+ return waitingPutters > 0;
+ }
+
private void setWakeUp(boolean value) {
wakeUp = value;
}
diff --git
a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java
b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java
index 9f7d132cec4..5d34693d703 100644
---
a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java
+++
b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.connector.base.source.reader.fetcher;
import org.apache.flink.api.connector.source.SourceSplit;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.reader.RecordsBySplits;
import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import org.apache.flink.connector.base.source.reader.SourceReaderOptions;
import
org.apache.flink.connector.base.source.reader.mocks.TestingRecordsWithSplitIds;
@@ -28,7 +29,9 @@ import
org.apache.flink.connector.base.source.reader.mocks.TestingSplitReader;
import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange;
import
org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
+import
org.apache.flink.connector.base.source.reader.synchronization.QueueProbe;
import org.apache.flink.core.testutils.OneShotLatch;
+import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.junit.Test;
@@ -104,6 +107,56 @@ public class SplitFetcherManagerTest {
assertThat(fetcherManager.getRunningFetcher()).isNull();
}
+ /**
+ * Each completed fetcher lifecycle must release its queue state, so
monotonically increasing
+ * fetcher ids do not accumulate historical state.
+ */
+ @Test
+ public void testFetcherShutdownReleasesWakeupStateAcrossLifecycles()
throws Exception {
+ final String splitId = "testSplit";
+ final SplitFetcherManager<Integer, TestingSourceSplit> fetcherManager =
+ new SingleThreadFetcherManager<>(
+ () ->
+ new AwaitingReader<>(
+ new IOException("Should not happen"),
+ new RecordsBySplits<>(
+ Collections.emptyMap(),
+
Collections.singleton(splitId))),
+ new Configuration());
+ final FutureCompletingBlockingQueue<RecordsWithSplitIds<Integer>>
queue =
+ fetcherManager.getQueue();
+
+ try {
+ for (int expectedFetcherId = 0; expectedFetcherId < 3;
expectedFetcherId++) {
+ fetcherManager.addSplits(
+ Collections.singletonList(new
TestingSourceSplit(splitId)));
+ assertThat(fetcherManager.fetchers).hasSize(1);
+ assertThat(fetcherManager.fetchers.keySet().iterator().next())
+ .isEqualTo(expectedFetcherId);
+
+ CommonTestUtils.waitUntilCondition(() -> queue.size() == 1);
+ // Give the fetcher wakeup state that its shutdown hook has to
release.
+ queue.wakeUpPuttingThread(expectedFetcherId);
+ assertThat(QueueProbe.liveProducerStates(queue)).isOne();
+
+ CommonTestUtils.waitUntilCondition(
+ () -> {
+ fetcherManager.maybeShutdownFinishedFetchers();
+ return fetcherManager.fetchers.isEmpty();
+ });
+ CommonTestUtils.waitUntilCondition(() ->
QueueProbe.liveProducerStates(queue) == 0);
+
+ queue.poll().recycle();
+ }
+ } finally {
+ RecordsWithSplitIds<Integer> batch;
+ while ((batch = queue.poll()) != null) {
+ batch.recycle();
+ }
+ fetcherManager.close(10_000L);
+ }
+ }
+
// the final modifier is important so that '@SafeVarargs' is accepted on
Java 8
@SuppressWarnings("FinalPrivateMethod")
@SafeVarargs
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 d49b6a412e3..ca97c28efac 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
@@ -32,6 +32,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Lock;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
@@ -145,6 +146,203 @@ public class FutureCompletingBlockingQueueTest {
.isTrue();
}
+ /**
+ * Without {@link FutureCompletingBlockingQueue#releaseProducer(int)} the
queue keeps one
+ * condition per producer index it has ever seen. Because {@code
SplitFetcherManager} allocates
+ * a fresh, never-recycled index per {@code SplitFetcher}, a source with
short-lived fetchers
+ * accumulates them for the lifetime of the JVM.
+ */
+ @Test
+ public void testReleaseProducerBoundsTheWakeupState() throws
InterruptedException {
+ final FutureCompletingBlockingQueue<Integer> queue = new
FutureCompletingBlockingQueue<>(1);
+
+ for (int producer = 0; producer < 3; producer++) {
+ // Fill the single slot, so the next put takes the full-queue path
that registers
+ // wakeup state for this producer.
+ assertThat(queue.put(producer, producer)).isTrue();
+ queue.wakeUpPuttingThread(producer);
+ assertThat(queue.put(producer, producer)).isFalse();
+ queue.poll();
+
+ assertThat(queue.numberOfProducerStates()).isOne();
+
+ queue.releaseProducer(producer);
+
+ assertThat(queue.numberOfProducerStates()).isZero();
+ }
+
+ assertThat(queue.getNumberOfQueuedPutters()).isZero();
+ }
+
+ /** Release is idempotent, tolerates unknown ids, and only removes the
target state. */
+ @Test
+ public void testReleaseProducerOnlyRemovesTheTargetState() {
+ final FutureCompletingBlockingQueue<Integer> queue = new
FutureCompletingBlockingQueue<>(1);
+
+ queue.wakeUpPuttingThread(3);
+ queue.wakeUpPuttingThread(4);
+
+ queue.releaseProducer(7);
+ queue.releaseProducer(3);
+ queue.releaseProducer(3);
+
+ assertThat(queue.hasProducerState(3)).isFalse();
+ assertThat(queue.hasProducerState(4)).isTrue();
+ assertThat(queue.numberOfProducerStates()).isOne();
+
+ queue.releaseProducer(4);
+ assertThat(queue.numberOfProducerStates()).isZero();
+ }
+
+ /**
+ * Releasing a producer that is currently parked in {@code waitOnPut}
would discard the wakeUp
+ * flag it is about to read and could leave it parked for good, so the
release must be refused
+ * while it is still waiting.
+ */
+ @Test
+ public void testReleaseProducerIsRefusedWhileTheProducerIsParked() throws
Exception {
+ final FutureCompletingBlockingQueue<Integer> queue = new
FutureCompletingBlockingQueue<>(1);
+
+ queue.put(0, 0);
+
+ final AtomicBoolean parkedPutterResult = new AtomicBoolean(true);
+ final Thread parkedPutter =
+ new Thread(() -> parkedPutterResult.set(putUnchecked(queue, 1,
1)), "parkedPutter");
+ parkedPutter.start();
+ try {
+ CommonTestUtils.waitUntilCondition(() ->
queue.getNumberOfQueuedPutters() == 1);
+
+ queue.releaseProducer(1);
+ assertThat(queue.numberOfProducerStates())
+ .as("must not drop wakeup state for a producer currently
inside put()")
+ .isOne();
+
+ // The graceful wakeup still reaches it, which is what the refusal
protects.
+ queue.wakeUpPuttingThread(1);
+ joinWithinTimeout(parkedPutter);
+ assertThat(parkedPutterResult).isFalse();
+
+ // Once it has left put(), the release goes through.
+ queue.releaseProducer(1);
+ assertThat(queue.numberOfProducerStates()).isZero();
+ } finally {
+ if (parkedPutter.isAlive()) {
+ queue.wakeUpPuttingThread(1);
+ parkedPutter.interrupt();
+ queue.poll();
+ parkedPutter.join(TimeUnit.SECONDS.toMillis(10));
+ }
+ queue.releaseProducer(1);
+ }
+ }
+
+ /** An interrupted wait must not leave the producer permanently marked as
waiting. */
+ @Test
+ public void testInterruptedPutterDoesNotPreventProducerRelease() throws
Exception {
+ final FutureCompletingBlockingQueue<Integer> queue = new
FutureCompletingBlockingQueue<>(1);
+ assertThat(queue.put(0, 0)).isTrue();
+
+ final CompletableFuture<Boolean> putInterrupted = new
CompletableFuture<>();
+ final Thread putter =
+ new Thread(
+ () -> {
+ try {
+ queue.put(1, 1);
+ putInterrupted.complete(false);
+ } catch (InterruptedException expected) {
+ putInterrupted.complete(true);
+ } catch (Throwable failure) {
+ putInterrupted.completeExceptionally(failure);
+ }
+ },
+ "interruptedPutter");
+ putter.start();
+ try {
+ CommonTestUtils.waitUntilCondition(() ->
queue.getNumberOfQueuedPutters() == 1);
+
+ putter.interrupt();
+ assertThat(putInterrupted.get(10, TimeUnit.SECONDS)).isTrue();
+ joinWithinTimeout(putter);
+ assertThat(queue.getNumberOfQueuedPutters()).isZero();
+ assertThat(queue.hasProducerState(1)).isTrue();
+
+ queue.releaseProducer(1);
+ assertThat(queue.hasProducerState(1)).isFalse();
+ } finally {
+ if (putter.isAlive()) {
+ putter.interrupt();
+ queue.poll();
+ putter.join(TimeUnit.SECONDS.toMillis(10));
+ }
+ queue.releaseProducer(1);
+ }
+ }
+
+ /**
+ * The companion to the test above, for the window that makes membership
of {@code notFull} an
+ * unreliable answer to "is this producer still inside {@code put()}?".
+ *
+ * <p>{@code signalNextPutter()} removes a condition from {@code notFull}
at signal time, not
+ * when its producer resumes, so between the signal and that producer
reacquiring the lock it is
+ * still inside {@code put()} while absent from {@code notFull}. A release
in that window would
+ * drop the state together with a wakeUp flag the producer has not read
yet, and the producer
+ * would then build fresh state with no flag and park again. The {@code
waitingPutters} counter
+ * spans the whole of {@code cond.await()} and therefore covers it.
+ *
+ * <p>The window is driven deterministically rather than raced for: the
test thread takes the
+ * queue's own lock, so the signalled producer cannot reacquire it and
cannot leave {@code
+ * put()} until the test releases it.
+ */
+ @Test
+ public void
testReleaseProducerIsRefusedAfterSignalBeforePutterReacquiresLock()
+ throws Exception {
+ final FutureCompletingBlockingQueue<Integer> queue = new
FutureCompletingBlockingQueue<>(1);
+ assertThat(queue.put(0, 0)).isTrue();
+
+ final AtomicBoolean putterResult = new AtomicBoolean(true);
+ final Thread putter =
+ new Thread(() -> putterResult.set(putUnchecked(queue, 1, 1)),
"signalledPutter");
+ putter.start();
+ try {
+ CommonTestUtils.waitUntilCondition(() ->
queue.getNumberOfQueuedPutters() == 1);
+
+ final Lock queueLock = queue.lock();
+ queueLock.lock();
+ try {
+ assertThat(queue.poll()).isZero();
+ assertThat(queue.getNumberOfQueuedPutters()).isZero();
+
+ queue.wakeUpPuttingThread(1);
+ queue.releaseProducer(1);
+ assertThat(queue.hasProducerState(1)).isTrue();
+
+ assertThat(queue.put(2, 2)).isTrue();
+ } finally {
+ queueLock.unlock();
+ }
+
+ joinWithinTimeout(putter);
+ assertThat(putterResult).isFalse();
+
+ queue.releaseProducer(1);
+ assertThat(queue.numberOfProducerStates()).isZero();
+ } finally {
+ if (putter.isAlive()) {
+ queue.wakeUpPuttingThread(1);
+ putter.interrupt();
+ queue.poll();
+ putter.join(TimeUnit.SECONDS.toMillis(10));
+ }
+ queue.releaseProducer(1);
+ queue.poll();
+ }
+ }
+
+ private static void joinWithinTimeout(Thread thread) throws
InterruptedException {
+ thread.join(TimeUnit.SECONDS.toMillis(10));
+ assertThat(thread.isAlive()).as("The putting thread should have
terminated").isFalse();
+ }
+
private static boolean putUnchecked(
FutureCompletingBlockingQueue<Integer> queue, int threadIndex, int
value) {
try {
diff --git
a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/synchronization/QueueProbe.java
b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/synchronization/QueueProbe.java
new file mode 100644
index 00000000000..e6b8461ec65
--- /dev/null
+++
b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/synchronization/QueueProbe.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.base.source.reader.synchronization;
+
+/** Test-only bridge exposing queue state to tests in other packages. */
+public final class QueueProbe {
+ private QueueProbe() {}
+
+ /** Returns the number of producers for which the queue currently holds
wakeup state. */
+ public static int liveProducerStates(FutureCompletingBlockingQueue<?>
queue) {
+ return queue.numberOfProducerStates();
+ }
+}
