private SourceFunction<RowData> createPulsarSourceWithoutAdmin(
            ClientConfigurationData clientConfigurationData,
            PulsarDeserializationSchema<RowData> deserializationSchema) {
        FlinkPulsarSourceWithoutAdmin<RowData> source = new FlinkPulsarSourceWithoutAdmin<>(
                serviceUrl,
                clientConfigurationData,
                deserializationSchema,
                properties,
                inlongMetric,
                auditHostAndPorts,
                auditKeys);

        if (watermarkStrategy != null) {
            source.assignTimestampsAndWatermarks(watermarkStrategy);
        }

        switch (startupOptions.startupMode) {
            case EARLIEST:
                source.setStartFromEarliest();
                break;
            case LATEST:
                source.setStartFromLatest();
                break;
            default:
                throw new IllegalArgumentException(
                        "Unknown startup mode option for tdmq pulsar: " + startupOptions.startupMode);
        }
        return source;
    }
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