[FIP-28] Support consuming historical partition changelog in Flink

[luoyuxia]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

Historical partition support allows writes targeting expired business partitions to be routed through the internal __historical__ partition when table.datalake.historical-partition.enabled = true. The Flink streaming source currently does not subscribe to this partition's changelog.

Add support for consuming these historical-partition changes in Flink. This would allow downstream streaming pipelines to process late-arriving records and corrections to historical data after the original business partition has expired from active Fluss storage.

For example, after an order partition has been tiered to Paimon and retired from active Fluss storage, an application may update an order in that partition through historical writes. A Flink streaming job should be able to consume that change and update its downstream results.

Solution

Support Flink streaming consumption of the changelog from the internal historical partition for tables with historical partition access enabled. The same Flink source should consume changes from both active and historical partitions, while preserving the connector’s existing changelog, partition-filtering, and checkpoint/recovery semantics.

The current partition discovery path calls listPartitionInfos(tablePath), which excludes system partitions by default. Main already provides an includeSystemPartitions API that can be used as a starting point, with explicit handling for the historical partition during filtering and split initialization.

Add coverage for historical writes after streaming startup, partition predicates, checkpoint recovery, and partition retirement.

Anything else?

Part of #3631 (FIP-28). Related historical write support: #4119.

Source reference: FlinkSourceEnumerator.listPartitions().

Willingness to contribute

• I'm willing to submit a PR!