private List<IcebergSourceSplit> planSplitsForBatch(String threadName) {
        ExecutorService workerPool =
                ThreadPools.newWorkerPool(threadName, scanContext.planParallelism());
        try {
            List<IcebergSourceSplit> splits =
                    FlinkSplitPlanner.planIcebergSourceSplits(lazyTable(), scanContext, workerPool);
            LOG.info(
                    "Discovered {} splits from table {} during job initialization",
                    splits.size(),
                    lazyTable().name());
            return splits;
        } finally {
            workerPool.shutdown();
        }
    }
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/IcebergSource.java