@SuppressWarnings("java:S3776")
    public PulsarSource<OUT> build() {

        // Ensure the topic subscriber for pulsar.
        checkNotNull(subscriber, "No topic names or topic pattern are provided.");

        SubscriptionType subscriptionType = configBuilder.get(PULSAR_SUBSCRIPTION_TYPE);
        if (subscriptionType == SubscriptionType.Key_Shared) {
            if (rangeGenerator == null) {
                LOG.warn(
                        "No range generator provided for key_shared subscription,"
                                + " we would use the SplitRangeGenerator as the default range generator.");
                this.rangeGenerator = new SplitRangeGenerator();
            }
        } else {
            // Override the range generator.
            this.rangeGenerator = new FullRangeGenerator();
        }

        if (boundedness == null) {
            LOG.warn("No boundedness was set, mark it as a endless stream.");
            this.boundedness = Boundedness.CONTINUOUS_UNBOUNDED;
        }
        if (boundedness == Boundedness.BOUNDED
                && configBuilder.get(PULSAR_PARTITION_DISCOVERY_INTERVAL_MS) >= 0) {
            LOG.warn(
                    "{} property is overridden to -1 because the source is bounded.",
                    PULSAR_PARTITION_DISCOVERY_INTERVAL_MS);
            configBuilder.override(PULSAR_PARTITION_DISCOVERY_INTERVAL_MS, -1L);
        }

        checkNotNull(deserializationSchema, "deserializationSchema should be set.");

        // Enable transaction if the cursor auto commit is disabled for Key_Shared & Shared.
        if (FALSE.equals(configBuilder.get(PULSAR_ENABLE_AUTO_ACKNOWLEDGE_MESSAGE))
                && (subscriptionType == SubscriptionType.Key_Shared
                        || subscriptionType == SubscriptionType.Shared)) {
            LOG.info(
                    "Pulsar cursor auto commit is disabled, make sure checkpoint is enabled "
                            + "and your pulsar cluster is support the transaction.");
            configBuilder.override(PULSAR_ENABLE_TRANSACTION, true);

            if (!configBuilder.contains(PULSAR_READ_TRANSACTION_TIMEOUT)) {
                LOG.warn(
                        "The default pulsar transaction timeout is 3 hours, "
                                + "make sure it was greater than your checkpoint interval.");
            } else {
                Long timeout = configBuilder.get(PULSAR_READ_TRANSACTION_TIMEOUT);
                LOG.warn(
                        "The configured transaction timeout is {} mille seconds, "
                                + "make sure it was greater than your checkpoint interval.",
                        timeout);
            }
        }

        if (!configBuilder.contains(PULSAR_CONSUMER_NAME)) {
            LOG.warn(
                    "We recommend set a readable consumer name through setConsumerName(String) in production mode.");
        } else {
            String consumerName = configBuilder.get(PULSAR_CONSUMER_NAME);
            if (!consumerName.contains("%s")) {
                configBuilder.override(PULSAR_CONSUMER_NAME, consumerName + " - %s");
            }
        }

        // Since these implementations could be a lambda, make sure they are serializable.
        checkState(isSerializable(startCursor), "StartCursor isn't serializable");
        checkState(isSerializable(stopCursor), "StopCursor isn't serializable");
        checkState(isSerializable(rangeGenerator), "RangeGenerator isn't serializable");

        // Check builder configuration.
        SourceConfiguration sourceConfiguration =
                configBuilder.build(SOURCE_CONFIG_VALIDATOR, SourceConfiguration::new);

        return new PulsarSource<>(
                sourceConfiguration,
                subscriber,
                rangeGenerator,
                startCursor,
                stopCursor,
                boundedness,
                deserializationSchema,
                enableLogReport);
    }
Build the {@link PulsarSource}.

@return a PulsarSource with the settings made for this builder.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/source/PulsarSourceBuilder.java