@Override
    public void handleSplitsChanges(SplitsChange<IcebergSourceSplit> splitsChange) {
        if (!(splitsChange instanceof SplitsAddition)) {
            throw new UnsupportedOperationException(
                    String.format("Unsupported split change: %s", splitsChange.getClass()));
        }

        LOG.info("Add {} splits to reader", splitsChange.splits().size());
        splits.addAll(splitsChange.splits());
        metrics.incrementAssignedSplits(splitsChange.splits().size());
        metrics.incrementAssignedBytes(calculateBytes(splitsChange));
    }
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/IcebergSourceSplitReader.java