This is an automated email from the ASF dual-hosted git repository. MartijnVisser pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/flink.git
commit cfa3094deddbac36d0fec1010920938a3908cc7d Author: Martijn Visser <[email protected]> AuthorDate: Tue Jun 16 15:14:42 2026 -0700 [FLINK-37663][connector-base] Remove stale put conditions to fix lost wakeup Remove a blocked putter's condition from the notFull queue when its await() returns, so signalNextPutter() no longer signals a putter that has stopped waiting after a graceful wakeUpPuttingThread() or a spurious wakeup. Generated-by: Claude Code (Claude Opus 4.8) --- .../reader/synchronization/FutureCompletingBlockingQueue.java | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) 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 aaf90f4b5ee..799fd0c4489 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 @@ -373,7 +373,13 @@ public class FutureCompletingBlockingQueue<T> { maybeCreateCondition(fetcherIndex); Condition cond = putConditionAndFlags[fetcherIndex].condition(); notFull.add(cond); - cond.await(); + try { + 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); + } } @GuardedBy("lock")
