private MetadataConverter[] getMetadataConverters() {
        if (CollectionUtils.isEmpty(metadataKeys)) {
            return new MetadataConverter[0];
        }
        return metadataKeys.stream()
                .map(
                        key -> Stream.of(IcebergReadableMetadata.values())
                                .filter(m -> m.getKey().equals(key))
                                .findFirst()
                                .orElseThrow(IllegalStateException::new))
                .map(IcebergReadableMetadata::getConverter)
                .toArray(MetadataConverter[]::new);
    }
Flink Iceberg table source.
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/IcebergTableSource.java