private PulsarDeserializationSchema<RowData> createPulsarDeserialization(
            DeserializationSchema<RowData> keyDeserialization, DeserializationSchema<RowData> valueDeserialization,
            TypeInformation<RowData> producedTypeInfo) {
        final DynamicPulsarDeserializationSchema.MetadataConverter[] metadataConverters = metadataKeys.stream()
                .map(k -> Stream.of(ReadableMetadata.values())
                        .filter(rm -> rm.key.equals(k))
                        .findFirst()
                        .orElseThrow(IllegalStateException::new))
                .map(m -> m.converter)
                .toArray(DynamicPulsarDeserializationSchema.MetadataConverter[]::new);

        // check if connector metadata is used at all
        final boolean hasMetadata = metadataKeys.size() > 0;

        // adjust physical arity with value format's metadata
        final int adjustedPhysicalArity = producedDataType.getChildren().size() - metadataKeys.size();

        // adjust value format projection to include value format's metadata columns at the end
        final int[] adjustedValueProjection = IntStream.concat(
                IntStream.of(valueProjection),
                IntStream.range(keyProjection.length + valueProjection.length, adjustedPhysicalArity))
                .toArray();

        return new DynamicPulsarDeserializationSchema(
                adjustedPhysicalArity,
                keyDeserialization,
                keyProjection,
                valueDeserialization,
                adjustedValueProjection,
                hasMetadata,
                metadataConverters,
                producedTypeInfo,
                upsertMode);
    }
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