[FLINK-37663] [connector-base] Address potential sync issue in FutureCompletingBlockingQueue wakeup

[herbherbherb]

What is the purpose of the change

This pull request addresses a potential thread synchronization issue identified in FutureCompletingBlockingQueue (see FLINK-37663). The concern is that the wakeUpPuttingThread method signals a putting thread's condition variable but might not remove it from the internal notFull condition queue.

This could lead to an inconsistent state where subsequent signals (triggered when dequeue finds space) are sent to the condition of an already woken or potentially closed thread, effectively losing the signal for a genuinely waiting thread. This scenario could potentially cause source fetchers to stall indefinitely. This change aims to prevent this potential lost signal by ensuring the condition queue remains consistent.

Brief change log

• Modified FutureCompletingBlockingQueue#wakeUpPuttingThread(int fetcherIndex) to explicitly remove the corresponding condition variable from the notFull queue after retrieving it and before signaling it.
• Added FutureCompletingBlockingQueueTest#testManualWakeupRemovesCondition (or a similar name) to specifically verify that a manually woken thread's condition is removed from the notFull queue and does not interfere with subsequent signals intended for other waiting threads.

Verifying this change

This change added tests and can be verified as follows:

• Added FutureCompletingBlockingQueueTest#testManualWakeupRemovesCondition (or your actual test name) which simulates the scenario described in FLINK-37663:
  1. A thread blocks on put().
  2. It gets manually woken up via wakeUpPuttingThread().
  3. Another thread blocks on put().
  4. An element is polled from the queue using poll().
  5. The test verifies that the second thread (which was genuinely waiting) is woken up by the signal triggered during poll(), confirming the first thread's condition was removed upon manual wakeup.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no (FutureCompletingBlockingQueue is @Internal)
• The serializers: no
• The runtime per-record code paths (performance sensitive): don't know (This queue is used in source connectors' data path; the change adds a queue removal operation which has minor performance overhead?)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: don't know (If source connectors stall due to this bug, it can affect checkpointing progress and overall job health/recovery).
• The S3 file system connector: don't know (It affects all connectors using the base source framework, potentially including S3 if it uses this mechanism, but the change is in connector-base).

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

[flinkbot]

CI report:

• 01a0d0e Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[github-actions]

This PR is being marked as stale since it has not had any activity in the last 90 days.
If you would like to keep this PR alive, please leave a comment asking for a review.
If the PR has merge conflicts, update it with the latest from the base branch.

If you are having difficulty finding a reviewer, please reach out to the
community, contact details can be found here: https://flink.apache.org/what-is-flink/community/

If this PR is no longer valid or desired, please feel free to close it.
If no activity occurs in the next 30 days, it will be automatically closed.

[github-actions]

This PR has been closed since it has not had any activity in 120 days.
If you feel like this was a mistake, or you would like to continue working on it,
please feel free to re-open the PR and ask for a review.