Hello Ran Tao, Thanks for this FLIP.
I have a comment about the handover of switching context between sources. You have proposed to define interfaces named around timestamps, SupportsGetEndTimestamp and SupportsSwitchedStartTimestamp. These work well with KafkaSource as the downstream child source but looks very Kafka specific. Can we make this handover more generalized? Something like following ``` interface SupportsGetUpperBound { Map<String, String> getUpperBound(); } ``` So, an upstream source can provide a map of key-value which the downstream source can parse and use appropriately.