Hi there. Recently we were faced by two issues we identified in the HybridSource (FLIP-150), they are: 1. We found there is a possibility of lost splits if there is a snapshot and crash during source switch. a. At this part, when snapshot state, if currentReader is null, it will return empty list directly. b. So when there is a recovery, where the job recovers from a checkpoint (then the splits from checkpoint was put to restoredSplits field) c. Then there is a savepoint/checkpoint happening, however, the currentReader may still be null because the reader just requested a new enumerator. d. So the empty list is snapshot as state, but then the job crashes again immediately. e. Upon the next recovery, the job no longer see the last restored splits, they were cleared due to last checkpoint. f. (although this seems to be a very rare situation but we seems indeed has encountered once). 2. The second is the same as FLINK-30514, the

start()

method was called before

restoredSplits

was set to the underlying readers. a. This is critical to us because we found this behavior is in-compatible with the Iceberg source reader because it request split in the

start

method b. In the iceberg source, on start it checks if there is recovered splits, but because of the order of calling

start

in the HybridSource, it never finds the recovered splits. c. this will become a issue because the reader will double request splits on recovery, this messed up watermark if we enabled watermark for the iceberg reader (multiple splits was assigned to one reader at the sametime, the second will block watermarks from emitting).