private SourceFunction<RowData> createPulsarSource(
            ClientConfigurationData clientConfigurationData,
            PulsarDeserializationSchema<RowData> deserializationSchema) {
        FlinkPulsarSource source =
                new FlinkPulsarSource(
                        adminUrl,
                        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;
            case SPECIFIC_OFFSETS:
                source.setStartFromSpecificOffsets(startupOptions.specificOffsets);
                break;
            case EXTERNAL_SUBSCRIPTION:
                MessageId subscriptionPosition = MessageId.latest;
                if (CONNECTOR_STARTUP_MODE_VALUE_EARLIEST.equals(startupOptions.externalSubStartOffset)) {
                    subscriptionPosition = MessageId.earliest;
                }
                source.setStartFromSubscription(startupOptions.externalSubscriptionName, subscriptionPosition);
        }
        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