@Override
    public SourceReader<T, IcebergSourceSplit> createReader(SourceReaderContext readerContext) {
        InlongIcebergSourceReaderMetrics<T> metrics =
                new InlongIcebergSourceReaderMetrics<>(readerContext.metricGroup(), lazyTable().name());
        metrics.registerMetrics(metricOption);
        return new IcebergSourceReader<>(metrics, readerFunction, readerContext, enableLogReport);
    }
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