[flink] Avoid having QueryFileMonitor always busy - #9291
Conversation
The reader slept on the mailbox thread and reported MORE_AVAILABLE while idle, so the source task was always measured as busy. Complete the sleep asynchronously and report NOTHING_AVAILABLE instead, matching MonitorSource. Generated-by: Claude Code
| assertThat(status).isEqualTo(InputStatus.NOTHING_AVAILABLE); | ||
| assertThat(output.getEmittedRecords()).isEmpty(); | ||
| // the poll must not block the mailbox thread for the discovery interval | ||
| assertThat(elapsed).isLessThan(DISCOVERY_INTERVAL_MS / 3); |
There was a problem hiding this comment.
nit: this sleep can introduce flakyness under heavy CI contention, we may want to manually control this
There was a problem hiding this comment.
@ArnavBalyan Thanks — you're right, dropped that assertion.
The elapsed-time bound was the only wall-clock dependency in the test: NOTHING_AVAILABLE plus an isAvailable() future that isn't done yet already prove the poll didn't block for the discovery interval, so removing it loses no coverage. QueryFileMonitorTest still passes.
The test measured how long pollNext took and asserted it stayed under a third of the discovery interval. That bound depends on wall-clock timing: pollNext runs a table scan, so under heavy CI contention it can exceed the bound even when the reader behaves correctly. Returning NOTHING_AVAILABLE together with an isAvailable() future that is not yet done already proves the poll did not block for the discovery interval, so dropping the timing assertion loses no coverage. Generated-by: Claude Code
JingsongLi
left a comment
JingsongLi
left a comment
There was a problem hiding this comment.
The mailbox-facing availability change is correct, but the delay should not occupy the JVM-wide common pool.
| if (isEmpty) { | ||
| Thread.sleep(monitorInterval); | ||
| availableFuture = | ||
| CompletableFuture.runAsync( |
There was a problem hiding this comment.
[P2] Do not block the shared common pool for the discovery interval
CompletableFuture.runAsync uses ForkJoinPool.commonPool(), and this task occupies one of its workers in Thread.sleep for the full interval. When a TaskManager hosts more idle QueryFileMonitor readers than the common-pool parallelism, the remaining futures stay queued and do not even start their delay: later readers can wake after roughly 2×, 3×, ... the configured discovery interval, while unrelated common-pool work is starved as well. close() also leaves a long sleep running. Please use a scheduled/timer executor that does not block a worker, own/cancel it with the reader lifecycle, and cover multiple concurrent waits with a controllable scheduler.
3 participants
Purpose
fix #9290
QueryFileMonitor.Reader.pollNext()sleeps on the mailbox thread and returnsMORE_AVAILABLEwhen a scan finds nothing, and it never overridesisAvailable(), so Flink always sees a completed future. The source is measured as 100% busy and shown as a permanent bottleneck in the Web UI, and checkpoint triggering waits for the sleeping mailbox.Await the discovery interval asynchronously and return
NOTHING_AVAILABLEinstead, so idle polls are measured as idle. This is the change #6396 made toMonitorSource, applied to the remaining instance of that pattern.Tests
QueryFileMonitorTest#testPollWithoutNewFilesReportsNothingAvailableand#testPollWithNewFilesReportsMoreAvailable; no test covered this reader before. The first fails on master withexpected: NOTHING_AVAILABLE.mvn -pl paimon-flink/paimon-flink-common -Pflink1 clean install: 2303 tests, 0 failures, 0 errors. Both new tests also pass with-Pflink2.