[Feature] Flink Source throw the fetch timestamp %s is larger than the max timestamp %s.

[loserwang1024]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

Currently, if the table bucket is not updated for two days, the max timestamp also will be earlier value.If I specified the source start from current, the exception will be thrown.

Solution

just like what kafka source [1] and paimon do.

kafka:

// First get the current end offsets of the partitions. This is going to be used
        // in case we cannot find a suitable offsets based on the timestamp, i.e. the message
        // meeting the requirement of the timestamp have not been produced to Kafka yet, in
        // this case, we just use the latest offset.
        // We need to get the latest offsets before querying offsets by time to ensure that
        // no message is going to be missed.

paimon

/**
     * Returns the latest snapshot earlier than the timestamp mills. A non-existent snapshot may be
     * returned if all snapshots are equal to or later than the timestamp mills.
     */
    public @Nullable Long earlierThanTimeMills(long timestampMills, boolean startFromChangelog) {
    }

[1] https://github.com/apache/flink-connector-kafka/blob/f6a077a9dd8d1d5e43fc545cc9baab227d8438a0/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/initializer/TimestampOffsetsInitializer.java#L51
[2] https://github.com/apache/paimon/blob/ba4576657edf1dde60ac321423d4fa6032169119/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java#L306