@Override
    public void applyReadableMetadata(List<String> allMetadataKeys, DataType producedDataType) {
        // separate connector and format metadata
        final List<String> formatMetadataKeys =
                allMetadataKeys.stream()
                        .filter(k -> k.startsWith(FORMAT_METADATA_PREFIX))
                        .collect(Collectors.toList());

        final List<String> connectorMetadataKeys = new ArrayList<>(allMetadataKeys);
        connectorMetadataKeys.removeAll(formatMetadataKeys);

        // push down format metadata
        final Map<String, DataType> formatMetadata =
                decodingFormatForReadingMetadata.listReadableMetadata();
        if (formatMetadata.size() > 0) {
            final List<String> requestedFormatMetadataKeys =
                    formatMetadataKeys.stream()
                            .map(k -> k.substring(FORMAT_METADATA_PREFIX.length()))
                            .collect(Collectors.toList());
            decodingFormatForReadingMetadata.applyReadableMetadata(requestedFormatMetadataKeys);
        }

        // update the factory attributes.
        deserializationSchemaFactory.setConnectorMetadataKeys(connectorMetadataKeys);
        deserializationSchemaFactory.setProducedDataType(producedDataType);
    }
According to convention, the order of the final row must be PHYSICAL + FORMAT METADATA +
CONNECTOR METADATA where the format metadata has the highest precedence.

@return
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/table/PulsarTableSource.java