@Override
    public ScanRuntimeProvider getScanRuntimeProvider(ScanContext context) {
        PulsarDeserializationSchema<RowData> deserializationSchema =
                deserializationSchemaFactory.createPulsarDeserialization(context);
        LOG.info("pulsar source init with properties: {}", properties);
        PulsarSource<RowData> source =
                PulsarSource.builder()
                        .setTopics(topics)
                        .setStartCursor(startCursor)
                        .setUnboundedStopCursor(stopCursor)
                        .setDeserializationSchema(deserializationSchema)
                        .setProperties(properties)
                        // only support exclusive since shared mode requires pulsar with transaction enabled
                        // and supporting transaction consumes more resources in pulsar broker
                        .setSubscriptionType(SubscriptionType.Exclusive)
                        .enableLogReport(enableLogReport)
                        .build();
        return SourceProvider.of(source);
    }
Usually it is the same as the valueDecodingFormat, but use a different naming to show that it
is used to list all the format metadata keys.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/table/PulsarTableSource.java