@Override
    public void start() {
        if (this.enableLogReport) {
            this.openTelemetryLogger.install();
        }
        // We request a split only if we did not get splits during the checkpoint restore.
        // Otherwise, reader restarts will keep requesting more and more splits.
        if (getNumberOfCurrentlyAssignedSplits() == 0) {
            requestSplit(Collections.emptyList());
        }
    }
Copy from iceberg-flink:iceberg-flink-1.15:1.3.1
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/source/reader/IcebergSourceReader.java