[test] Fix unstable unaware bucket read test and assert split distribution

[wombatu-kun]

Purpose

AppendOnlyTableITCase.testReadUnawareBucketTableWithRebalanceShuffle flakes on unrelated PRs (latest #8839, on UTCase and ITCase Flink 2.x on JDK 11) and never checked what its name says.

It asserted that the first 2 of 8 rows were {1, 2}. Each written file holds (1,'AAA') then (2,'BBB'), so every reader emits its 1 before its 2, and [1, 2] only appears when the rebalance happens not to interleave; [1, 1] is the normal parallel outcome. Before #3955 all splits of a single bucket table went to channel 0, one subtask read them serially, and the first two rows were still 1 and 2. So the assertion passes with the bug and flakes without it.

Tests

Asserting the distribution needs per subtask metrics, hence a JobID and a MiniCluster with a metric reporter. AppendOnlyTableITCase has neither, hence the move.

The case is now UnawareBucketSplitShuffleITCase, a MiniCluster with an InMemoryReporter as in SourceMetricsITCase, keeping the DDL, the SQL and the data. It asserts all 8 rows order independent, plus numRecordsIn > 0 on each of the 4 reader subtasks, which is the split distribution the old test claimed to cover.

Checked to be non vacuous: forcing MonitorSource.buildSource onto shuffleOrdered makes it fail with [0, 0, 0, 8] while the row assertion still passes. Run 20 times on flink2 / JDK 11 and on flink1 / JDK 8, together with the full AppendOnlyTableITCase.

[JingsongLi]

+1