@Override
    public ScanRuntimeProvider getScanRuntimeProvider(ScanContext context) {
        PulsarDeserializationSchema<RowData> deserializationSchema =
                deserializationSchemaFactory.createPulsarDeserialization(context);
        PulsarSourceBuilder<RowData> sourceBuilder = PulsarSource.builder();
        sourceBuilder
                .setTopics(topics)
                .setStartCursor(startCursor)
                .setDeserializationSchema(deserializationSchema)
                .setProperties(properties);
        if (!(stopCursor instanceof NeverStopCursor)) {
            sourceBuilder.setBoundedStopCursor(stopCursor);
        } else {
            sourceBuilder.setUnboundedStopCursor(stopCursor);
        }
        return SourceProvider.of(sourceBuilder.build());
    }
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.18/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/table/source/PulsarTableSource.java