@Override
    public ScanRuntimeProvider getScanRuntimeProvider(ScanContext context) {

        final DeserializationSchema<RowData> keyDeserialization =
                createDeserialization(context, keyDecodingFormat, keyProjection, keyPrefix);

        final DeserializationSchema<RowData> valueDeserialization =
                createDeserialization(context, valueDecodingFormat, valueProjection, "");
        final TypeInformation<RowData> producedTypeInfo =
                context.createTypeInformation(producedDataType);
        PulsarDeserializationSchema<RowData> deserializationSchema = createPulsarDeserialization(keyDeserialization,
                valueDeserialization,
                producedTypeInfo);
        final ClientConfigurationData clientConfigurationData = PulsarClientUtils.newClientConf(serviceUrl, properties);
        SourceFunction<RowData> source = adminUrl != null
                ? createPulsarSource(clientConfigurationData, deserializationSchema)
                : createPulsarSourceWithoutAdmin(clientConfigurationData, deserializationSchema);
        return SourceFunctionProvider.of(source, false);
    }
Flag to determine source mode. In upsert mode, it will keep the tombstone message. *
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/table/PulsarDynamicTableSource.java