[flink] Flaky testPendingRecordsMetric due to uncertainty bucket fetch ordering #4966

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

1.0.0 (latest release)

Please describe the bug 🐞

FlinkSourceSplitReaderTest.testPendingRecordsMetric can fail intermittently in CI because it assumes that two streaming buckets will both have partially consumed records during fetching.

The test uses aggregationVerified to ensure that both buckets have non-zero pending records:

if (firstStreamingFetched < recordsPerBucket
        && secondStreamingFetched < recordsPerBucket) {
    aggregationVerified = true;
}

However, the fetch order across buckets is not guaranteed. One bucket may be fully consumed before the other starts returning records, so this condition may never be satisfied even when the pending records metric is correct.

refer to: https://github.com/apache/fluss/actions/runs/41158209879/job/124409618777?pr=4795

FlinkSourceSplitReaderTest.testPendingRecordsMetric:354
Expecting value to be true but was false

Solution

Fix test:

• Use two split readers sharing the same FlinkSourceReaderMetrics instance.
• Control consumption so that both readers have non-zero pending records before verifying metric aggregation.
• Preserve the existing assertions for bounded split completion, removal of unread records from the metric, and scanner closure.
• Ensure that aggregation of pending records across multiple buckets within a single scanner remains covered by a separate test.

Are you willing to submit a PR?

• I'm willing to submit a PR!