[GitHub] [flink] AHeise commented on a change in pull request #15304: [FLINK-20731][connector] Introduce Pulsar Source

[GitBox <gi...@apache.org>]

AHeise commented on a change in pull request #15304:
URL: https://github.com/apache/flink/pull/15304#discussion_r683984197



##########
File path: flink-connectors/flink-connector-elasticsearch6/pom.xml
##########
@@ -91,7 +91,7 @@ under the License.
 		<dependency>
 			<groupId>org.testcontainers</groupId>
 			<artifactId>elasticsearch</artifactId>
-			<version>1.15.1</version>
+			<version>${testcontainers.version}</version>

Review comment:
       Good idea but these changes seem unrelated to Pulsar source and seem to be some preparation. Please factor them out into one commit (for example with commit message "[FLINK-20731][connector] Aligning testcontainers version").

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigUtils.java
##########
@@ -0,0 +1,277 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.common.config;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
+import org.apache.flink.connector.pulsar.source.PulsarSourceOptions;
+
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.api.AuthenticationFactory;
+import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+import java.util.function.Function;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAMS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAM_MAP;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PLUGIN_CLASS_NAME;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTO_CERT_REFRESH_TIME;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_CONCURRENT_LOOKUP_REQUEST;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_CONNECTIONS_PER_BROKER;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_CONNECTION_TIMEOUT_MS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_CONNECT_TIMEOUT;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_BUSY_WAIT;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_INITIAL_BACKOFF_INTERVAL_NANOS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_KEEP_ALIVE_INTERVAL_SECONDS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_LISTENER_NAME;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MAX_BACKOFF_INTERVAL_NANOS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MAX_LOOKUP_REDIRECTS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MAX_LOOKUP_REQUEST;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MAX_NUMBER_OF_REJECTED_REQUEST_PER_CONNECTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MEMORY_LIMIT_BYTES;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_NUM_IO_THREADS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_NUM_LISTENER_THREADS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_OPERATION_TIMEOUT_MS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_PROXY_PROTOCOL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_PROXY_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_READ_TIMEOUT;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_REQUEST_TIMEOUT;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_REQUEST_TIMEOUT_MS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SSL_PROVIDER;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_STATS_INTERVAL_SECONDS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_ALLOW_INSECURE_CONNECTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_CIPHERS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_HOSTNAME_VERIFICATION_ENABLE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_PROTOCOLS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_TRUST_CERTS_FILE_PATH;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_TRUST_STORE_PASSWORD;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_TRUST_STORE_PATH;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_TRUST_STORE_TYPE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_USE_KEY_STORE_TLS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_USE_TCP_NO_DELAY;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_USE_TLS;
+import static org.apache.flink.connector.pulsar.common.exception.PulsarExceptionUtils.sneaky;
+import static org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils.configMap;
+
+/** The util for creating pulsar configuration class from flink's {@link Configuration}. */
+public final class PulsarConfigUtils {
+
+    private PulsarConfigUtils() {
+        // No need to create instance.
+    }
+
+    /**
+     * Create a common {@link ClientConfigurationData} for both pulsar source and sink connector by
+     * using the given {@link Configuration}. Add the configuration was listed in {@link
+     * PulsarSourceOptions}, and we would parse it one by one.
+     *
+     * @param configuration The flink configuration
+     */
+    public static ClientConfigurationData createClientConfig(

Review comment:
       Is it okay for most of the options to be set to null (e.g. is there no null-check on the ClientConfigurationData)?
   We could alternatively use `configuration.getOptional(...).ifPresent(...)`.

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarTransactionUtils.java
##########
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.common.utils;
+
+import org.apache.pulsar.client.api.transaction.Transaction;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.client.impl.transaction.TransactionImpl;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+
+import static org.apache.flink.connector.pulsar.common.exception.PulsarExceptionUtils.sneaky;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Transaction was introduced into pulsar since 2.7.0, but the interface {@link Transaction} didn't
+ * provide a id method until 2.8.1. We have to add this util for acquiring the {@link TxnID} for
+ * compatible consideration.
+ */
+@SuppressWarnings("java:S3011")
+public final class PulsarTransactionUtils {

Review comment:
       Do we need this because Pulsar 2.8.1 is not yet release? Can we bump version afterwards and remove this hack?

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSource.java
##########
@@ -0,0 +1,265 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.java.ClosureCleaner;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumState;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumStateSerializer;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumerator;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.PulsarSourceReaderFactory;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarSchemaInitializationContext;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitSerializer;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.createAdmin;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.createClient;
+import static org.apache.flink.connector.pulsar.common.exception.PulsarExceptionUtils.sneaky;
+
+/**
+ * The Source implementation of Pulsar. Please use a {@link PulsarSourceBuilder} to construct a
+ * {@link PulsarSource}. The following example shows how to create a PulsarSource emitting records
+ * of <code>String</code> type.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource.<byte[], String>builder()

Review comment:
       We can replace `<byte[], String>builder()`  with `builder()` when extracting the type information at `setDeserializationSchema`. I'll annotate that again in the Builder.

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
##########
@@ -0,0 +1,575 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.common.exception.PulsarRuntimeException;
+import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.DivideRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FullRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.RegexSubscriptionMode;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.function.Consumer;
+import java.util.regex.Pattern;
+
+import static java.lang.Boolean.FALSE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.getOptionValue;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ENABLE_AUTO_COMMIT_CURSOR;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_REGEX_SUBSCRIPTION_MODE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPICS_PATTERN;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPIC_NAMES;
+import static org.apache.flink.connector.pulsar.source.config.PulsarSourceConfigUtils.checkConfigurations;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The builder class for {@link PulsarSource} to make it easier for the users to construct a {@link
+ * PulsarSource}.
+ *
+ * <p>The following example shows the minimum setup to create a PulsarSource that reads the String
+ * values from a Pulsar topic.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .build();
+ * }</pre>
+ *
+ * <p>The service url, admin url, subscription name, topics to consume, and the record deserializer
+ * are required fields that must be set.
+ *
+ * <p>To specify the starting position of PulsarSource, one can call {@link
+ * #setStartCursorSupplier(SerializableSupplier)}.
+ *
+ * <p>By default the PulsarSource runs in an {@link Boundedness#CONTINUOUS_UNBOUNDED} mode and never
+ * stop until the Flink job is canceled or fails. To let the PulsarSource run in {@link
+ * Boundedness#CONTINUOUS_UNBOUNDED} but stops at some given offsets, one can call {@link
+ * #setUnbounded(SerializableSupplier)}. For example the following PulsarSource stops after it
+ * consumes up to a event time when the Flink started.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .setUnbounded(StopCursor.atEventTime(System.currentTimeMillis()))
+ *     .build();
+ * }</pre>
+ *
+ * @param <IN> The input type of the pulsar {@link Message <?>}
+ * @param <OUT> The output type of the source.
+ */
+@PublicEvolving
+public final class PulsarSourceBuilder<IN, OUT> {
+    private static final Logger LOG = LoggerFactory.getLogger(PulsarSourceBuilder.class);
+
+    private final Configuration configuration;
+    private PulsarSubscriber subscriber;
+    private RangeGenerator rangeGenerator;
+    private SerializableSupplier<StartCursor> startCursorSupplier;
+    private SerializableSupplier<StopCursor> stopCursorSupplier;
+    private Boundedness boundedness;
+    private PulsarDeserializationSchema<IN, OUT> deserializationSchema;
+    private ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer;
+    private ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+            consumerConfigurationCustomizer;
+
+    // private builder constructor.
+    PulsarSourceBuilder() {
+        // The default configuration holder.
+        this.configuration = new Configuration();
+    }
+
+    /**
+     * Sets the admin endpoint for the PulsarAdmin of the PulsarSource.
+     *
+     * @param adminUrl the url for the PulsarAdmin.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setAdminUrl(String adminUrl) {
+        configuration.set(PULSAR_ADMIN_URL, adminUrl);
+        return this;
+    }
+
+    /**
+     * Sets the server's link for the PulsarConsumer of the PulsarSource.
+     *
+     * @param serviceUrl the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setServiceUrl(String serviceUrl) {
+        configuration.set(PULSAR_SERVICE_URL, serviceUrl);
+        return this;
+    }
+
+    /**
+     * Sets the name for this pulsar subscription.
+     *
+     * @param subscriptionName the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionName(String subscriptionName) {
+        configuration.set(PULSAR_SUBSCRIPTION_NAME, subscriptionName);
+        return this;
+    }
+
+    /**
+     * {@link SubscriptionType} is the consuming behavior for pulsar, we would generator different
+     * split by the given subscription type, it's required and quite important for end-user.
+     *
+     * @param subscriptionType The type of subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionType(SubscriptionType subscriptionType) {
+        configuration.set(PULSAR_SUBSCRIPTION_TYPE, subscriptionType);
+        return this;
+    }
+
+    /**
+     * Set a pulsar topic list for flink source. Some topic may not exist currently, consuming this
+     * non-existed topic wouldn't throw any exception. But the best solution is just consuming by
+     * using a topic regex. This method is conflict with {@code setTopicPattern}.
+     *
+     * @param topics The topic list you would like to consume message.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopics(String... topics) {
+        return setTopics(Arrays.asList(topics));
+    }
+
+    /**
+     * Set a pulsar topic list for flink source. Some topic may not exist currently, consuming this
+     * non-existed topic wouldn't throw any exception. But the best solution is just consuming by
+     * using a topic regex. This method is conflict with {@code setTopicPattern}.
+     *
+     * @param topics The topic list you would like to consume message.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopics(List<String> topics) {
+        ensureSubscriberIsNull("topics");
+        configuration.set(PULSAR_TOPIC_NAMES, PulsarJsonUtils.toString(topics));
+
+        this.subscriber = PulsarSubscriber.getTopicListSubscriber(topics);
+
+        return this;
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(String topicsPattern) {
+        return setTopicPattern(Pattern.compile(topicsPattern));
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(Pattern topicsPattern) {
+        return setTopicPattern(topicsPattern, RegexSubscriptionMode.AllTopics);
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @param regexSubscriptionMode The topic filter for regex subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(
+            String topicsPattern, RegexSubscriptionMode regexSubscriptionMode) {
+        return setTopicPattern(Pattern.compile(topicsPattern), regexSubscriptionMode);
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @param regexSubscriptionMode The topic filter for regex subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(
+            Pattern topicsPattern, RegexSubscriptionMode regexSubscriptionMode) {
+        ensureSubscriberIsNull("topic pattern");
+        configuration.set(PULSAR_TOPICS_PATTERN, topicsPattern.toString());
+        this.subscriber =
+                PulsarSubscriber.getTopicPatternSubscriber(topicsPattern, regexSubscriptionMode);
+
+        return setRegexSubscriptionMode(regexSubscriptionMode);
+    }
+
+    /**
+     * When subscribing to a topic using a regular expression, you can pick a certain type of
+     * topics.
+     *
+     * @param regexSubscriptionMode The topic type.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setRegexSubscriptionMode(
+            RegexSubscriptionMode regexSubscriptionMode) {
+        configuration.set(PULSAR_REGEX_SUBSCRIPTION_MODE, regexSubscriptionMode);
+        return this;
+    }
+
+    /**
+     * Set a topic range generator for Key_Shared subscription.

Review comment:
       Add that setSubscriptionType(Key_Shared) must be set before. 
   Alternatively, you can set the type in this method and LOG a warning if there was a type already set.

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
##########
@@ -0,0 +1,575 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.common.exception.PulsarRuntimeException;
+import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.DivideRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FullRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.RegexSubscriptionMode;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.function.Consumer;
+import java.util.regex.Pattern;
+
+import static java.lang.Boolean.FALSE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.getOptionValue;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ENABLE_AUTO_COMMIT_CURSOR;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_REGEX_SUBSCRIPTION_MODE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPICS_PATTERN;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPIC_NAMES;
+import static org.apache.flink.connector.pulsar.source.config.PulsarSourceConfigUtils.checkConfigurations;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The builder class for {@link PulsarSource} to make it easier for the users to construct a {@link
+ * PulsarSource}.
+ *
+ * <p>The following example shows the minimum setup to create a PulsarSource that reads the String
+ * values from a Pulsar topic.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .build();
+ * }</pre>
+ *
+ * <p>The service url, admin url, subscription name, topics to consume, and the record deserializer
+ * are required fields that must be set.
+ *
+ * <p>To specify the starting position of PulsarSource, one can call {@link
+ * #setStartCursorSupplier(SerializableSupplier)}.
+ *
+ * <p>By default the PulsarSource runs in an {@link Boundedness#CONTINUOUS_UNBOUNDED} mode and never
+ * stop until the Flink job is canceled or fails. To let the PulsarSource run in {@link
+ * Boundedness#CONTINUOUS_UNBOUNDED} but stops at some given offsets, one can call {@link
+ * #setUnbounded(SerializableSupplier)}. For example the following PulsarSource stops after it
+ * consumes up to a event time when the Flink started.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .setUnbounded(StopCursor.atEventTime(System.currentTimeMillis()))
+ *     .build();
+ * }</pre>
+ *
+ * @param <IN> The input type of the pulsar {@link Message <?>}
+ * @param <OUT> The output type of the source.
+ */
+@PublicEvolving
+public final class PulsarSourceBuilder<IN, OUT> {
+    private static final Logger LOG = LoggerFactory.getLogger(PulsarSourceBuilder.class);
+
+    private final Configuration configuration;
+    private PulsarSubscriber subscriber;
+    private RangeGenerator rangeGenerator;
+    private SerializableSupplier<StartCursor> startCursorSupplier;
+    private SerializableSupplier<StopCursor> stopCursorSupplier;
+    private Boundedness boundedness;
+    private PulsarDeserializationSchema<IN, OUT> deserializationSchema;
+    private ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer;
+    private ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+            consumerConfigurationCustomizer;
+
+    // private builder constructor.
+    PulsarSourceBuilder() {
+        // The default configuration holder.
+        this.configuration = new Configuration();
+    }
+
+    /**
+     * Sets the admin endpoint for the PulsarAdmin of the PulsarSource.
+     *
+     * @param adminUrl the url for the PulsarAdmin.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setAdminUrl(String adminUrl) {
+        configuration.set(PULSAR_ADMIN_URL, adminUrl);
+        return this;
+    }
+
+    /**
+     * Sets the server's link for the PulsarConsumer of the PulsarSource.
+     *
+     * @param serviceUrl the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setServiceUrl(String serviceUrl) {
+        configuration.set(PULSAR_SERVICE_URL, serviceUrl);
+        return this;
+    }
+
+    /**
+     * Sets the name for this pulsar subscription.
+     *
+     * @param subscriptionName the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionName(String subscriptionName) {
+        configuration.set(PULSAR_SUBSCRIPTION_NAME, subscriptionName);
+        return this;
+    }
+
+    /**
+     * {@link SubscriptionType} is the consuming behavior for pulsar, we would generator different
+     * split by the given subscription type, it's required and quite important for end-user.
+     *
+     * @param subscriptionType The type of subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionType(SubscriptionType subscriptionType) {
+        configuration.set(PULSAR_SUBSCRIPTION_TYPE, subscriptionType);
+        return this;
+    }
+
+    /**
+     * Set a pulsar topic list for flink source. Some topic may not exist currently, consuming this
+     * non-existed topic wouldn't throw any exception. But the best solution is just consuming by
+     * using a topic regex. This method is conflict with {@code setTopicPattern}.
+     *
+     * @param topics The topic list you would like to consume message.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopics(String... topics) {
+        return setTopics(Arrays.asList(topics));
+    }
+
+    /**
+     * Set a pulsar topic list for flink source. Some topic may not exist currently, consuming this
+     * non-existed topic wouldn't throw any exception. But the best solution is just consuming by
+     * using a topic regex. This method is conflict with {@code setTopicPattern}.
+     *
+     * @param topics The topic list you would like to consume message.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopics(List<String> topics) {
+        ensureSubscriberIsNull("topics");
+        configuration.set(PULSAR_TOPIC_NAMES, PulsarJsonUtils.toString(topics));
+
+        this.subscriber = PulsarSubscriber.getTopicListSubscriber(topics);
+
+        return this;
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(String topicsPattern) {
+        return setTopicPattern(Pattern.compile(topicsPattern));
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(Pattern topicsPattern) {
+        return setTopicPattern(topicsPattern, RegexSubscriptionMode.AllTopics);
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @param regexSubscriptionMode The topic filter for regex subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(
+            String topicsPattern, RegexSubscriptionMode regexSubscriptionMode) {
+        return setTopicPattern(Pattern.compile(topicsPattern), regexSubscriptionMode);
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @param regexSubscriptionMode The topic filter for regex subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(
+            Pattern topicsPattern, RegexSubscriptionMode regexSubscriptionMode) {
+        ensureSubscriberIsNull("topic pattern");
+        configuration.set(PULSAR_TOPICS_PATTERN, topicsPattern.toString());
+        this.subscriber =
+                PulsarSubscriber.getTopicPatternSubscriber(topicsPattern, regexSubscriptionMode);
+
+        return setRegexSubscriptionMode(regexSubscriptionMode);
+    }
+
+    /**
+     * When subscribing to a topic using a regular expression, you can pick a certain type of
+     * topics.
+     *
+     * @param regexSubscriptionMode The topic type.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setRegexSubscriptionMode(
+            RegexSubscriptionMode regexSubscriptionMode) {
+        configuration.set(PULSAR_REGEX_SUBSCRIPTION_MODE, regexSubscriptionMode);
+        return this;
+    }
+
+    /**
+     * Set a topic range generator for Key_Shared subscription.
+     *
+     * @param rangeGenerator A generator which would generate a set of {@link TopicRange} for given
+     *     topic.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setRangeGenerator(RangeGenerator rangeGenerator) {
+        this.rangeGenerator = rangeGenerator;
+        if (configuration.contains(PULSAR_SUBSCRIPTION_TYPE)) {
+            SubscriptionType subscriptionType = configuration.get(PULSAR_SUBSCRIPTION_TYPE);
+            if (subscriptionType != SubscriptionType.Key_Shared) {
+                LOG.warn(
+                        "Key_Shared subscription should be used for custom rangeGenerator instead of {}",
+                        subscriptionType);
+            }
+        }
+
+        return this;
+    }
+
+    /**
+     * Specify from which offsets the PulsarSource should start consume from by providing an {@link
+     * StartCursor}.
+     *
+     * @param startCursorSupplier the supplier providing a {@link StartCursor} which set the
+     *     starting offsets for the Source.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setStartCursorSupplier(
+            SerializableSupplier<StartCursor> startCursorSupplier) {
+        this.startCursorSupplier = startCursorSupplier;
+        return this;
+    }
+
+    /**
+     * By default the PulsarSource is set to run in {@link Boundedness#CONTINUOUS_UNBOUNDED} manner
+     * and thus never stops until the Flink job fails or is canceled. To let the PulsarSource run as
+     * a streaming source but still stops at some point, one can set an {@link StartCursor} to

Review comment:
       ```suggestion
        * a streaming source but still stops at some point, one can set an {@link StopCursor} to
   ```

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/SplitsAssignmentState.java
##########
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source.enumerator;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.connector.source.SplitsAssignment;
+import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+/** The state class for recording the split assignment. */
+@Internal
+public class SplitsAssignmentState implements Serializable {
+    private static final long serialVersionUID = -3244274150389546170L;
+
+    private final SerializableSupplier<StartCursor> startCursorSupplier;
+    private final SerializableSupplier<StopCursor> stopCursorSupplier;
+    private final SourceConfiguration sourceConfiguration;
+
+    // The dynamic states for checkpoint.
+
+    private final Set<TopicPartition> appendedPartitions;
+    private final Set<PulsarPartitionSplit> pendingPartitionSplits;
+    private final Map<String, Set<Integer>> splitReaderMapping;
+    private boolean initialized;
+
+    public SplitsAssignmentState(
+            SerializableSupplier<StartCursor> startCursorSupplier,
+            SerializableSupplier<StopCursor> stopCursorSupplier,
+            SourceConfiguration sourceConfiguration) {
+        this.startCursorSupplier = startCursorSupplier;
+        this.stopCursorSupplier = stopCursorSupplier;
+        this.sourceConfiguration = sourceConfiguration;
+        this.appendedPartitions = new HashSet<>();
+        this.pendingPartitionSplits = new HashSet<>();
+        this.splitReaderMapping = new HashMap<>();
+        this.initialized = false;
+    }
+
+    public SplitsAssignmentState(
+            SerializableSupplier<StartCursor> startCursorSupplier,
+            SerializableSupplier<StopCursor> stopCursorSupplier,
+            SourceConfiguration sourceConfiguration,
+            PulsarSourceEnumState sourceEnumState) {
+        this.startCursorSupplier = startCursorSupplier;
+        this.stopCursorSupplier = stopCursorSupplier;
+        this.sourceConfiguration = sourceConfiguration;
+        this.appendedPartitions = sourceEnumState.getAppendedPartitions();
+        this.pendingPartitionSplits = sourceEnumState.getPendingPartitionSplits();
+        this.splitReaderMapping = sourceEnumState.getSplitReaderMapping();
+        this.initialized = sourceEnumState.isInitialized();
+    }
+
+    public PulsarSourceEnumState snapshotState() {
+        return new PulsarSourceEnumState(
+                appendedPartitions, pendingPartitionSplits, splitReaderMapping, initialized);
+    }
+
+    /**
+     * Append the new fetched partitions to current state. We would generate pending source split
+     * for downstream pulsar readers. Since the {@link SplitEnumeratorContext} don't support put the
+     * split back to enumerator, we don't support partition deletion.
+     *
+     * @param fetchedPartitions The partitions from the {@link PulsarSubscriber}.
+     */
+    public void appendTopicPartitions(Set<TopicPartition> fetchedPartitions) {
+        for (TopicPartition fetchedPartition : fetchedPartitions) {
+            if (!appendedPartitions.contains(fetchedPartition)) {
+                // This is a new topic partition.
+                PulsarPartitionSplit split =
+                        new PulsarPartitionSplit(
+                                fetchedPartition,
+                                startCursorSupplier.get(),
+                                stopCursorSupplier.get());
+
+                pendingPartitionSplits.add(split);
+                appendedPartitions.add(fetchedPartition);
+            }
+        }
+
+        // Update this initialize flag.
+        if (!initialized) {
+            this.initialized = true;
+        }
+    }
+
+    /** Put these splits to pending list. */
+    public void addPendingPartitionSplits(List<PulsarPartitionSplit> splits, int readerId) {
+        if (sharePartition()) {
+            for (PulsarPartitionSplit split : splits) {
+                Set<Integer> assignedReaders =
+                        splitReaderMapping.computeIfAbsent(split.splitId(), i -> new HashSet<>());
+                assignedReaders.remove(readerId);
+            }
+        }
+
+        // Put these splits back to pending list.
+        pendingPartitionSplits.addAll(splits);
+    }
+
+    public Optional<SplitsAssignment<PulsarPartitionSplit>> assignSplits(
+            List<Integer> pendingReaders) {
+        // We would support custom split assignment in the future.
+        Map<Integer, List<PulsarPartitionSplit>> assignMap = new HashMap<>();
+
+        while (!pendingPartitionSplits.isEmpty()) {
+            // Drain a list of splits.
+            List<PulsarPartitionSplit> pendingSplits = drainPendingPartitionsSplits();
+
+            if (sharePartition()) {
+                for (PulsarPartitionSplit split : pendingSplits) {
+                    Set<Integer> registeredReaders =
+                            splitReaderMapping.computeIfAbsent(
+                                    split.splitId(), i -> new HashSet<>());
+                    for (Integer pendingReader : pendingReaders) {
+                        if (!registeredReaders.contains(pendingReader)) {
+                            assignMap
+                                    .computeIfAbsent(pendingReader, id -> new ArrayList<>())
+                                    .add(split);
+                            registeredReaders.add(pendingReader);
+                        }
+                    }
+                }
+            } else {
+                for (int i = 0; i < pendingSplits.size(); i++) {
+                    PulsarPartitionSplit split = pendingSplits.get(i);
+                    int readerId = pendingReaders.get(i % pendingReaders.size());
+                    assignMap.computeIfAbsent(readerId, id -> new ArrayList<>()).add(split);
+                }
+            }
+        }
+
+        if (assignMap.isEmpty()) {
+            return Optional.empty();
+        } else {
+            return Optional.of(new SplitsAssignment<>(assignMap));
+        }
+    }
+
+    /**
+     * @return It would return true only if periodically partition discovery is disabled, the
+     *     initializing partition discovery has finished AND there is no pending splits for
+     *     assignment.
+     */
+    public boolean noMoreNewPartitionSplits() {
+        return !sourceConfiguration.enablePartitionDiscovery()
+                && initialized
+                && pendingPartitionSplits.isEmpty();
+    }
+
+    // ----------------- private methods -------------------
+
+    private List<PulsarPartitionSplit> drainPendingPartitionsSplits() {
+        List<PulsarPartitionSplit> splits = new ArrayList<>(pendingPartitionSplits);
+        splits.forEach(pendingPartitionSplits::remove);

Review comment:
       pendingPartitionSplits.clear()

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
##########
@@ -0,0 +1,575 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.common.exception.PulsarRuntimeException;
+import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.DivideRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FullRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.RegexSubscriptionMode;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.function.Consumer;
+import java.util.regex.Pattern;
+
+import static java.lang.Boolean.FALSE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.getOptionValue;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ENABLE_AUTO_COMMIT_CURSOR;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_REGEX_SUBSCRIPTION_MODE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPICS_PATTERN;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPIC_NAMES;
+import static org.apache.flink.connector.pulsar.source.config.PulsarSourceConfigUtils.checkConfigurations;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The builder class for {@link PulsarSource} to make it easier for the users to construct a {@link
+ * PulsarSource}.
+ *
+ * <p>The following example shows the minimum setup to create a PulsarSource that reads the String
+ * values from a Pulsar topic.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .build();
+ * }</pre>
+ *
+ * <p>The service url, admin url, subscription name, topics to consume, and the record deserializer
+ * are required fields that must be set.
+ *
+ * <p>To specify the starting position of PulsarSource, one can call {@link
+ * #setStartCursorSupplier(SerializableSupplier)}.
+ *
+ * <p>By default the PulsarSource runs in an {@link Boundedness#CONTINUOUS_UNBOUNDED} mode and never
+ * stop until the Flink job is canceled or fails. To let the PulsarSource run in {@link
+ * Boundedness#CONTINUOUS_UNBOUNDED} but stops at some given offsets, one can call {@link
+ * #setUnbounded(SerializableSupplier)}. For example the following PulsarSource stops after it
+ * consumes up to a event time when the Flink started.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .setUnbounded(StopCursor.atEventTime(System.currentTimeMillis()))
+ *     .build();
+ * }</pre>
+ *
+ * @param <IN> The input type of the pulsar {@link Message <?>}
+ * @param <OUT> The output type of the source.
+ */
+@PublicEvolving
+public final class PulsarSourceBuilder<IN, OUT> {
+    private static final Logger LOG = LoggerFactory.getLogger(PulsarSourceBuilder.class);
+
+    private final Configuration configuration;
+    private PulsarSubscriber subscriber;
+    private RangeGenerator rangeGenerator;
+    private SerializableSupplier<StartCursor> startCursorSupplier;
+    private SerializableSupplier<StopCursor> stopCursorSupplier;
+    private Boundedness boundedness;
+    private PulsarDeserializationSchema<IN, OUT> deserializationSchema;
+    private ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer;
+    private ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+            consumerConfigurationCustomizer;
+
+    // private builder constructor.
+    PulsarSourceBuilder() {
+        // The default configuration holder.
+        this.configuration = new Configuration();
+    }
+
+    /**
+     * Sets the admin endpoint for the PulsarAdmin of the PulsarSource.
+     *
+     * @param adminUrl the url for the PulsarAdmin.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setAdminUrl(String adminUrl) {
+        configuration.set(PULSAR_ADMIN_URL, adminUrl);
+        return this;
+    }
+
+    /**
+     * Sets the server's link for the PulsarConsumer of the PulsarSource.
+     *
+     * @param serviceUrl the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setServiceUrl(String serviceUrl) {
+        configuration.set(PULSAR_SERVICE_URL, serviceUrl);
+        return this;
+    }
+
+    /**
+     * Sets the name for this pulsar subscription.
+     *
+     * @param subscriptionName the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionName(String subscriptionName) {
+        configuration.set(PULSAR_SUBSCRIPTION_NAME, subscriptionName);
+        return this;
+    }
+
+    /**
+     * {@link SubscriptionType} is the consuming behavior for pulsar, we would generator different
+     * split by the given subscription type, it's required and quite important for end-user.

Review comment:
       I'm assuming we cannot have a default here as you want to make the subscription type a conscious decision of the user?

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigUtils.java
##########
@@ -0,0 +1,277 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.common.config;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
+import org.apache.flink.connector.pulsar.source.PulsarSourceOptions;
+
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.api.AuthenticationFactory;
+import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+import java.util.function.Function;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAMS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAM_MAP;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PLUGIN_CLASS_NAME;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTO_CERT_REFRESH_TIME;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_CONCURRENT_LOOKUP_REQUEST;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_CONNECTIONS_PER_BROKER;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_CONNECTION_TIMEOUT_MS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_CONNECT_TIMEOUT;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_BUSY_WAIT;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_INITIAL_BACKOFF_INTERVAL_NANOS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_KEEP_ALIVE_INTERVAL_SECONDS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_LISTENER_NAME;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MAX_BACKOFF_INTERVAL_NANOS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MAX_LOOKUP_REDIRECTS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MAX_LOOKUP_REQUEST;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MAX_NUMBER_OF_REJECTED_REQUEST_PER_CONNECTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MEMORY_LIMIT_BYTES;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_NUM_IO_THREADS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_NUM_LISTENER_THREADS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_OPERATION_TIMEOUT_MS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_PROXY_PROTOCOL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_PROXY_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_READ_TIMEOUT;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_REQUEST_TIMEOUT;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_REQUEST_TIMEOUT_MS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SSL_PROVIDER;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_STATS_INTERVAL_SECONDS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_ALLOW_INSECURE_CONNECTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_CIPHERS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_HOSTNAME_VERIFICATION_ENABLE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_PROTOCOLS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_TRUST_CERTS_FILE_PATH;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_TRUST_STORE_PASSWORD;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_TRUST_STORE_PATH;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_TRUST_STORE_TYPE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_USE_KEY_STORE_TLS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_USE_TCP_NO_DELAY;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_USE_TLS;
+import static org.apache.flink.connector.pulsar.common.exception.PulsarExceptionUtils.sneaky;
+import static org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils.configMap;
+
+/** The util for creating pulsar configuration class from flink's {@link Configuration}. */
+public final class PulsarConfigUtils {
+
+    private PulsarConfigUtils() {
+        // No need to create instance.
+    }
+
+    /**
+     * Create a common {@link ClientConfigurationData} for both pulsar source and sink connector by
+     * using the given {@link Configuration}. Add the configuration was listed in {@link
+     * PulsarSourceOptions}, and we would parse it one by one.
+     *
+     * @param configuration The flink configuration
+     */
+    public static ClientConfigurationData createClientConfig(
+            Configuration configuration,
+            @Nullable ConfigurationDataCustomizer<ClientConfigurationData> customizer) {
+        ClientConfigurationData data = new ClientConfigurationData();
+
+        // Set the properties one by one.
+        data.setServiceUrl(configuration.get(PULSAR_SERVICE_URL));
+        data.setAuthPluginClassName(configuration.get(PULSAR_AUTH_PLUGIN_CLASS_NAME));
+        data.setAuthParams(configuration.get(PULSAR_AUTH_PARAMS));
+        data.setAuthParamMap(
+                getOptionValue(
+                        configuration,
+                        PULSAR_AUTH_PARAM_MAP,
+                        v -> PulsarJsonUtils.toMap(String.class, String.class, v)));
+        data.setOperationTimeoutMs(configuration.get(PULSAR_OPERATION_TIMEOUT_MS));
+        data.setStatsIntervalSeconds(configuration.get(PULSAR_STATS_INTERVAL_SECONDS));
+        data.setNumIoThreads(configuration.get(PULSAR_NUM_IO_THREADS));
+        data.setNumListenerThreads(configuration.get(PULSAR_NUM_LISTENER_THREADS));
+        data.setConnectionsPerBroker(configuration.get(PULSAR_CONNECTIONS_PER_BROKER));
+        data.setUseTcpNoDelay(configuration.get(PULSAR_USE_TCP_NO_DELAY));
+        data.setUseTls(configuration.get(PULSAR_USE_TLS));
+        data.setTlsTrustCertsFilePath(configuration.get(PULSAR_TLS_TRUST_CERTS_FILE_PATH));
+        data.setTlsAllowInsecureConnection(configuration.get(PULSAR_TLS_ALLOW_INSECURE_CONNECTION));
+        data.setTlsHostnameVerificationEnable(
+                configuration.get(PULSAR_TLS_HOSTNAME_VERIFICATION_ENABLE));
+        data.setConcurrentLookupRequest(configuration.get(PULSAR_CONCURRENT_LOOKUP_REQUEST));
+        data.setMaxLookupRequest(configuration.get(PULSAR_MAX_LOOKUP_REQUEST));
+        data.setMaxLookupRedirects(configuration.get(PULSAR_MAX_LOOKUP_REDIRECTS));
+        data.setMaxNumberOfRejectedRequestPerConnection(
+                configuration.get(PULSAR_MAX_NUMBER_OF_REJECTED_REQUEST_PER_CONNECTION));
+        data.setKeepAliveIntervalSeconds(configuration.get(PULSAR_KEEP_ALIVE_INTERVAL_SECONDS));
+        data.setConnectionTimeoutMs(configuration.get(PULSAR_CONNECTION_TIMEOUT_MS));
+        data.setRequestTimeoutMs(configuration.get(PULSAR_REQUEST_TIMEOUT_MS));
+        data.setInitialBackoffIntervalNanos(
+                configuration.get(PULSAR_INITIAL_BACKOFF_INTERVAL_NANOS));
+        data.setMaxBackoffIntervalNanos(configuration.get(PULSAR_MAX_BACKOFF_INTERVAL_NANOS));
+        data.setEnableBusyWait(configuration.get(PULSAR_ENABLE_BUSY_WAIT));
+        data.setListenerName(configuration.get(PULSAR_LISTENER_NAME));
+        data.setUseKeyStoreTls(configuration.get(PULSAR_USE_KEY_STORE_TLS));
+        data.setSslProvider(configuration.get(PULSAR_SSL_PROVIDER));
+        data.setTlsTrustStoreType(configuration.get(PULSAR_TLS_TRUST_STORE_TYPE));
+        data.setTlsTrustStorePath(configuration.get(PULSAR_TLS_TRUST_STORE_PATH));
+        data.setTlsTrustStorePassword(configuration.get(PULSAR_TLS_TRUST_STORE_PASSWORD));
+        data.setTlsCiphers(
+                getOptionValue(
+                        configuration,
+                        PULSAR_TLS_CIPHERS,
+                        v -> PulsarJsonUtils.toSet(String.class, v)));
+        data.setTlsProtocols(
+                getOptionValue(
+                        configuration,
+                        PULSAR_TLS_PROTOCOLS,
+                        v -> PulsarJsonUtils.toSet(String.class, v)));
+        data.setMemoryLimitBytes(configuration.get(PULSAR_MEMORY_LIMIT_BYTES));
+        data.setProxyServiceUrl(configuration.get(PULSAR_PROXY_SERVICE_URL));
+        data.setProxyProtocol(configuration.get(PULSAR_PROXY_PROTOCOL));
+        data.setEnableTransaction(configuration.get(PULSAR_ENABLE_TRANSACTION));
+
+        // Create Authentication instance like pulsar command line tools.
+        // The token and TLS authentication should be created by the below factory method.
+        data.setAuthentication(createAuthentication(configuration));
+
+        if (customizer != null) {
+            customizer.customize(data);
+        }
+
+        return data;
+    }
+
+    /**
+     * Create the {@link Authentication} instance for both {@code PulsarClient} and {@code
+     * PulsarAdmin}. If the user didn't provide configuration, a {@link AuthenticationDisabled}
+     * instance would be returned.
+     */
+    public static Authentication createAuthentication(Configuration configuration) {
+        if (configuration.contains(PULSAR_AUTH_PLUGIN_CLASS_NAME)) {
+            String authPluginClassName = configuration.get(PULSAR_AUTH_PLUGIN_CLASS_NAME);
+
+            if (configuration.contains(PULSAR_AUTH_PARAMS)) {
+                String authParamsString = configuration.get(PULSAR_AUTH_PARAMS);
+                return sneaky(

Review comment:
       Which kinds of exceptions are swallowed? I'm assuming that there is no recovery possible anyways when authentication config is incorrect but then it's weird that AuthenticationFactory uses checked exceptions...

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarOptions.java
##########
@@ -0,0 +1,353 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.common.config;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+
+import org.apache.pulsar.client.api.ProxyProtocol;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/**
+ * Configuration for Pulsar Client, these config options would be used for both source, sink and
+ * table.
+ */
+@PublicEvolving
+public final class PulsarOptions {

Review comment:
       Wow, that is really complete now. Is there any way to rank the options or subdivide them in more common or uncommon options? As a new user, I wouldn't know which one to use. Or are all of them supposed to be advanced?

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSource.java
##########
@@ -0,0 +1,265 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.java.ClosureCleaner;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumState;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumStateSerializer;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumerator;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.PulsarSourceReaderFactory;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarSchemaInitializationContext;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitSerializer;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.createAdmin;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.createClient;
+import static org.apache.flink.connector.pulsar.common.exception.PulsarExceptionUtils.sneaky;
+
+/**
+ * The Source implementation of Pulsar. Please use a {@link PulsarSourceBuilder} to construct a
+ * {@link PulsarSource}. The following example shows how to create a PulsarSource emitting records
+ * of <code>String</code> type.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource.<byte[], String>builder()
+ *     .setTopics(TOPIC1， TOPIC2)
+ *     .setServiceUrl(getServiceUrl())
+ *     .setAdminUrl(getAdminUrl())
+ *     .setSubscriptionName("test")
+ *     .setDeserializationSchema(flinkSchema(new SimpleStringSchema()))
+ *     .setBounded(StopCursor::defaultStopCursor)
+ *     .build();
+ * }</pre>
+ *
+ * <p>See {@link PulsarSourceBuilder} for more details.
+ *
+ * @param <IN> The input type of the pulsar {@link Message}.
+ * @param <OUT> The output type of the source.
+ */
+@PublicEvolving
+public final class PulsarSource<IN, OUT>
+        implements Source<OUT, PulsarPartitionSplit, PulsarSourceEnumState>,
+                ResultTypeQueryable<OUT> {
+    private static final long serialVersionUID = 7773108631275567433L;
+
+    /**
+     * The common configuration for pulsar source, we don't support the pulsar's configuration class
+     * directly.
+     */
+    private final Configuration configuration;
+
+    private final SourceConfiguration sourceConfiguration;
+
+    private final PulsarSubscriber subscriber;
+
+    private final RangeGenerator rangeGenerator;
+
+    private final SerializableSupplier<StartCursor> startCursorSupplier;
+
+    private final SerializableSupplier<StopCursor> stopCursorSupplier;
+
+    /**
+     * Boundedness for source, we only support {@link Boundedness#CONTINUOUS_UNBOUNDED} currently.

Review comment:
       This comment seems outdated.

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigUtils.java
##########
@@ -0,0 +1,277 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.common.config;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
+import org.apache.flink.connector.pulsar.source.PulsarSourceOptions;
+
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.api.AuthenticationFactory;
+import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+import java.util.function.Function;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAMS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAM_MAP;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PLUGIN_CLASS_NAME;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTO_CERT_REFRESH_TIME;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_CONCURRENT_LOOKUP_REQUEST;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_CONNECTIONS_PER_BROKER;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_CONNECTION_TIMEOUT_MS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_CONNECT_TIMEOUT;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_BUSY_WAIT;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_INITIAL_BACKOFF_INTERVAL_NANOS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_KEEP_ALIVE_INTERVAL_SECONDS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_LISTENER_NAME;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MAX_BACKOFF_INTERVAL_NANOS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MAX_LOOKUP_REDIRECTS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MAX_LOOKUP_REQUEST;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MAX_NUMBER_OF_REJECTED_REQUEST_PER_CONNECTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MEMORY_LIMIT_BYTES;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_NUM_IO_THREADS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_NUM_LISTENER_THREADS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_OPERATION_TIMEOUT_MS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_PROXY_PROTOCOL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_PROXY_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_READ_TIMEOUT;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_REQUEST_TIMEOUT;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_REQUEST_TIMEOUT_MS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SSL_PROVIDER;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_STATS_INTERVAL_SECONDS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_ALLOW_INSECURE_CONNECTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_CIPHERS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_HOSTNAME_VERIFICATION_ENABLE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_PROTOCOLS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_TRUST_CERTS_FILE_PATH;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_TRUST_STORE_PASSWORD;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_TRUST_STORE_PATH;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_TRUST_STORE_TYPE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_USE_KEY_STORE_TLS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_USE_TCP_NO_DELAY;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_USE_TLS;
+import static org.apache.flink.connector.pulsar.common.exception.PulsarExceptionUtils.sneaky;
+import static org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils.configMap;
+
+/** The util for creating pulsar configuration class from flink's {@link Configuration}. */
+public final class PulsarConfigUtils {

Review comment:
       `@Internal`?

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigUtils.java
##########
@@ -0,0 +1,277 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.common.config;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
+import org.apache.flink.connector.pulsar.source.PulsarSourceOptions;
+
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.api.AuthenticationFactory;
+import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+import java.util.function.Function;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAMS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAM_MAP;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PLUGIN_CLASS_NAME;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTO_CERT_REFRESH_TIME;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_CONCURRENT_LOOKUP_REQUEST;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_CONNECTIONS_PER_BROKER;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_CONNECTION_TIMEOUT_MS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_CONNECT_TIMEOUT;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_BUSY_WAIT;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_INITIAL_BACKOFF_INTERVAL_NANOS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_KEEP_ALIVE_INTERVAL_SECONDS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_LISTENER_NAME;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MAX_BACKOFF_INTERVAL_NANOS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MAX_LOOKUP_REDIRECTS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MAX_LOOKUP_REQUEST;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MAX_NUMBER_OF_REJECTED_REQUEST_PER_CONNECTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_MEMORY_LIMIT_BYTES;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_NUM_IO_THREADS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_NUM_LISTENER_THREADS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_OPERATION_TIMEOUT_MS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_PROXY_PROTOCOL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_PROXY_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_READ_TIMEOUT;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_REQUEST_TIMEOUT;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_REQUEST_TIMEOUT_MS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SSL_PROVIDER;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_STATS_INTERVAL_SECONDS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_ALLOW_INSECURE_CONNECTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_CIPHERS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_HOSTNAME_VERIFICATION_ENABLE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_PROTOCOLS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_TRUST_CERTS_FILE_PATH;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_TRUST_STORE_PASSWORD;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_TRUST_STORE_PATH;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_TRUST_STORE_TYPE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_USE_KEY_STORE_TLS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_USE_TCP_NO_DELAY;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_USE_TLS;
+import static org.apache.flink.connector.pulsar.common.exception.PulsarExceptionUtils.sneaky;
+import static org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils.configMap;
+
+/** The util for creating pulsar configuration class from flink's {@link Configuration}. */
+public final class PulsarConfigUtils {
+
+    private PulsarConfigUtils() {
+        // No need to create instance.
+    }
+
+    /**
+     * Create a common {@link ClientConfigurationData} for both pulsar source and sink connector by
+     * using the given {@link Configuration}. Add the configuration was listed in {@link
+     * PulsarSourceOptions}, and we would parse it one by one.
+     *
+     * @param configuration The flink configuration
+     */
+    public static ClientConfigurationData createClientConfig(
+            Configuration configuration,
+            @Nullable ConfigurationDataCustomizer<ClientConfigurationData> customizer) {
+        ClientConfigurationData data = new ClientConfigurationData();
+
+        // Set the properties one by one.
+        data.setServiceUrl(configuration.get(PULSAR_SERVICE_URL));
+        data.setAuthPluginClassName(configuration.get(PULSAR_AUTH_PLUGIN_CLASS_NAME));
+        data.setAuthParams(configuration.get(PULSAR_AUTH_PARAMS));
+        data.setAuthParamMap(
+                getOptionValue(
+                        configuration,
+                        PULSAR_AUTH_PARAM_MAP,
+                        v -> PulsarJsonUtils.toMap(String.class, String.class, v)));
+        data.setOperationTimeoutMs(configuration.get(PULSAR_OPERATION_TIMEOUT_MS));
+        data.setStatsIntervalSeconds(configuration.get(PULSAR_STATS_INTERVAL_SECONDS));
+        data.setNumIoThreads(configuration.get(PULSAR_NUM_IO_THREADS));
+        data.setNumListenerThreads(configuration.get(PULSAR_NUM_LISTENER_THREADS));
+        data.setConnectionsPerBroker(configuration.get(PULSAR_CONNECTIONS_PER_BROKER));
+        data.setUseTcpNoDelay(configuration.get(PULSAR_USE_TCP_NO_DELAY));
+        data.setUseTls(configuration.get(PULSAR_USE_TLS));
+        data.setTlsTrustCertsFilePath(configuration.get(PULSAR_TLS_TRUST_CERTS_FILE_PATH));
+        data.setTlsAllowInsecureConnection(configuration.get(PULSAR_TLS_ALLOW_INSECURE_CONNECTION));
+        data.setTlsHostnameVerificationEnable(
+                configuration.get(PULSAR_TLS_HOSTNAME_VERIFICATION_ENABLE));
+        data.setConcurrentLookupRequest(configuration.get(PULSAR_CONCURRENT_LOOKUP_REQUEST));
+        data.setMaxLookupRequest(configuration.get(PULSAR_MAX_LOOKUP_REQUEST));
+        data.setMaxLookupRedirects(configuration.get(PULSAR_MAX_LOOKUP_REDIRECTS));
+        data.setMaxNumberOfRejectedRequestPerConnection(
+                configuration.get(PULSAR_MAX_NUMBER_OF_REJECTED_REQUEST_PER_CONNECTION));
+        data.setKeepAliveIntervalSeconds(configuration.get(PULSAR_KEEP_ALIVE_INTERVAL_SECONDS));
+        data.setConnectionTimeoutMs(configuration.get(PULSAR_CONNECTION_TIMEOUT_MS));
+        data.setRequestTimeoutMs(configuration.get(PULSAR_REQUEST_TIMEOUT_MS));
+        data.setInitialBackoffIntervalNanos(
+                configuration.get(PULSAR_INITIAL_BACKOFF_INTERVAL_NANOS));
+        data.setMaxBackoffIntervalNanos(configuration.get(PULSAR_MAX_BACKOFF_INTERVAL_NANOS));
+        data.setEnableBusyWait(configuration.get(PULSAR_ENABLE_BUSY_WAIT));
+        data.setListenerName(configuration.get(PULSAR_LISTENER_NAME));
+        data.setUseKeyStoreTls(configuration.get(PULSAR_USE_KEY_STORE_TLS));
+        data.setSslProvider(configuration.get(PULSAR_SSL_PROVIDER));
+        data.setTlsTrustStoreType(configuration.get(PULSAR_TLS_TRUST_STORE_TYPE));
+        data.setTlsTrustStorePath(configuration.get(PULSAR_TLS_TRUST_STORE_PATH));
+        data.setTlsTrustStorePassword(configuration.get(PULSAR_TLS_TRUST_STORE_PASSWORD));
+        data.setTlsCiphers(
+                getOptionValue(
+                        configuration,
+                        PULSAR_TLS_CIPHERS,
+                        v -> PulsarJsonUtils.toSet(String.class, v)));
+        data.setTlsProtocols(
+                getOptionValue(
+                        configuration,
+                        PULSAR_TLS_PROTOCOLS,
+                        v -> PulsarJsonUtils.toSet(String.class, v)));
+        data.setMemoryLimitBytes(configuration.get(PULSAR_MEMORY_LIMIT_BYTES));
+        data.setProxyServiceUrl(configuration.get(PULSAR_PROXY_SERVICE_URL));
+        data.setProxyProtocol(configuration.get(PULSAR_PROXY_PROTOCOL));
+        data.setEnableTransaction(configuration.get(PULSAR_ENABLE_TRANSACTION));
+
+        // Create Authentication instance like pulsar command line tools.
+        // The token and TLS authentication should be created by the below factory method.

Review comment:
       I think this comment should be on #createAuthentication. But it's a good idea to have it exactly as CLI.

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/exception/PulsarExceptionUtils.java
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.common.exception;
+
+import org.apache.flink.util.function.SupplierWithException;
+import org.apache.flink.util.function.ThrowingRunnable;
+
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.PulsarClientException;
+
+/**
+ * Util class for pulsar checked exceptions. Sneaky throw {@link PulsarAdminException} and {@link
+ * PulsarClientException}.
+ */
+public final class PulsarExceptionUtils {

Review comment:
       We need to be really careful with such sneaky throws. The usual downside is that we forget or are unable to catch certain exceptions internally. I understand that we simply can't handle most `PulsarClientException` and `PulsarAdminException` but I'd feel more comfortable if we change `sneaky` to just hide these two exceptions.
   So some `sneakyClient(SupplierWithException<T, PulsarClientException> supplier)` and `sneakyAdmin(SupplierWithException<T, PulsarAdminException> supplier)`.

##########
File path: flink-connectors/flink-connector-pulsar/pom.xml
##########
@@ -0,0 +1,162 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+		 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+	<modelVersion>4.0.0</modelVersion>
+
+	<parent>
+		<artifactId>flink-connectors</artifactId>
+		<groupId>org.apache.flink</groupId>
+		<version>1.14-SNAPSHOT</version>
+		<relativePath>..</relativePath>
+	</parent>
+
+	<artifactId>flink-connector-pulsar_${scala.binary.version}</artifactId>
+	<name>Flink : Connectors : Pulsar</name>
+
+	<packaging>jar</packaging>
+
+	<properties>
+		<pulsar.version>2.8.0</pulsar.version>
+	</properties>
+	<dependencies>
+		<!-- streaming-java dependencies -->
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+			<version>${project.version}</version>
+			<scope>provided</scope>
+		</dependency>
+
+		<!-- Add Pulsar 2.x as a dependency -->
+		<dependency>
+			<groupId>org.apache.pulsar</groupId>
+			<artifactId>pulsar-client-all</artifactId>
+			<version>${pulsar.version}</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-connector-base</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+
+		<!-- flink formats dependencies -->
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-avro</artifactId>
+			<version>${project.version}</version>
+			<scope>provided</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-json</artifactId>
+			<version>${project.version}</version>
+			<scope>provided</scope>
+		</dependency>

Review comment:
       It is counter-intuitive to have them in here. Also they are not provided and need to be bundled into the user jar. Do you meant to declare them optional?

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/exception/PulsarClosingException.java
##########
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.common.exception;
+
+/** Exception for timeout when closing pulsar clients. */
+public class PulsarClosingException extends PulsarRuntimeException {

Review comment:
       I don't think we need to introduce any new exceptions. They are also not user facing. I think in most cases we could simply use `IllegalStateException` and `Preconditions.checkState`.

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSource.java
##########
@@ -0,0 +1,265 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.java.ClosureCleaner;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumState;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumStateSerializer;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumerator;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.PulsarSourceReaderFactory;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarSchemaInitializationContext;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitSerializer;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.createAdmin;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.createClient;
+import static org.apache.flink.connector.pulsar.common.exception.PulsarExceptionUtils.sneaky;
+
+/**
+ * The Source implementation of Pulsar. Please use a {@link PulsarSourceBuilder} to construct a
+ * {@link PulsarSource}. The following example shows how to create a PulsarSource emitting records
+ * of <code>String</code> type.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource.<byte[], String>builder()
+ *     .setTopics(TOPIC1， TOPIC2)
+ *     .setServiceUrl(getServiceUrl())
+ *     .setAdminUrl(getAdminUrl())
+ *     .setSubscriptionName("test")
+ *     .setDeserializationSchema(flinkSchema(new SimpleStringSchema()))
+ *     .setBounded(StopCursor::defaultStopCursor)
+ *     .build();
+ * }</pre>
+ *
+ * <p>See {@link PulsarSourceBuilder} for more details.
+ *
+ * @param <IN> The input type of the pulsar {@link Message}.
+ * @param <OUT> The output type of the source.
+ */
+@PublicEvolving
+public final class PulsarSource<IN, OUT>
+        implements Source<OUT, PulsarPartitionSplit, PulsarSourceEnumState>,
+                ResultTypeQueryable<OUT> {
+    private static final long serialVersionUID = 7773108631275567433L;
+
+    /**
+     * The common configuration for pulsar source, we don't support the pulsar's configuration class
+     * directly.
+     */
+    private final Configuration configuration;
+
+    private final SourceConfiguration sourceConfiguration;
+
+    private final PulsarSubscriber subscriber;
+
+    private final RangeGenerator rangeGenerator;
+
+    private final SerializableSupplier<StartCursor> startCursorSupplier;
+
+    private final SerializableSupplier<StopCursor> stopCursorSupplier;
+
+    /**
+     * Boundedness for source, we only support {@link Boundedness#CONTINUOUS_UNBOUNDED} currently.
+     */
+    private final Boundedness boundedness;
+
+    /** The pulsar deserialization schema used for deserialize message. */
+    private final PulsarDeserializationSchema<IN, OUT> deserializationSchema;
+
+    /** Modify the flink generated {@link ClientConfigurationData}. */
+    private final ConfigurationDataCustomizer<ClientConfigurationData>
+            clientConfigurationCustomizer;
+
+    /** Modify the flink generated {@link ConsumerConfigurationData}. */
+    private final ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+            consumerConfigurationCustomizer;
+
+    /** A lazy evaluated pulsar client for split reader. */
+    private transient volatile PulsarClient pulsarClient;
+
+    /** A lazy evaluated pulsar admin for source enumerator. */
+    private transient volatile PulsarAdmin pulsarAdmin;
+
+    /**
+     * The constructor for PulsarSource, it's package protected for forcing using {@link
+     * PulsarSourceBuilder}.
+     */
+    PulsarSource(
+            Configuration configuration,
+            PulsarSubscriber subscriber,
+            RangeGenerator rangeGenerator,
+            SerializableSupplier<StartCursor> startCursorSupplier,
+            SerializableSupplier<StopCursor> stopCursorSupplier,
+            Boundedness boundedness,
+            PulsarDeserializationSchema<IN, OUT> deserializationSchema,
+            ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer,
+            ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+                    consumerConfigurationCustomizer) {
+        // Since these implementation could be a lambda, make sure they are serializable.
+        ClosureCleaner.clean(
+                startCursorSupplier, ExecutionConfig.ClosureCleanerLevel.RECURSIVE, true);
+        ClosureCleaner.clean(
+                stopCursorSupplier, ExecutionConfig.ClosureCleanerLevel.RECURSIVE, true);
+        ClosureCleaner.clean(
+                clientConfigurationCustomizer, ExecutionConfig.ClosureCleanerLevel.RECURSIVE, true);
+        ClosureCleaner.clean(
+                consumerConfigurationCustomizer,
+                ExecutionConfig.ClosureCleanerLevel.RECURSIVE,
+                true);

Review comment:
       These cleaner should be called in the builder as all other parameter sanitization is also there.

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSource.java
##########
@@ -0,0 +1,265 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.java.ClosureCleaner;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumState;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumStateSerializer;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumerator;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.PulsarSourceReaderFactory;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarSchemaInitializationContext;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitSerializer;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.createAdmin;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.createClient;
+import static org.apache.flink.connector.pulsar.common.exception.PulsarExceptionUtils.sneaky;
+
+/**
+ * The Source implementation of Pulsar. Please use a {@link PulsarSourceBuilder} to construct a
+ * {@link PulsarSource}. The following example shows how to create a PulsarSource emitting records
+ * of <code>String</code> type.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource.<byte[], String>builder()
+ *     .setTopics(TOPIC1， TOPIC2)
+ *     .setServiceUrl(getServiceUrl())
+ *     .setAdminUrl(getAdminUrl())
+ *     .setSubscriptionName("test")
+ *     .setDeserializationSchema(flinkSchema(new SimpleStringSchema()))
+ *     .setBounded(StopCursor::defaultStopCursor)
+ *     .build();
+ * }</pre>
+ *
+ * <p>See {@link PulsarSourceBuilder} for more details.
+ *
+ * @param <IN> The input type of the pulsar {@link Message}.

Review comment:
       Having an IN type is very atypical for sources. Usually, it's just `byte[]` and we can reduce complexity for the user to hide that type.
   If it's another type than IN, it usually means that Pulsar Client already does some automatic parsing and then Flink would deserialize that again?
   If you want to keep it, make sure that you also have tests with different tests...
   For Kafka, we simply force `byte[]` and fail if the user tries to use a different kafka deserializer.

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/exception/PulsarRuntimeException.java
##########
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.common.exception;
+
+import org.apache.pulsar.client.api.PulsarClientException;
+
+/**
+ * A exception wrapper for pulsar checked {@link PulsarClientException}. Or you can use it directly
+ * for some source exceptions.
+ */
+public class PulsarRuntimeException extends RuntimeException {

Review comment:
       I don't think we need to introduce any new exceptions. They are also not user facing. I think in most cases we could simply use `IllegalStateException` and `Preconditions.checkState`.

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
##########
@@ -0,0 +1,575 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.common.exception.PulsarRuntimeException;
+import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.DivideRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FullRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.RegexSubscriptionMode;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.function.Consumer;
+import java.util.regex.Pattern;
+
+import static java.lang.Boolean.FALSE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.getOptionValue;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ENABLE_AUTO_COMMIT_CURSOR;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_REGEX_SUBSCRIPTION_MODE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPICS_PATTERN;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPIC_NAMES;
+import static org.apache.flink.connector.pulsar.source.config.PulsarSourceConfigUtils.checkConfigurations;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The builder class for {@link PulsarSource} to make it easier for the users to construct a {@link
+ * PulsarSource}.
+ *
+ * <p>The following example shows the minimum setup to create a PulsarSource that reads the String
+ * values from a Pulsar topic.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .build();
+ * }</pre>
+ *
+ * <p>The service url, admin url, subscription name, topics to consume, and the record deserializer
+ * are required fields that must be set.
+ *
+ * <p>To specify the starting position of PulsarSource, one can call {@link
+ * #setStartCursorSupplier(SerializableSupplier)}.
+ *
+ * <p>By default the PulsarSource runs in an {@link Boundedness#CONTINUOUS_UNBOUNDED} mode and never
+ * stop until the Flink job is canceled or fails. To let the PulsarSource run in {@link
+ * Boundedness#CONTINUOUS_UNBOUNDED} but stops at some given offsets, one can call {@link
+ * #setUnbounded(SerializableSupplier)}. For example the following PulsarSource stops after it
+ * consumes up to a event time when the Flink started.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .setUnbounded(StopCursor.atEventTime(System.currentTimeMillis()))
+ *     .build();
+ * }</pre>
+ *
+ * @param <IN> The input type of the pulsar {@link Message <?>}
+ * @param <OUT> The output type of the source.
+ */
+@PublicEvolving
+public final class PulsarSourceBuilder<IN, OUT> {
+    private static final Logger LOG = LoggerFactory.getLogger(PulsarSourceBuilder.class);
+
+    private final Configuration configuration;
+    private PulsarSubscriber subscriber;
+    private RangeGenerator rangeGenerator;
+    private SerializableSupplier<StartCursor> startCursorSupplier;
+    private SerializableSupplier<StopCursor> stopCursorSupplier;
+    private Boundedness boundedness;
+    private PulsarDeserializationSchema<IN, OUT> deserializationSchema;
+    private ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer;
+    private ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+            consumerConfigurationCustomizer;
+
+    // private builder constructor.
+    PulsarSourceBuilder() {
+        // The default configuration holder.
+        this.configuration = new Configuration();
+    }
+
+    /**
+     * Sets the admin endpoint for the PulsarAdmin of the PulsarSource.
+     *
+     * @param adminUrl the url for the PulsarAdmin.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setAdminUrl(String adminUrl) {
+        configuration.set(PULSAR_ADMIN_URL, adminUrl);
+        return this;
+    }
+
+    /**
+     * Sets the server's link for the PulsarConsumer of the PulsarSource.
+     *
+     * @param serviceUrl the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setServiceUrl(String serviceUrl) {
+        configuration.set(PULSAR_SERVICE_URL, serviceUrl);
+        return this;
+    }
+
+    /**
+     * Sets the name for this pulsar subscription.
+     *
+     * @param subscriptionName the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionName(String subscriptionName) {
+        configuration.set(PULSAR_SUBSCRIPTION_NAME, subscriptionName);
+        return this;
+    }
+
+    /**
+     * {@link SubscriptionType} is the consuming behavior for pulsar, we would generator different
+     * split by the given subscription type, it's required and quite important for end-user.

Review comment:
       I'd omit `and quite important for end-user` or add why it's so super important.

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
##########
@@ -0,0 +1,575 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.common.exception.PulsarRuntimeException;
+import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.DivideRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FullRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.RegexSubscriptionMode;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.function.Consumer;
+import java.util.regex.Pattern;
+
+import static java.lang.Boolean.FALSE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.getOptionValue;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ENABLE_AUTO_COMMIT_CURSOR;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_REGEX_SUBSCRIPTION_MODE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPICS_PATTERN;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPIC_NAMES;
+import static org.apache.flink.connector.pulsar.source.config.PulsarSourceConfigUtils.checkConfigurations;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The builder class for {@link PulsarSource} to make it easier for the users to construct a {@link
+ * PulsarSource}.
+ *
+ * <p>The following example shows the minimum setup to create a PulsarSource that reads the String
+ * values from a Pulsar topic.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .build();
+ * }</pre>
+ *
+ * <p>The service url, admin url, subscription name, topics to consume, and the record deserializer
+ * are required fields that must be set.
+ *
+ * <p>To specify the starting position of PulsarSource, one can call {@link
+ * #setStartCursorSupplier(SerializableSupplier)}.
+ *
+ * <p>By default the PulsarSource runs in an {@link Boundedness#CONTINUOUS_UNBOUNDED} mode and never
+ * stop until the Flink job is canceled or fails. To let the PulsarSource run in {@link
+ * Boundedness#CONTINUOUS_UNBOUNDED} but stops at some given offsets, one can call {@link
+ * #setUnbounded(SerializableSupplier)}. For example the following PulsarSource stops after it
+ * consumes up to a event time when the Flink started.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .setUnbounded(StopCursor.atEventTime(System.currentTimeMillis()))
+ *     .build();
+ * }</pre>
+ *
+ * @param <IN> The input type of the pulsar {@link Message <?>}
+ * @param <OUT> The output type of the source.
+ */
+@PublicEvolving
+public final class PulsarSourceBuilder<IN, OUT> {
+    private static final Logger LOG = LoggerFactory.getLogger(PulsarSourceBuilder.class);
+
+    private final Configuration configuration;
+    private PulsarSubscriber subscriber;
+    private RangeGenerator rangeGenerator;
+    private SerializableSupplier<StartCursor> startCursorSupplier;
+    private SerializableSupplier<StopCursor> stopCursorSupplier;
+    private Boundedness boundedness;
+    private PulsarDeserializationSchema<IN, OUT> deserializationSchema;
+    private ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer;
+    private ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+            consumerConfigurationCustomizer;
+
+    // private builder constructor.
+    PulsarSourceBuilder() {
+        // The default configuration holder.
+        this.configuration = new Configuration();
+    }
+
+    /**
+     * Sets the admin endpoint for the PulsarAdmin of the PulsarSource.
+     *
+     * @param adminUrl the url for the PulsarAdmin.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setAdminUrl(String adminUrl) {
+        configuration.set(PULSAR_ADMIN_URL, adminUrl);
+        return this;
+    }
+
+    /**
+     * Sets the server's link for the PulsarConsumer of the PulsarSource.
+     *
+     * @param serviceUrl the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setServiceUrl(String serviceUrl) {
+        configuration.set(PULSAR_SERVICE_URL, serviceUrl);
+        return this;
+    }
+
+    /**
+     * Sets the name for this pulsar subscription.
+     *
+     * @param subscriptionName the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionName(String subscriptionName) {
+        configuration.set(PULSAR_SUBSCRIPTION_NAME, subscriptionName);
+        return this;
+    }
+
+    /**
+     * {@link SubscriptionType} is the consuming behavior for pulsar, we would generator different
+     * split by the given subscription type, it's required and quite important for end-user.
+     *
+     * @param subscriptionType The type of subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionType(SubscriptionType subscriptionType) {
+        configuration.set(PULSAR_SUBSCRIPTION_TYPE, subscriptionType);

Review comment:
       Check and fail if subscription is already set? Or else the user could do
   ```
   setSubscriptionType(Key_Shared)
   setRangeGenerator(...)
   setSubscriptionType(something else)
   ```

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
##########
@@ -0,0 +1,575 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.common.exception.PulsarRuntimeException;
+import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.DivideRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FullRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.RegexSubscriptionMode;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.function.Consumer;
+import java.util.regex.Pattern;
+
+import static java.lang.Boolean.FALSE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.getOptionValue;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ENABLE_AUTO_COMMIT_CURSOR;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_REGEX_SUBSCRIPTION_MODE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPICS_PATTERN;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPIC_NAMES;
+import static org.apache.flink.connector.pulsar.source.config.PulsarSourceConfigUtils.checkConfigurations;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The builder class for {@link PulsarSource} to make it easier for the users to construct a {@link
+ * PulsarSource}.
+ *
+ * <p>The following example shows the minimum setup to create a PulsarSource that reads the String
+ * values from a Pulsar topic.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .build();
+ * }</pre>
+ *
+ * <p>The service url, admin url, subscription name, topics to consume, and the record deserializer
+ * are required fields that must be set.
+ *
+ * <p>To specify the starting position of PulsarSource, one can call {@link
+ * #setStartCursorSupplier(SerializableSupplier)}.
+ *
+ * <p>By default the PulsarSource runs in an {@link Boundedness#CONTINUOUS_UNBOUNDED} mode and never
+ * stop until the Flink job is canceled or fails. To let the PulsarSource run in {@link
+ * Boundedness#CONTINUOUS_UNBOUNDED} but stops at some given offsets, one can call {@link
+ * #setUnbounded(SerializableSupplier)}. For example the following PulsarSource stops after it
+ * consumes up to a event time when the Flink started.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .setUnbounded(StopCursor.atEventTime(System.currentTimeMillis()))
+ *     .build();
+ * }</pre>
+ *
+ * @param <IN> The input type of the pulsar {@link Message <?>}
+ * @param <OUT> The output type of the source.
+ */
+@PublicEvolving
+public final class PulsarSourceBuilder<IN, OUT> {
+    private static final Logger LOG = LoggerFactory.getLogger(PulsarSourceBuilder.class);
+
+    private final Configuration configuration;
+    private PulsarSubscriber subscriber;
+    private RangeGenerator rangeGenerator;
+    private SerializableSupplier<StartCursor> startCursorSupplier;
+    private SerializableSupplier<StopCursor> stopCursorSupplier;
+    private Boundedness boundedness;
+    private PulsarDeserializationSchema<IN, OUT> deserializationSchema;
+    private ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer;
+    private ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+            consumerConfigurationCustomizer;
+
+    // private builder constructor.
+    PulsarSourceBuilder() {
+        // The default configuration holder.
+        this.configuration = new Configuration();
+    }
+
+    /**
+     * Sets the admin endpoint for the PulsarAdmin of the PulsarSource.
+     *
+     * @param adminUrl the url for the PulsarAdmin.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setAdminUrl(String adminUrl) {
+        configuration.set(PULSAR_ADMIN_URL, adminUrl);
+        return this;
+    }
+
+    /**
+     * Sets the server's link for the PulsarConsumer of the PulsarSource.
+     *
+     * @param serviceUrl the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setServiceUrl(String serviceUrl) {
+        configuration.set(PULSAR_SERVICE_URL, serviceUrl);
+        return this;
+    }
+
+    /**
+     * Sets the name for this pulsar subscription.
+     *
+     * @param subscriptionName the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionName(String subscriptionName) {
+        configuration.set(PULSAR_SUBSCRIPTION_NAME, subscriptionName);
+        return this;
+    }
+
+    /**
+     * {@link SubscriptionType} is the consuming behavior for pulsar, we would generator different
+     * split by the given subscription type, it's required and quite important for end-user.
+     *
+     * @param subscriptionType The type of subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionType(SubscriptionType subscriptionType) {
+        configuration.set(PULSAR_SUBSCRIPTION_TYPE, subscriptionType);
+        return this;
+    }
+
+    /**
+     * Set a pulsar topic list for flink source. Some topic may not exist currently, consuming this
+     * non-existed topic wouldn't throw any exception. But the best solution is just consuming by
+     * using a topic regex. This method is conflict with {@code setTopicPattern}.
+     *
+     * @param topics The topic list you would like to consume message.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopics(String... topics) {
+        return setTopics(Arrays.asList(topics));
+    }
+
+    /**
+     * Set a pulsar topic list for flink source. Some topic may not exist currently, consuming this
+     * non-existed topic wouldn't throw any exception. But the best solution is just consuming by
+     * using a topic regex. This method is conflict with {@code setTopicPattern}.
+     *
+     * @param topics The topic list you would like to consume message.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopics(List<String> topics) {
+        ensureSubscriberIsNull("topics");
+        configuration.set(PULSAR_TOPIC_NAMES, PulsarJsonUtils.toString(topics));
+
+        this.subscriber = PulsarSubscriber.getTopicListSubscriber(topics);
+
+        return this;
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(String topicsPattern) {
+        return setTopicPattern(Pattern.compile(topicsPattern));
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(Pattern topicsPattern) {
+        return setTopicPattern(topicsPattern, RegexSubscriptionMode.AllTopics);
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @param regexSubscriptionMode The topic filter for regex subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(
+            String topicsPattern, RegexSubscriptionMode regexSubscriptionMode) {
+        return setTopicPattern(Pattern.compile(topicsPattern), regexSubscriptionMode);
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @param regexSubscriptionMode The topic filter for regex subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(
+            Pattern topicsPattern, RegexSubscriptionMode regexSubscriptionMode) {
+        ensureSubscriberIsNull("topic pattern");
+        configuration.set(PULSAR_TOPICS_PATTERN, topicsPattern.toString());
+        this.subscriber =
+                PulsarSubscriber.getTopicPatternSubscriber(topicsPattern, regexSubscriptionMode);
+
+        return setRegexSubscriptionMode(regexSubscriptionMode);
+    }
+
+    /**
+     * When subscribing to a topic using a regular expression, you can pick a certain type of
+     * topics.
+     *
+     * @param regexSubscriptionMode The topic type.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setRegexSubscriptionMode(
+            RegexSubscriptionMode regexSubscriptionMode) {
+        configuration.set(PULSAR_REGEX_SUBSCRIPTION_MODE, regexSubscriptionMode);
+        return this;
+    }
+
+    /**
+     * Set a topic range generator for Key_Shared subscription.
+     *
+     * @param rangeGenerator A generator which would generate a set of {@link TopicRange} for given
+     *     topic.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setRangeGenerator(RangeGenerator rangeGenerator) {
+        this.rangeGenerator = rangeGenerator;
+        if (configuration.contains(PULSAR_SUBSCRIPTION_TYPE)) {
+            SubscriptionType subscriptionType = configuration.get(PULSAR_SUBSCRIPTION_TYPE);
+            if (subscriptionType != SubscriptionType.Key_Shared) {
+                LOG.warn(
+                        "Key_Shared subscription should be used for custom rangeGenerator instead of {}",
+                        subscriptionType);
+            }
+        }
+
+        return this;
+    }
+
+    /**
+     * Specify from which offsets the PulsarSource should start consume from by providing an {@link
+     * StartCursor}.
+     *
+     * @param startCursorSupplier the supplier providing a {@link StartCursor} which set the
+     *     starting offsets for the Source.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setStartCursorSupplier(
+            SerializableSupplier<StartCursor> startCursorSupplier) {
+        this.startCursorSupplier = startCursorSupplier;
+        return this;
+    }
+
+    /**
+     * By default the PulsarSource is set to run in {@link Boundedness#CONTINUOUS_UNBOUNDED} manner
+     * and thus never stops until the Flink job fails or is canceled. To let the PulsarSource run as
+     * a streaming source but still stops at some point, one can set an {@link StartCursor} to
+     * specify the stopping offsets for each partition. When all the partitions have reached their
+     * stopping offsets, the PulsarSource will then exit.
+     *
+     * <p>This method is different from {@link #setBounded(SerializableSupplier)} that after setting
+     * the stopping offsets with this method, {@link PulsarSource#getBoundedness()} will still
+     * return {@link Boundedness#CONTINUOUS_UNBOUNDED} even though it will stop at the stopping
+     * offsets specified by the stopping offsets {@link StartCursor}.
+     *
+     * @param stopCursorSupplier The {@link StopCursor} to specify the stopping offset.
+     * @return this PulsarSourceBuilder.
+     * @see #setBounded(SerializableSupplier)
+     */
+    public PulsarSourceBuilder<IN, OUT> setUnbounded(
+            SerializableSupplier<StopCursor> stopCursorSupplier) {
+        this.boundedness = Boundedness.CONTINUOUS_UNBOUNDED;
+        this.stopCursorSupplier = stopCursorSupplier;
+        return this;
+    }
+
+    /**
+     * By default the PulsarSource is set to run in {@link Boundedness#CONTINUOUS_UNBOUNDED} manner
+     * and thus never stops until the Flink job fails or is canceled. To let the PulsarSource run in
+     * {@link Boundedness#BOUNDED} manner and stops at some point, one can set an {@link
+     * StartCursor} to specify the stopping offsets for each partition. When all the partitions have
+     * reached their stopping offsets, the PulsarSource will then exit.
+     *
+     * <p>This method is different from {@link #setUnbounded(SerializableSupplier)} that after
+     * setting the stopping offsets with this method, {@link PulsarSource#getBoundedness()} will
+     * return {@link Boundedness#BOUNDED} instead of {@link Boundedness#CONTINUOUS_UNBOUNDED}.
+     *
+     * @param stopCursorSupplier the {@link StopCursor} to specify the stopping offsets.
+     * @return this PulsarSourceBuilder.
+     * @see #setUnbounded(SerializableSupplier)
+     */
+    public PulsarSourceBuilder<IN, OUT> setBounded(
+            SerializableSupplier<StopCursor> stopCursorSupplier) {
+        this.boundedness = Boundedness.BOUNDED;
+        this.stopCursorSupplier = stopCursorSupplier;
+        return this;
+    }
+
+    /**
+     * DeserializationSchema is required for getting the {@link Schema} for deserialize message from
+     * pulsar and getting the {@link TypeInformation} for message serialization in flink.
+     *
+     * <p>We have defined a set of implementations, using {@code
+     * PulsarDeserializationSchema#pulsarSchema} or {@code PulsarDeserializationSchema#flinkSchema}
+     * for creating the desired schema.
+     */
+    public PulsarSourceBuilder<IN, OUT> setDeserializationSchema(

Review comment:
       `public <DESER_IN extends IN, DESER_OUT extends OUT> setDeserializationSchema`
   allows us to omit the type information when calling `PulsarSource.builder`. IN and OUT will then be simply Object and any `deserializationSchema` will suffice the lower bounds.

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSource.java
##########
@@ -0,0 +1,265 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.java.ClosureCleaner;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumState;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumStateSerializer;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumerator;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.PulsarSourceReaderFactory;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarSchemaInitializationContext;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitSerializer;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.createAdmin;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.createClient;
+import static org.apache.flink.connector.pulsar.common.exception.PulsarExceptionUtils.sneaky;
+
+/**
+ * The Source implementation of Pulsar. Please use a {@link PulsarSourceBuilder} to construct a
+ * {@link PulsarSource}. The following example shows how to create a PulsarSource emitting records
+ * of <code>String</code> type.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource.<byte[], String>builder()
+ *     .setTopics(TOPIC1， TOPIC2)
+ *     .setServiceUrl(getServiceUrl())
+ *     .setAdminUrl(getAdminUrl())
+ *     .setSubscriptionName("test")
+ *     .setDeserializationSchema(flinkSchema(new SimpleStringSchema()))
+ *     .setBounded(StopCursor::defaultStopCursor)
+ *     .build();
+ * }</pre>
+ *
+ * <p>See {@link PulsarSourceBuilder} for more details.
+ *
+ * @param <IN> The input type of the pulsar {@link Message}.
+ * @param <OUT> The output type of the source.
+ */
+@PublicEvolving
+public final class PulsarSource<IN, OUT>
+        implements Source<OUT, PulsarPartitionSplit, PulsarSourceEnumState>,
+                ResultTypeQueryable<OUT> {
+    private static final long serialVersionUID = 7773108631275567433L;
+
+    /**
+     * The common configuration for pulsar source, we don't support the pulsar's configuration class
+     * directly.
+     */
+    private final Configuration configuration;
+
+    private final SourceConfiguration sourceConfiguration;
+
+    private final PulsarSubscriber subscriber;
+
+    private final RangeGenerator rangeGenerator;
+
+    private final SerializableSupplier<StartCursor> startCursorSupplier;
+
+    private final SerializableSupplier<StopCursor> stopCursorSupplier;
+
+    /**
+     * Boundedness for source, we only support {@link Boundedness#CONTINUOUS_UNBOUNDED} currently.
+     */
+    private final Boundedness boundedness;
+
+    /** The pulsar deserialization schema used for deserialize message. */
+    private final PulsarDeserializationSchema<IN, OUT> deserializationSchema;
+
+    /** Modify the flink generated {@link ClientConfigurationData}. */
+    private final ConfigurationDataCustomizer<ClientConfigurationData>
+            clientConfigurationCustomizer;
+
+    /** Modify the flink generated {@link ConsumerConfigurationData}. */
+    private final ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+            consumerConfigurationCustomizer;
+
+    /** A lazy evaluated pulsar client for split reader. */
+    private transient volatile PulsarClient pulsarClient;
+
+    /** A lazy evaluated pulsar admin for source enumerator. */
+    private transient volatile PulsarAdmin pulsarAdmin;

Review comment:
       `volatile` and synchronization is unnecessary. All methods here will be invokved only from the main thread. In fact, all `create` methods are only invoked once; so you don't even need to cache the clients here.

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSource.java
##########
@@ -0,0 +1,265 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.java.ClosureCleaner;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumState;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumStateSerializer;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumerator;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.PulsarSourceReaderFactory;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarSchemaInitializationContext;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitSerializer;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.createAdmin;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.createClient;
+import static org.apache.flink.connector.pulsar.common.exception.PulsarExceptionUtils.sneaky;
+
+/**
+ * The Source implementation of Pulsar. Please use a {@link PulsarSourceBuilder} to construct a
+ * {@link PulsarSource}. The following example shows how to create a PulsarSource emitting records
+ * of <code>String</code> type.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource.<byte[], String>builder()
+ *     .setTopics(TOPIC1， TOPIC2)
+ *     .setServiceUrl(getServiceUrl())
+ *     .setAdminUrl(getAdminUrl())
+ *     .setSubscriptionName("test")
+ *     .setDeserializationSchema(flinkSchema(new SimpleStringSchema()))
+ *     .setBounded(StopCursor::defaultStopCursor)
+ *     .build();
+ * }</pre>
+ *
+ * <p>See {@link PulsarSourceBuilder} for more details.
+ *
+ * @param <IN> The input type of the pulsar {@link Message}.
+ * @param <OUT> The output type of the source.
+ */
+@PublicEvolving
+public final class PulsarSource<IN, OUT>
+        implements Source<OUT, PulsarPartitionSplit, PulsarSourceEnumState>,
+                ResultTypeQueryable<OUT> {
+    private static final long serialVersionUID = 7773108631275567433L;
+
+    /**
+     * The common configuration for pulsar source, we don't support the pulsar's configuration class
+     * directly.
+     */
+    private final Configuration configuration;
+
+    private final SourceConfiguration sourceConfiguration;
+
+    private final PulsarSubscriber subscriber;
+
+    private final RangeGenerator rangeGenerator;
+
+    private final SerializableSupplier<StartCursor> startCursorSupplier;
+
+    private final SerializableSupplier<StopCursor> stopCursorSupplier;
+
+    /**
+     * Boundedness for source, we only support {@link Boundedness#CONTINUOUS_UNBOUNDED} currently.
+     */
+    private final Boundedness boundedness;
+
+    /** The pulsar deserialization schema used for deserialize message. */
+    private final PulsarDeserializationSchema<IN, OUT> deserializationSchema;
+
+    /** Modify the flink generated {@link ClientConfigurationData}. */
+    private final ConfigurationDataCustomizer<ClientConfigurationData>
+            clientConfigurationCustomizer;
+
+    /** Modify the flink generated {@link ConsumerConfigurationData}. */
+    private final ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+            consumerConfigurationCustomizer;
+
+    /** A lazy evaluated pulsar client for split reader. */
+    private transient volatile PulsarClient pulsarClient;
+
+    /** A lazy evaluated pulsar admin for source enumerator. */
+    private transient volatile PulsarAdmin pulsarAdmin;
+
+    /**
+     * The constructor for PulsarSource, it's package protected for forcing using {@link
+     * PulsarSourceBuilder}.
+     */
+    PulsarSource(
+            Configuration configuration,
+            PulsarSubscriber subscriber,
+            RangeGenerator rangeGenerator,
+            SerializableSupplier<StartCursor> startCursorSupplier,
+            SerializableSupplier<StopCursor> stopCursorSupplier,
+            Boundedness boundedness,
+            PulsarDeserializationSchema<IN, OUT> deserializationSchema,
+            ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer,
+            ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+                    consumerConfigurationCustomizer) {
+        // Since these implementation could be a lambda, make sure they are serializable.
+        ClosureCleaner.clean(
+                startCursorSupplier, ExecutionConfig.ClosureCleanerLevel.RECURSIVE, true);
+        ClosureCleaner.clean(
+                stopCursorSupplier, ExecutionConfig.ClosureCleanerLevel.RECURSIVE, true);
+        ClosureCleaner.clean(
+                clientConfigurationCustomizer, ExecutionConfig.ClosureCleanerLevel.RECURSIVE, true);
+        ClosureCleaner.clean(
+                consumerConfigurationCustomizer,
+                ExecutionConfig.ClosureCleanerLevel.RECURSIVE,
+                true);
+
+        this.configuration = configuration;
+        this.sourceConfiguration = new SourceConfiguration(configuration);
+        this.subscriber = subscriber;
+        this.rangeGenerator = rangeGenerator;
+        this.startCursorSupplier = startCursorSupplier;
+        this.stopCursorSupplier = stopCursorSupplier;
+        this.boundedness = boundedness;
+        this.deserializationSchema = deserializationSchema;
+        this.clientConfigurationCustomizer = clientConfigurationCustomizer;
+        this.consumerConfigurationCustomizer = consumerConfigurationCustomizer;
+    }
+
+    /**
+     * Get a PulsarSourceBuilder to builder a {@link PulsarSource}.
+     *
+     * @return a Pulsar source builder.
+     */
+    public static <A, B> PulsarSourceBuilder<A, B> builder() {
+        return new PulsarSourceBuilder<>();
+    }
+
+    @Override
+    public Boundedness getBoundedness() {
+        return boundedness;
+    }
+
+    @Override
+    public SourceReader<OUT, PulsarPartitionSplit> createReader(SourceReaderContext readerContext)
+            throws Exception {
+        // Initialize the deserialization schema before creating the pulsar reader.
+        deserializationSchema.open(new PulsarSchemaInitializationContext(readerContext));
+
+        return PulsarSourceReaderFactory.create(
+                readerContext,
+                deserializationSchema,
+                getPulsarClient(),
+                getPulsarAdmin(),
+                configuration,
+                sourceConfiguration,
+                consumerConfigurationCustomizer);
+    }
+
+    @Override
+    public SplitEnumerator<PulsarPartitionSplit, PulsarSourceEnumState> createEnumerator(
+            SplitEnumeratorContext<PulsarPartitionSplit> enumContext) {
+        return new PulsarSourceEnumerator(
+                getPulsarAdmin(),
+                subscriber,
+                rangeGenerator,
+                startCursorSupplier,
+                stopCursorSupplier,
+                configuration,
+                sourceConfiguration,
+                enumContext);
+    }
+
+    @Override
+    public SplitEnumerator<PulsarPartitionSplit, PulsarSourceEnumState> restoreEnumerator(
+            SplitEnumeratorContext<PulsarPartitionSplit> enumContext,
+            PulsarSourceEnumState checkpoint) {
+        return new PulsarSourceEnumerator(
+                getPulsarAdmin(),
+                subscriber,
+                rangeGenerator,
+                startCursorSupplier,
+                stopCursorSupplier,
+                configuration,
+                sourceConfiguration,
+                enumContext,
+                checkpoint);
+    }
+
+    @Override
+    public SimpleVersionedSerializer<PulsarPartitionSplit> getSplitSerializer() {
+        return new PulsarPartitionSplitSerializer();
+    }
+
+    @Override
+    public SimpleVersionedSerializer<PulsarSourceEnumState> getEnumeratorCheckpointSerializer() {
+        return new PulsarSourceEnumStateSerializer();
+    }
+
+    @Override
+    public TypeInformation<OUT> getProducedType() {
+        return deserializationSchema.getProducedType();
+    }
+
+    // ------------- private helpers  --------------
+
+    private PulsarClient getPulsarClient() {
+        if (pulsarClient == null) {
+            synchronized (this) {
+                if (pulsarClient == null) {
+                    pulsarClient = createClient(configuration, clientConfigurationCustomizer);
+                }
+            }
+        }
+
+        return pulsarClient;
+    }
+
+    private PulsarAdmin getPulsarAdmin() {
+        if (pulsarAdmin == null) {
+            synchronized (this) {
+                if (pulsarAdmin == null) {
+                    pulsarAdmin =
+                            sneaky(() -> createAdmin(configuration, clientConfigurationCustomizer));
+                }
+            }
+        }
+
+        return pulsarAdmin;
+    }

Review comment:
       This can be removed as they will only be called once from the main thread.

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
##########
@@ -0,0 +1,575 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.common.exception.PulsarRuntimeException;
+import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.DivideRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FullRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.RegexSubscriptionMode;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.function.Consumer;
+import java.util.regex.Pattern;
+
+import static java.lang.Boolean.FALSE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.getOptionValue;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ENABLE_AUTO_COMMIT_CURSOR;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_REGEX_SUBSCRIPTION_MODE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPICS_PATTERN;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPIC_NAMES;
+import static org.apache.flink.connector.pulsar.source.config.PulsarSourceConfigUtils.checkConfigurations;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The builder class for {@link PulsarSource} to make it easier for the users to construct a {@link
+ * PulsarSource}.
+ *
+ * <p>The following example shows the minimum setup to create a PulsarSource that reads the String
+ * values from a Pulsar topic.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .build();
+ * }</pre>
+ *
+ * <p>The service url, admin url, subscription name, topics to consume, and the record deserializer
+ * are required fields that must be set.
+ *
+ * <p>To specify the starting position of PulsarSource, one can call {@link
+ * #setStartCursorSupplier(SerializableSupplier)}.
+ *
+ * <p>By default the PulsarSource runs in an {@link Boundedness#CONTINUOUS_UNBOUNDED} mode and never
+ * stop until the Flink job is canceled or fails. To let the PulsarSource run in {@link
+ * Boundedness#CONTINUOUS_UNBOUNDED} but stops at some given offsets, one can call {@link
+ * #setUnbounded(SerializableSupplier)}. For example the following PulsarSource stops after it
+ * consumes up to a event time when the Flink started.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .setUnbounded(StopCursor.atEventTime(System.currentTimeMillis()))
+ *     .build();
+ * }</pre>
+ *
+ * @param <IN> The input type of the pulsar {@link Message <?>}
+ * @param <OUT> The output type of the source.
+ */
+@PublicEvolving
+public final class PulsarSourceBuilder<IN, OUT> {
+    private static final Logger LOG = LoggerFactory.getLogger(PulsarSourceBuilder.class);
+
+    private final Configuration configuration;
+    private PulsarSubscriber subscriber;
+    private RangeGenerator rangeGenerator;
+    private SerializableSupplier<StartCursor> startCursorSupplier;
+    private SerializableSupplier<StopCursor> stopCursorSupplier;
+    private Boundedness boundedness;
+    private PulsarDeserializationSchema<IN, OUT> deserializationSchema;
+    private ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer;
+    private ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+            consumerConfigurationCustomizer;
+
+    // private builder constructor.
+    PulsarSourceBuilder() {
+        // The default configuration holder.
+        this.configuration = new Configuration();
+    }
+
+    /**
+     * Sets the admin endpoint for the PulsarAdmin of the PulsarSource.
+     *
+     * @param adminUrl the url for the PulsarAdmin.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setAdminUrl(String adminUrl) {
+        configuration.set(PULSAR_ADMIN_URL, adminUrl);
+        return this;
+    }
+
+    /**
+     * Sets the server's link for the PulsarConsumer of the PulsarSource.
+     *
+     * @param serviceUrl the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setServiceUrl(String serviceUrl) {
+        configuration.set(PULSAR_SERVICE_URL, serviceUrl);
+        return this;
+    }
+
+    /**
+     * Sets the name for this pulsar subscription.
+     *
+     * @param subscriptionName the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionName(String subscriptionName) {
+        configuration.set(PULSAR_SUBSCRIPTION_NAME, subscriptionName);
+        return this;
+    }
+
+    /**
+     * {@link SubscriptionType} is the consuming behavior for pulsar, we would generator different
+     * split by the given subscription type, it's required and quite important for end-user.
+     *
+     * @param subscriptionType The type of subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionType(SubscriptionType subscriptionType) {
+        configuration.set(PULSAR_SUBSCRIPTION_TYPE, subscriptionType);
+        return this;
+    }
+
+    /**
+     * Set a pulsar topic list for flink source. Some topic may not exist currently, consuming this
+     * non-existed topic wouldn't throw any exception. But the best solution is just consuming by
+     * using a topic regex. This method is conflict with {@code setTopicPattern}.

Review comment:
       ```suggestion
        * using a topic regex. You can set topics once either with {@setTopics} or {@code setTopicPattern} in this builder.
   ```

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSource.java
##########
@@ -0,0 +1,265 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.java.ClosureCleaner;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumState;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumStateSerializer;
+import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumerator;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.PulsarSourceReaderFactory;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarSchemaInitializationContext;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitSerializer;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.createAdmin;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.createClient;
+import static org.apache.flink.connector.pulsar.common.exception.PulsarExceptionUtils.sneaky;
+
+/**
+ * The Source implementation of Pulsar. Please use a {@link PulsarSourceBuilder} to construct a
+ * {@link PulsarSource}. The following example shows how to create a PulsarSource emitting records
+ * of <code>String</code> type.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource.<byte[], String>builder()
+ *     .setTopics(TOPIC1， TOPIC2)
+ *     .setServiceUrl(getServiceUrl())
+ *     .setAdminUrl(getAdminUrl())
+ *     .setSubscriptionName("test")
+ *     .setDeserializationSchema(flinkSchema(new SimpleStringSchema()))
+ *     .setBounded(StopCursor::defaultStopCursor)
+ *     .build();
+ * }</pre>
+ *
+ * <p>See {@link PulsarSourceBuilder} for more details.
+ *
+ * @param <IN> The input type of the pulsar {@link Message}.
+ * @param <OUT> The output type of the source.
+ */
+@PublicEvolving
+public final class PulsarSource<IN, OUT>
+        implements Source<OUT, PulsarPartitionSplit, PulsarSourceEnumState>,
+                ResultTypeQueryable<OUT> {
+    private static final long serialVersionUID = 7773108631275567433L;
+
+    /**
+     * The common configuration for pulsar source, we don't support the pulsar's configuration class
+     * directly.
+     */
+    private final Configuration configuration;
+
+    private final SourceConfiguration sourceConfiguration;
+
+    private final PulsarSubscriber subscriber;
+
+    private final RangeGenerator rangeGenerator;
+
+    private final SerializableSupplier<StartCursor> startCursorSupplier;
+
+    private final SerializableSupplier<StopCursor> stopCursorSupplier;
+
+    /**
+     * Boundedness for source, we only support {@link Boundedness#CONTINUOUS_UNBOUNDED} currently.
+     */
+    private final Boundedness boundedness;
+
+    /** The pulsar deserialization schema used for deserialize message. */
+    private final PulsarDeserializationSchema<IN, OUT> deserializationSchema;
+
+    /** Modify the flink generated {@link ClientConfigurationData}. */
+    private final ConfigurationDataCustomizer<ClientConfigurationData>
+            clientConfigurationCustomizer;
+
+    /** Modify the flink generated {@link ConsumerConfigurationData}. */
+    private final ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+            consumerConfigurationCustomizer;
+
+    /** A lazy evaluated pulsar client for split reader. */
+    private transient volatile PulsarClient pulsarClient;
+
+    /** A lazy evaluated pulsar admin for source enumerator. */
+    private transient volatile PulsarAdmin pulsarAdmin;
+
+    /**
+     * The constructor for PulsarSource, it's package protected for forcing using {@link
+     * PulsarSourceBuilder}.
+     */
+    PulsarSource(
+            Configuration configuration,
+            PulsarSubscriber subscriber,
+            RangeGenerator rangeGenerator,
+            SerializableSupplier<StartCursor> startCursorSupplier,
+            SerializableSupplier<StopCursor> stopCursorSupplier,
+            Boundedness boundedness,
+            PulsarDeserializationSchema<IN, OUT> deserializationSchema,
+            ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer,
+            ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+                    consumerConfigurationCustomizer) {
+        // Since these implementation could be a lambda, make sure they are serializable.
+        ClosureCleaner.clean(
+                startCursorSupplier, ExecutionConfig.ClosureCleanerLevel.RECURSIVE, true);
+        ClosureCleaner.clean(
+                stopCursorSupplier, ExecutionConfig.ClosureCleanerLevel.RECURSIVE, true);
+        ClosureCleaner.clean(
+                clientConfigurationCustomizer, ExecutionConfig.ClosureCleanerLevel.RECURSIVE, true);
+        ClosureCleaner.clean(
+                consumerConfigurationCustomizer,
+                ExecutionConfig.ClosureCleanerLevel.RECURSIVE,
+                true);
+
+        this.configuration = configuration;
+        this.sourceConfiguration = new SourceConfiguration(configuration);
+        this.subscriber = subscriber;
+        this.rangeGenerator = rangeGenerator;
+        this.startCursorSupplier = startCursorSupplier;
+        this.stopCursorSupplier = stopCursorSupplier;
+        this.boundedness = boundedness;
+        this.deserializationSchema = deserializationSchema;
+        this.clientConfigurationCustomizer = clientConfigurationCustomizer;
+        this.consumerConfigurationCustomizer = consumerConfigurationCustomizer;
+    }
+
+    /**
+     * Get a PulsarSourceBuilder to builder a {@link PulsarSource}.
+     *
+     * @return a Pulsar source builder.
+     */
+    public static <A, B> PulsarSourceBuilder<A, B> builder() {

Review comment:
       <IN, OUT>

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
##########
@@ -0,0 +1,575 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.common.exception.PulsarRuntimeException;
+import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.DivideRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FullRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.RegexSubscriptionMode;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.function.Consumer;
+import java.util.regex.Pattern;
+
+import static java.lang.Boolean.FALSE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.getOptionValue;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ENABLE_AUTO_COMMIT_CURSOR;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_REGEX_SUBSCRIPTION_MODE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPICS_PATTERN;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPIC_NAMES;
+import static org.apache.flink.connector.pulsar.source.config.PulsarSourceConfigUtils.checkConfigurations;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The builder class for {@link PulsarSource} to make it easier for the users to construct a {@link
+ * PulsarSource}.
+ *
+ * <p>The following example shows the minimum setup to create a PulsarSource that reads the String
+ * values from a Pulsar topic.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .build();
+ * }</pre>
+ *
+ * <p>The service url, admin url, subscription name, topics to consume, and the record deserializer
+ * are required fields that must be set.
+ *
+ * <p>To specify the starting position of PulsarSource, one can call {@link
+ * #setStartCursorSupplier(SerializableSupplier)}.
+ *
+ * <p>By default the PulsarSource runs in an {@link Boundedness#CONTINUOUS_UNBOUNDED} mode and never
+ * stop until the Flink job is canceled or fails. To let the PulsarSource run in {@link
+ * Boundedness#CONTINUOUS_UNBOUNDED} but stops at some given offsets, one can call {@link
+ * #setUnbounded(SerializableSupplier)}. For example the following PulsarSource stops after it
+ * consumes up to a event time when the Flink started.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .setUnbounded(StopCursor.atEventTime(System.currentTimeMillis()))
+ *     .build();
+ * }</pre>
+ *
+ * @param <IN> The input type of the pulsar {@link Message <?>}
+ * @param <OUT> The output type of the source.
+ */
+@PublicEvolving
+public final class PulsarSourceBuilder<IN, OUT> {
+    private static final Logger LOG = LoggerFactory.getLogger(PulsarSourceBuilder.class);
+
+    private final Configuration configuration;
+    private PulsarSubscriber subscriber;
+    private RangeGenerator rangeGenerator;
+    private SerializableSupplier<StartCursor> startCursorSupplier;
+    private SerializableSupplier<StopCursor> stopCursorSupplier;
+    private Boundedness boundedness;
+    private PulsarDeserializationSchema<IN, OUT> deserializationSchema;
+    private ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer;
+    private ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+            consumerConfigurationCustomizer;
+
+    // private builder constructor.
+    PulsarSourceBuilder() {
+        // The default configuration holder.
+        this.configuration = new Configuration();
+    }
+
+    /**
+     * Sets the admin endpoint for the PulsarAdmin of the PulsarSource.
+     *
+     * @param adminUrl the url for the PulsarAdmin.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setAdminUrl(String adminUrl) {
+        configuration.set(PULSAR_ADMIN_URL, adminUrl);
+        return this;
+    }
+
+    /**
+     * Sets the server's link for the PulsarConsumer of the PulsarSource.
+     *
+     * @param serviceUrl the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setServiceUrl(String serviceUrl) {
+        configuration.set(PULSAR_SERVICE_URL, serviceUrl);
+        return this;
+    }
+
+    /**
+     * Sets the name for this pulsar subscription.
+     *
+     * @param subscriptionName the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionName(String subscriptionName) {
+        configuration.set(PULSAR_SUBSCRIPTION_NAME, subscriptionName);
+        return this;
+    }
+
+    /**
+     * {@link SubscriptionType} is the consuming behavior for pulsar, we would generator different
+     * split by the given subscription type, it's required and quite important for end-user.
+     *
+     * @param subscriptionType The type of subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionType(SubscriptionType subscriptionType) {
+        configuration.set(PULSAR_SUBSCRIPTION_TYPE, subscriptionType);
+        return this;
+    }
+
+    /**
+     * Set a pulsar topic list for flink source. Some topic may not exist currently, consuming this
+     * non-existed topic wouldn't throw any exception. But the best solution is just consuming by
+     * using a topic regex. This method is conflict with {@code setTopicPattern}.
+     *
+     * @param topics The topic list you would like to consume message.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopics(String... topics) {
+        return setTopics(Arrays.asList(topics));
+    }
+
+    /**
+     * Set a pulsar topic list for flink source. Some topic may not exist currently, consuming this
+     * non-existed topic wouldn't throw any exception. But the best solution is just consuming by
+     * using a topic regex. This method is conflict with {@code setTopicPattern}.
+     *
+     * @param topics The topic list you would like to consume message.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopics(List<String> topics) {
+        ensureSubscriberIsNull("topics");
+        configuration.set(PULSAR_TOPIC_NAMES, PulsarJsonUtils.toString(topics));
+
+        this.subscriber = PulsarSubscriber.getTopicListSubscriber(topics);
+
+        return this;
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(String topicsPattern) {
+        return setTopicPattern(Pattern.compile(topicsPattern));
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(Pattern topicsPattern) {
+        return setTopicPattern(topicsPattern, RegexSubscriptionMode.AllTopics);
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @param regexSubscriptionMode The topic filter for regex subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(
+            String topicsPattern, RegexSubscriptionMode regexSubscriptionMode) {
+        return setTopicPattern(Pattern.compile(topicsPattern), regexSubscriptionMode);
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @param regexSubscriptionMode The topic filter for regex subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(
+            Pattern topicsPattern, RegexSubscriptionMode regexSubscriptionMode) {
+        ensureSubscriberIsNull("topic pattern");
+        configuration.set(PULSAR_TOPICS_PATTERN, topicsPattern.toString());
+        this.subscriber =
+                PulsarSubscriber.getTopicPatternSubscriber(topicsPattern, regexSubscriptionMode);
+
+        return setRegexSubscriptionMode(regexSubscriptionMode);
+    }
+
+    /**
+     * When subscribing to a topic using a regular expression, you can pick a certain type of
+     * topics.
+     *
+     * @param regexSubscriptionMode The topic type.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setRegexSubscriptionMode(
+            RegexSubscriptionMode regexSubscriptionMode) {
+        configuration.set(PULSAR_REGEX_SUBSCRIPTION_MODE, regexSubscriptionMode);
+        return this;
+    }
+
+    /**
+     * Set a topic range generator for Key_Shared subscription.
+     *
+     * @param rangeGenerator A generator which would generate a set of {@link TopicRange} for given
+     *     topic.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setRangeGenerator(RangeGenerator rangeGenerator) {
+        this.rangeGenerator = rangeGenerator;
+        if (configuration.contains(PULSAR_SUBSCRIPTION_TYPE)) {
+            SubscriptionType subscriptionType = configuration.get(PULSAR_SUBSCRIPTION_TYPE);
+            if (subscriptionType != SubscriptionType.Key_Shared) {
+                LOG.warn(
+                        "Key_Shared subscription should be used for custom rangeGenerator instead of {}",
+                        subscriptionType);
+            }
+        }
+
+        return this;
+    }
+
+    /**
+     * Specify from which offsets the PulsarSource should start consume from by providing an {@link
+     * StartCursor}.
+     *
+     * @param startCursorSupplier the supplier providing a {@link StartCursor} which set the
+     *     starting offsets for the Source.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setStartCursorSupplier(
+            SerializableSupplier<StartCursor> startCursorSupplier) {
+        this.startCursorSupplier = startCursorSupplier;
+        return this;
+    }
+
+    /**
+     * By default the PulsarSource is set to run in {@link Boundedness#CONTINUOUS_UNBOUNDED} manner
+     * and thus never stops until the Flink job fails or is canceled. To let the PulsarSource run as
+     * a streaming source but still stops at some point, one can set an {@link StartCursor} to
+     * specify the stopping offsets for each partition. When all the partitions have reached their
+     * stopping offsets, the PulsarSource will then exit.
+     *
+     * <p>This method is different from {@link #setBounded(SerializableSupplier)} that after setting
+     * the stopping offsets with this method, {@link PulsarSource#getBoundedness()} will still
+     * return {@link Boundedness#CONTINUOUS_UNBOUNDED} even though it will stop at the stopping
+     * offsets specified by the stopping offsets {@link StartCursor}.
+     *
+     * @param stopCursorSupplier The {@link StopCursor} to specify the stopping offset.
+     * @return this PulsarSourceBuilder.
+     * @see #setBounded(SerializableSupplier)
+     */
+    public PulsarSourceBuilder<IN, OUT> setUnbounded(
+            SerializableSupplier<StopCursor> stopCursorSupplier) {
+        this.boundedness = Boundedness.CONTINUOUS_UNBOUNDED;
+        this.stopCursorSupplier = stopCursorSupplier;
+        return this;
+    }
+
+    /**
+     * By default the PulsarSource is set to run in {@link Boundedness#CONTINUOUS_UNBOUNDED} manner
+     * and thus never stops until the Flink job fails or is canceled. To let the PulsarSource run in
+     * {@link Boundedness#BOUNDED} manner and stops at some point, one can set an {@link
+     * StartCursor} to specify the stopping offsets for each partition. When all the partitions have

Review comment:
       ```suggestion
        * StopCursor} to specify the stopping offsets for each partition. When all the partitions have
   ```

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/CursorVerification.java
##########
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source.config;
+
+/** The enum class for defining the cursor verify behavior. */
+public enum CursorVerification {

Review comment:
       `@PublicEvolving`?

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumState.java
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source.enumerator;
+
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
+
+import java.io.Serializable;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The state class for pulsar source enumerator, used for storing the split state. This class is
+ * managed and controlled by {@link SplitsAssignmentState}.
+ */
+public class PulsarSourceEnumState implements Serializable {
+    private static final long serialVersionUID = -1228111807064171730L;
+
+    private final Set<TopicPartition> appendedPartitions;
+
+    private final Set<PulsarPartitionSplit> pendingPartitionSplits;
+
+    private final Map<String, Set<Integer>> splitReaderMapping;

Review comment:
       It would be nice to have a short description of the fields here.

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
##########
@@ -0,0 +1,575 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.common.exception.PulsarRuntimeException;
+import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.DivideRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FullRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.RegexSubscriptionMode;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.function.Consumer;
+import java.util.regex.Pattern;
+
+import static java.lang.Boolean.FALSE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.getOptionValue;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ENABLE_AUTO_COMMIT_CURSOR;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_REGEX_SUBSCRIPTION_MODE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPICS_PATTERN;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPIC_NAMES;
+import static org.apache.flink.connector.pulsar.source.config.PulsarSourceConfigUtils.checkConfigurations;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The builder class for {@link PulsarSource} to make it easier for the users to construct a {@link
+ * PulsarSource}.
+ *
+ * <p>The following example shows the minimum setup to create a PulsarSource that reads the String
+ * values from a Pulsar topic.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .build();
+ * }</pre>
+ *
+ * <p>The service url, admin url, subscription name, topics to consume, and the record deserializer
+ * are required fields that must be set.
+ *
+ * <p>To specify the starting position of PulsarSource, one can call {@link
+ * #setStartCursorSupplier(SerializableSupplier)}.
+ *
+ * <p>By default the PulsarSource runs in an {@link Boundedness#CONTINUOUS_UNBOUNDED} mode and never
+ * stop until the Flink job is canceled or fails. To let the PulsarSource run in {@link
+ * Boundedness#CONTINUOUS_UNBOUNDED} but stops at some given offsets, one can call {@link
+ * #setUnbounded(SerializableSupplier)}. For example the following PulsarSource stops after it
+ * consumes up to a event time when the Flink started.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .setUnbounded(StopCursor.atEventTime(System.currentTimeMillis()))
+ *     .build();
+ * }</pre>
+ *
+ * @param <IN> The input type of the pulsar {@link Message <?>}
+ * @param <OUT> The output type of the source.
+ */
+@PublicEvolving
+public final class PulsarSourceBuilder<IN, OUT> {
+    private static final Logger LOG = LoggerFactory.getLogger(PulsarSourceBuilder.class);
+
+    private final Configuration configuration;
+    private PulsarSubscriber subscriber;
+    private RangeGenerator rangeGenerator;
+    private SerializableSupplier<StartCursor> startCursorSupplier;
+    private SerializableSupplier<StopCursor> stopCursorSupplier;
+    private Boundedness boundedness;
+    private PulsarDeserializationSchema<IN, OUT> deserializationSchema;
+    private ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer;
+    private ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+            consumerConfigurationCustomizer;
+
+    // private builder constructor.
+    PulsarSourceBuilder() {
+        // The default configuration holder.
+        this.configuration = new Configuration();
+    }
+
+    /**
+     * Sets the admin endpoint for the PulsarAdmin of the PulsarSource.
+     *
+     * @param adminUrl the url for the PulsarAdmin.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setAdminUrl(String adminUrl) {
+        configuration.set(PULSAR_ADMIN_URL, adminUrl);
+        return this;
+    }
+
+    /**
+     * Sets the server's link for the PulsarConsumer of the PulsarSource.
+     *
+     * @param serviceUrl the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setServiceUrl(String serviceUrl) {
+        configuration.set(PULSAR_SERVICE_URL, serviceUrl);
+        return this;
+    }
+
+    /**
+     * Sets the name for this pulsar subscription.
+     *
+     * @param subscriptionName the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionName(String subscriptionName) {
+        configuration.set(PULSAR_SUBSCRIPTION_NAME, subscriptionName);
+        return this;
+    }
+
+    /**
+     * {@link SubscriptionType} is the consuming behavior for pulsar, we would generator different
+     * split by the given subscription type, it's required and quite important for end-user.
+     *
+     * @param subscriptionType The type of subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionType(SubscriptionType subscriptionType) {
+        configuration.set(PULSAR_SUBSCRIPTION_TYPE, subscriptionType);
+        return this;
+    }
+
+    /**
+     * Set a pulsar topic list for flink source. Some topic may not exist currently, consuming this
+     * non-existed topic wouldn't throw any exception. But the best solution is just consuming by
+     * using a topic regex. This method is conflict with {@code setTopicPattern}.
+     *
+     * @param topics The topic list you would like to consume message.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopics(String... topics) {
+        return setTopics(Arrays.asList(topics));
+    }
+
+    /**
+     * Set a pulsar topic list for flink source. Some topic may not exist currently, consuming this
+     * non-existed topic wouldn't throw any exception. But the best solution is just consuming by
+     * using a topic regex. This method is conflict with {@code setTopicPattern}.
+     *
+     * @param topics The topic list you would like to consume message.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopics(List<String> topics) {
+        ensureSubscriberIsNull("topics");
+        configuration.set(PULSAR_TOPIC_NAMES, PulsarJsonUtils.toString(topics));
+
+        this.subscriber = PulsarSubscriber.getTopicListSubscriber(topics);
+
+        return this;
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(String topicsPattern) {
+        return setTopicPattern(Pattern.compile(topicsPattern));
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(Pattern topicsPattern) {
+        return setTopicPattern(topicsPattern, RegexSubscriptionMode.AllTopics);
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @param regexSubscriptionMode The topic filter for regex subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(
+            String topicsPattern, RegexSubscriptionMode regexSubscriptionMode) {
+        return setTopicPattern(Pattern.compile(topicsPattern), regexSubscriptionMode);
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @param regexSubscriptionMode The topic filter for regex subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(
+            Pattern topicsPattern, RegexSubscriptionMode regexSubscriptionMode) {
+        ensureSubscriberIsNull("topic pattern");
+        configuration.set(PULSAR_TOPICS_PATTERN, topicsPattern.toString());
+        this.subscriber =
+                PulsarSubscriber.getTopicPatternSubscriber(topicsPattern, regexSubscriptionMode);
+
+        return setRegexSubscriptionMode(regexSubscriptionMode);
+    }
+
+    /**
+     * When subscribing to a topic using a regular expression, you can pick a certain type of
+     * topics.
+     *
+     * @param regexSubscriptionMode The topic type.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setRegexSubscriptionMode(
+            RegexSubscriptionMode regexSubscriptionMode) {
+        configuration.set(PULSAR_REGEX_SUBSCRIPTION_MODE, regexSubscriptionMode);
+        return this;
+    }
+
+    /**
+     * Set a topic range generator for Key_Shared subscription.
+     *
+     * @param rangeGenerator A generator which would generate a set of {@link TopicRange} for given
+     *     topic.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setRangeGenerator(RangeGenerator rangeGenerator) {
+        this.rangeGenerator = rangeGenerator;
+        if (configuration.contains(PULSAR_SUBSCRIPTION_TYPE)) {
+            SubscriptionType subscriptionType = configuration.get(PULSAR_SUBSCRIPTION_TYPE);
+            if (subscriptionType != SubscriptionType.Key_Shared) {
+                LOG.warn(
+                        "Key_Shared subscription should be used for custom rangeGenerator instead of {}",
+                        subscriptionType);
+            }
+        }
+
+        return this;
+    }
+
+    /**
+     * Specify from which offsets the PulsarSource should start consume from by providing an {@link
+     * StartCursor}.
+     *
+     * @param startCursorSupplier the supplier providing a {@link StartCursor} which set the
+     *     starting offsets for the Source.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setStartCursorSupplier(
+            SerializableSupplier<StartCursor> startCursorSupplier) {
+        this.startCursorSupplier = startCursorSupplier;
+        return this;
+    }
+
+    /**
+     * By default the PulsarSource is set to run in {@link Boundedness#CONTINUOUS_UNBOUNDED} manner
+     * and thus never stops until the Flink job fails or is canceled. To let the PulsarSource run as
+     * a streaming source but still stops at some point, one can set an {@link StartCursor} to
+     * specify the stopping offsets for each partition. When all the partitions have reached their
+     * stopping offsets, the PulsarSource will then exit.
+     *
+     * <p>This method is different from {@link #setBounded(SerializableSupplier)} that after setting
+     * the stopping offsets with this method, {@link PulsarSource#getBoundedness()} will still
+     * return {@link Boundedness#CONTINUOUS_UNBOUNDED} even though it will stop at the stopping
+     * offsets specified by the stopping offsets {@link StartCursor}.
+     *
+     * @param stopCursorSupplier The {@link StopCursor} to specify the stopping offset.
+     * @return this PulsarSourceBuilder.
+     * @see #setBounded(SerializableSupplier)
+     */
+    public PulsarSourceBuilder<IN, OUT> setUnbounded(

Review comment:
       `setUnboundedStopCursor`?

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
##########
@@ -0,0 +1,525 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.connector.pulsar.source.config.CursorVerification;
+
+import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
+import org.apache.pulsar.client.api.RegexSubscriptionMode;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionMode;
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/**
+ * Configurations for PulsarSource. All these options would be listed in pulsar flink connector
+ * documentation.
+ */
+@PublicEvolving
+public final class PulsarSourceOptions {

Review comment:
       These options are completely distinct from PulsarOptions? Should we add a link in JavaDoc to each other such that user can find options they might have searched at the wrong place?

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
##########
@@ -0,0 +1,575 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.common.exception.PulsarRuntimeException;
+import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.DivideRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FullRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.RegexSubscriptionMode;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.function.Consumer;
+import java.util.regex.Pattern;
+
+import static java.lang.Boolean.FALSE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.getOptionValue;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ENABLE_AUTO_COMMIT_CURSOR;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_REGEX_SUBSCRIPTION_MODE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPICS_PATTERN;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPIC_NAMES;
+import static org.apache.flink.connector.pulsar.source.config.PulsarSourceConfigUtils.checkConfigurations;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The builder class for {@link PulsarSource} to make it easier for the users to construct a {@link
+ * PulsarSource}.
+ *
+ * <p>The following example shows the minimum setup to create a PulsarSource that reads the String
+ * values from a Pulsar topic.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .build();
+ * }</pre>
+ *
+ * <p>The service url, admin url, subscription name, topics to consume, and the record deserializer
+ * are required fields that must be set.
+ *
+ * <p>To specify the starting position of PulsarSource, one can call {@link
+ * #setStartCursorSupplier(SerializableSupplier)}.
+ *
+ * <p>By default the PulsarSource runs in an {@link Boundedness#CONTINUOUS_UNBOUNDED} mode and never
+ * stop until the Flink job is canceled or fails. To let the PulsarSource run in {@link
+ * Boundedness#CONTINUOUS_UNBOUNDED} but stops at some given offsets, one can call {@link
+ * #setUnbounded(SerializableSupplier)}. For example the following PulsarSource stops after it
+ * consumes up to a event time when the Flink started.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .setUnbounded(StopCursor.atEventTime(System.currentTimeMillis()))
+ *     .build();
+ * }</pre>
+ *
+ * @param <IN> The input type of the pulsar {@link Message <?>}
+ * @param <OUT> The output type of the source.
+ */
+@PublicEvolving
+public final class PulsarSourceBuilder<IN, OUT> {
+    private static final Logger LOG = LoggerFactory.getLogger(PulsarSourceBuilder.class);
+
+    private final Configuration configuration;
+    private PulsarSubscriber subscriber;
+    private RangeGenerator rangeGenerator;
+    private SerializableSupplier<StartCursor> startCursorSupplier;
+    private SerializableSupplier<StopCursor> stopCursorSupplier;
+    private Boundedness boundedness;
+    private PulsarDeserializationSchema<IN, OUT> deserializationSchema;
+    private ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer;
+    private ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+            consumerConfigurationCustomizer;
+
+    // private builder constructor.
+    PulsarSourceBuilder() {
+        // The default configuration holder.
+        this.configuration = new Configuration();
+    }
+
+    /**
+     * Sets the admin endpoint for the PulsarAdmin of the PulsarSource.
+     *
+     * @param adminUrl the url for the PulsarAdmin.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setAdminUrl(String adminUrl) {
+        configuration.set(PULSAR_ADMIN_URL, adminUrl);
+        return this;
+    }
+
+    /**
+     * Sets the server's link for the PulsarConsumer of the PulsarSource.
+     *
+     * @param serviceUrl the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setServiceUrl(String serviceUrl) {
+        configuration.set(PULSAR_SERVICE_URL, serviceUrl);
+        return this;
+    }
+
+    /**
+     * Sets the name for this pulsar subscription.
+     *
+     * @param subscriptionName the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionName(String subscriptionName) {
+        configuration.set(PULSAR_SUBSCRIPTION_NAME, subscriptionName);
+        return this;
+    }
+
+    /**
+     * {@link SubscriptionType} is the consuming behavior for pulsar, we would generator different
+     * split by the given subscription type, it's required and quite important for end-user.
+     *
+     * @param subscriptionType The type of subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionType(SubscriptionType subscriptionType) {
+        configuration.set(PULSAR_SUBSCRIPTION_TYPE, subscriptionType);
+        return this;
+    }
+
+    /**
+     * Set a pulsar topic list for flink source. Some topic may not exist currently, consuming this
+     * non-existed topic wouldn't throw any exception. But the best solution is just consuming by
+     * using a topic regex. This method is conflict with {@code setTopicPattern}.
+     *
+     * @param topics The topic list you would like to consume message.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopics(String... topics) {
+        return setTopics(Arrays.asList(topics));
+    }
+
+    /**
+     * Set a pulsar topic list for flink source. Some topic may not exist currently, consuming this
+     * non-existed topic wouldn't throw any exception. But the best solution is just consuming by
+     * using a topic regex. This method is conflict with {@code setTopicPattern}.
+     *
+     * @param topics The topic list you would like to consume message.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopics(List<String> topics) {
+        ensureSubscriberIsNull("topics");
+        configuration.set(PULSAR_TOPIC_NAMES, PulsarJsonUtils.toString(topics));
+
+        this.subscriber = PulsarSubscriber.getTopicListSubscriber(topics);
+
+        return this;
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(String topicsPattern) {
+        return setTopicPattern(Pattern.compile(topicsPattern));
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(Pattern topicsPattern) {
+        return setTopicPattern(topicsPattern, RegexSubscriptionMode.AllTopics);
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @param regexSubscriptionMode The topic filter for regex subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(
+            String topicsPattern, RegexSubscriptionMode regexSubscriptionMode) {
+        return setTopicPattern(Pattern.compile(topicsPattern), regexSubscriptionMode);
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @param regexSubscriptionMode The topic filter for regex subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(
+            Pattern topicsPattern, RegexSubscriptionMode regexSubscriptionMode) {
+        ensureSubscriberIsNull("topic pattern");
+        configuration.set(PULSAR_TOPICS_PATTERN, topicsPattern.toString());
+        this.subscriber =
+                PulsarSubscriber.getTopicPatternSubscriber(topicsPattern, regexSubscriptionMode);
+
+        return setRegexSubscriptionMode(regexSubscriptionMode);
+    }
+
+    /**
+     * When subscribing to a topic using a regular expression, you can pick a certain type of
+     * topics.
+     *
+     * @param regexSubscriptionMode The topic type.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setRegexSubscriptionMode(
+            RegexSubscriptionMode regexSubscriptionMode) {
+        configuration.set(PULSAR_REGEX_SUBSCRIPTION_MODE, regexSubscriptionMode);
+        return this;
+    }
+
+    /**
+     * Set a topic range generator for Key_Shared subscription.
+     *
+     * @param rangeGenerator A generator which would generate a set of {@link TopicRange} for given
+     *     topic.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setRangeGenerator(RangeGenerator rangeGenerator) {
+        this.rangeGenerator = rangeGenerator;
+        if (configuration.contains(PULSAR_SUBSCRIPTION_TYPE)) {
+            SubscriptionType subscriptionType = configuration.get(PULSAR_SUBSCRIPTION_TYPE);
+            if (subscriptionType != SubscriptionType.Key_Shared) {
+                LOG.warn(
+                        "Key_Shared subscription should be used for custom rangeGenerator instead of {}",
+                        subscriptionType);
+            }
+        }
+
+        return this;
+    }
+
+    /**
+     * Specify from which offsets the PulsarSource should start consume from by providing an {@link
+     * StartCursor}.
+     *
+     * @param startCursorSupplier the supplier providing a {@link StartCursor} which set the
+     *     starting offsets for the Source.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setStartCursorSupplier(

Review comment:
       ```suggestion
       public PulsarSourceBuilder<IN, OUT> setStartCursor(
   ```
   ?

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
##########
@@ -0,0 +1,575 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.common.exception.PulsarRuntimeException;
+import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.DivideRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FullRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.RegexSubscriptionMode;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.function.Consumer;
+import java.util.regex.Pattern;
+
+import static java.lang.Boolean.FALSE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.getOptionValue;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ENABLE_AUTO_COMMIT_CURSOR;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_REGEX_SUBSCRIPTION_MODE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPICS_PATTERN;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPIC_NAMES;
+import static org.apache.flink.connector.pulsar.source.config.PulsarSourceConfigUtils.checkConfigurations;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The builder class for {@link PulsarSource} to make it easier for the users to construct a {@link
+ * PulsarSource}.
+ *
+ * <p>The following example shows the minimum setup to create a PulsarSource that reads the String
+ * values from a Pulsar topic.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .build();
+ * }</pre>
+ *
+ * <p>The service url, admin url, subscription name, topics to consume, and the record deserializer
+ * are required fields that must be set.
+ *
+ * <p>To specify the starting position of PulsarSource, one can call {@link
+ * #setStartCursorSupplier(SerializableSupplier)}.
+ *
+ * <p>By default the PulsarSource runs in an {@link Boundedness#CONTINUOUS_UNBOUNDED} mode and never
+ * stop until the Flink job is canceled or fails. To let the PulsarSource run in {@link
+ * Boundedness#CONTINUOUS_UNBOUNDED} but stops at some given offsets, one can call {@link
+ * #setUnbounded(SerializableSupplier)}. For example the following PulsarSource stops after it
+ * consumes up to a event time when the Flink started.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .setUnbounded(StopCursor.atEventTime(System.currentTimeMillis()))
+ *     .build();
+ * }</pre>
+ *
+ * @param <IN> The input type of the pulsar {@link Message <?>}
+ * @param <OUT> The output type of the source.
+ */
+@PublicEvolving
+public final class PulsarSourceBuilder<IN, OUT> {
+    private static final Logger LOG = LoggerFactory.getLogger(PulsarSourceBuilder.class);
+
+    private final Configuration configuration;
+    private PulsarSubscriber subscriber;
+    private RangeGenerator rangeGenerator;
+    private SerializableSupplier<StartCursor> startCursorSupplier;
+    private SerializableSupplier<StopCursor> stopCursorSupplier;
+    private Boundedness boundedness;
+    private PulsarDeserializationSchema<IN, OUT> deserializationSchema;
+    private ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer;
+    private ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+            consumerConfigurationCustomizer;
+
+    // private builder constructor.
+    PulsarSourceBuilder() {
+        // The default configuration holder.
+        this.configuration = new Configuration();
+    }
+
+    /**
+     * Sets the admin endpoint for the PulsarAdmin of the PulsarSource.
+     *
+     * @param adminUrl the url for the PulsarAdmin.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setAdminUrl(String adminUrl) {
+        configuration.set(PULSAR_ADMIN_URL, adminUrl);
+        return this;
+    }
+
+    /**
+     * Sets the server's link for the PulsarConsumer of the PulsarSource.
+     *
+     * @param serviceUrl the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setServiceUrl(String serviceUrl) {
+        configuration.set(PULSAR_SERVICE_URL, serviceUrl);
+        return this;
+    }
+
+    /**
+     * Sets the name for this pulsar subscription.
+     *
+     * @param subscriptionName the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionName(String subscriptionName) {
+        configuration.set(PULSAR_SUBSCRIPTION_NAME, subscriptionName);
+        return this;
+    }
+
+    /**
+     * {@link SubscriptionType} is the consuming behavior for pulsar, we would generator different
+     * split by the given subscription type, it's required and quite important for end-user.
+     *
+     * @param subscriptionType The type of subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionType(SubscriptionType subscriptionType) {
+        configuration.set(PULSAR_SUBSCRIPTION_TYPE, subscriptionType);
+        return this;
+    }
+
+    /**
+     * Set a pulsar topic list for flink source. Some topic may not exist currently, consuming this
+     * non-existed topic wouldn't throw any exception. But the best solution is just consuming by
+     * using a topic regex. This method is conflict with {@code setTopicPattern}.
+     *
+     * @param topics The topic list you would like to consume message.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopics(String... topics) {
+        return setTopics(Arrays.asList(topics));
+    }
+
+    /**
+     * Set a pulsar topic list for flink source. Some topic may not exist currently, consuming this
+     * non-existed topic wouldn't throw any exception. But the best solution is just consuming by
+     * using a topic regex. This method is conflict with {@code setTopicPattern}.
+     *
+     * @param topics The topic list you would like to consume message.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopics(List<String> topics) {
+        ensureSubscriberIsNull("topics");
+        configuration.set(PULSAR_TOPIC_NAMES, PulsarJsonUtils.toString(topics));
+
+        this.subscriber = PulsarSubscriber.getTopicListSubscriber(topics);
+
+        return this;
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(String topicsPattern) {
+        return setTopicPattern(Pattern.compile(topicsPattern));
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(Pattern topicsPattern) {
+        return setTopicPattern(topicsPattern, RegexSubscriptionMode.AllTopics);
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @param regexSubscriptionMode The topic filter for regex subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(
+            String topicsPattern, RegexSubscriptionMode regexSubscriptionMode) {
+        return setTopicPattern(Pattern.compile(topicsPattern), regexSubscriptionMode);
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @param regexSubscriptionMode The topic filter for regex subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(
+            Pattern topicsPattern, RegexSubscriptionMode regexSubscriptionMode) {
+        ensureSubscriberIsNull("topic pattern");
+        configuration.set(PULSAR_TOPICS_PATTERN, topicsPattern.toString());
+        this.subscriber =
+                PulsarSubscriber.getTopicPatternSubscriber(topicsPattern, regexSubscriptionMode);
+
+        return setRegexSubscriptionMode(regexSubscriptionMode);
+    }
+
+    /**
+     * When subscribing to a topic using a regular expression, you can pick a certain type of
+     * topics.
+     *
+     * @param regexSubscriptionMode The topic type.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setRegexSubscriptionMode(
+            RegexSubscriptionMode regexSubscriptionMode) {
+        configuration.set(PULSAR_REGEX_SUBSCRIPTION_MODE, regexSubscriptionMode);
+        return this;
+    }
+
+    /**
+     * Set a topic range generator for Key_Shared subscription.
+     *
+     * @param rangeGenerator A generator which would generate a set of {@link TopicRange} for given
+     *     topic.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setRangeGenerator(RangeGenerator rangeGenerator) {
+        this.rangeGenerator = rangeGenerator;
+        if (configuration.contains(PULSAR_SUBSCRIPTION_TYPE)) {
+            SubscriptionType subscriptionType = configuration.get(PULSAR_SUBSCRIPTION_TYPE);
+            if (subscriptionType != SubscriptionType.Key_Shared) {
+                LOG.warn(
+                        "Key_Shared subscription should be used for custom rangeGenerator instead of {}",
+                        subscriptionType);
+            }
+        }
+
+        return this;
+    }
+
+    /**
+     * Specify from which offsets the PulsarSource should start consume from by providing an {@link
+     * StartCursor}.
+     *
+     * @param startCursorSupplier the supplier providing a {@link StartCursor} which set the
+     *     starting offsets for the Source.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setStartCursorSupplier(
+            SerializableSupplier<StartCursor> startCursorSupplier) {

Review comment:
       Why do we need a serializableSupplier when the StartCursor is already serializable? It feels like an unnecessary indirection.

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/SourceConfiguration.java
##########
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source.config;
+
+import org.apache.flink.configuration.Configuration;
+
+import org.apache.pulsar.client.api.SubscriptionMode;
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.io.Serializable;
+import java.time.Duration;
+
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.getOptionValue;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_AUTO_COMMIT_CURSOR_INTERVAL;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_CLOSE_TIMEOUT_MS;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ENABLE_AUTO_COMMIT_CURSOR;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_MAX_FETCH_RECORDS;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_MAX_FETCH_TIME;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_MODE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_VERIFY_INITIAL_OFFSETS;
+
+/** The configure class for pulsar source. */
+public class SourceConfiguration implements Serializable {

Review comment:
       @Experimental/@PublicEvolving as it's passed to the cursors. Also please consider adding some JavaDocs to non-trivial fields.
   Alternatively, you could design StartCursoer/StopCursor as non-subclassable by the user. For example, you could make them abstract classes with private ctor and add all known implementations to the classes.

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
##########
@@ -0,0 +1,575 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
+import org.apache.flink.connector.pulsar.common.exception.PulsarRuntimeException;
+import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.DivideRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FullRangeGenerator;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.RegexSubscriptionMode;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.function.Consumer;
+import java.util.regex.Pattern;
+
+import static java.lang.Boolean.FALSE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.getOptionValue;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ENABLE_AUTO_COMMIT_CURSOR;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_REGEX_SUBSCRIPTION_MODE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_TYPE;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPICS_PATTERN;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPIC_NAMES;
+import static org.apache.flink.connector.pulsar.source.config.PulsarSourceConfigUtils.checkConfigurations;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The builder class for {@link PulsarSource} to make it easier for the users to construct a {@link
+ * PulsarSource}.
+ *
+ * <p>The following example shows the minimum setup to create a PulsarSource that reads the String
+ * values from a Pulsar topic.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .build();
+ * }</pre>
+ *
+ * <p>The service url, admin url, subscription name, topics to consume, and the record deserializer
+ * are required fields that must be set.
+ *
+ * <p>To specify the starting position of PulsarSource, one can call {@link
+ * #setStartCursorSupplier(SerializableSupplier)}.
+ *
+ * <p>By default the PulsarSource runs in an {@link Boundedness#CONTINUOUS_UNBOUNDED} mode and never
+ * stop until the Flink job is canceled or fails. To let the PulsarSource run in {@link
+ * Boundedness#CONTINUOUS_UNBOUNDED} but stops at some given offsets, one can call {@link
+ * #setUnbounded(SerializableSupplier)}. For example the following PulsarSource stops after it
+ * consumes up to a event time when the Flink started.
+ *
+ * <pre>{@code
+ * PulsarSource<byte[], String> source = PulsarSource
+ *     .<byte[], String>builder()
+ *     .setServiceUrl(PULSAR_BROKER_URL)
+ *     .setAdminUrl(PULSAR_BROKER_HTTP_URL)
+ *     .setSubscriptionName("flink-source-1")
+ *     .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ *     .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(Types.STRING))
+ *     .setUnbounded(StopCursor.atEventTime(System.currentTimeMillis()))
+ *     .build();
+ * }</pre>
+ *
+ * @param <IN> The input type of the pulsar {@link Message <?>}
+ * @param <OUT> The output type of the source.
+ */
+@PublicEvolving
+public final class PulsarSourceBuilder<IN, OUT> {
+    private static final Logger LOG = LoggerFactory.getLogger(PulsarSourceBuilder.class);
+
+    private final Configuration configuration;
+    private PulsarSubscriber subscriber;
+    private RangeGenerator rangeGenerator;
+    private SerializableSupplier<StartCursor> startCursorSupplier;
+    private SerializableSupplier<StopCursor> stopCursorSupplier;
+    private Boundedness boundedness;
+    private PulsarDeserializationSchema<IN, OUT> deserializationSchema;
+    private ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer;
+    private ConfigurationDataCustomizer<ConsumerConfigurationData<IN>>
+            consumerConfigurationCustomizer;
+
+    // private builder constructor.
+    PulsarSourceBuilder() {
+        // The default configuration holder.
+        this.configuration = new Configuration();
+    }
+
+    /**
+     * Sets the admin endpoint for the PulsarAdmin of the PulsarSource.
+     *
+     * @param adminUrl the url for the PulsarAdmin.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setAdminUrl(String adminUrl) {
+        configuration.set(PULSAR_ADMIN_URL, adminUrl);
+        return this;
+    }
+
+    /**
+     * Sets the server's link for the PulsarConsumer of the PulsarSource.
+     *
+     * @param serviceUrl the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setServiceUrl(String serviceUrl) {
+        configuration.set(PULSAR_SERVICE_URL, serviceUrl);
+        return this;
+    }
+
+    /**
+     * Sets the name for this pulsar subscription.
+     *
+     * @param subscriptionName the server url of the Pulsar cluster.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionName(String subscriptionName) {
+        configuration.set(PULSAR_SUBSCRIPTION_NAME, subscriptionName);
+        return this;
+    }
+
+    /**
+     * {@link SubscriptionType} is the consuming behavior for pulsar, we would generator different
+     * split by the given subscription type, it's required and quite important for end-user.
+     *
+     * @param subscriptionType The type of subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setSubscriptionType(SubscriptionType subscriptionType) {
+        configuration.set(PULSAR_SUBSCRIPTION_TYPE, subscriptionType);
+        return this;
+    }
+
+    /**
+     * Set a pulsar topic list for flink source. Some topic may not exist currently, consuming this
+     * non-existed topic wouldn't throw any exception. But the best solution is just consuming by
+     * using a topic regex. This method is conflict with {@code setTopicPattern}.
+     *
+     * @param topics The topic list you would like to consume message.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopics(String... topics) {
+        return setTopics(Arrays.asList(topics));
+    }
+
+    /**
+     * Set a pulsar topic list for flink source. Some topic may not exist currently, consuming this
+     * non-existed topic wouldn't throw any exception. But the best solution is just consuming by
+     * using a topic regex. This method is conflict with {@code setTopicPattern}.
+     *
+     * @param topics The topic list you would like to consume message.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopics(List<String> topics) {
+        ensureSubscriberIsNull("topics");
+        configuration.set(PULSAR_TOPIC_NAMES, PulsarJsonUtils.toString(topics));
+
+        this.subscriber = PulsarSubscriber.getTopicListSubscriber(topics);
+
+        return this;
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(String topicsPattern) {
+        return setTopicPattern(Pattern.compile(topicsPattern));
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(Pattern topicsPattern) {
+        return setTopicPattern(topicsPattern, RegexSubscriptionMode.AllTopics);
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @param regexSubscriptionMode The topic filter for regex subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(
+            String topicsPattern, RegexSubscriptionMode regexSubscriptionMode) {
+        return setTopicPattern(Pattern.compile(topicsPattern), regexSubscriptionMode);
+    }
+
+    /**
+     * Set a topic pattern to consume from use the java {@link Pattern}.
+     *
+     * @param topicsPattern the pattern of the topic name to consume from.
+     * @param regexSubscriptionMode The topic filter for regex subscription.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setTopicPattern(
+            Pattern topicsPattern, RegexSubscriptionMode regexSubscriptionMode) {
+        ensureSubscriberIsNull("topic pattern");
+        configuration.set(PULSAR_TOPICS_PATTERN, topicsPattern.toString());
+        this.subscriber =
+                PulsarSubscriber.getTopicPatternSubscriber(topicsPattern, regexSubscriptionMode);
+
+        return setRegexSubscriptionMode(regexSubscriptionMode);
+    }
+
+    /**
+     * When subscribing to a topic using a regular expression, you can pick a certain type of
+     * topics.
+     *
+     * @param regexSubscriptionMode The topic type.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setRegexSubscriptionMode(
+            RegexSubscriptionMode regexSubscriptionMode) {
+        configuration.set(PULSAR_REGEX_SUBSCRIPTION_MODE, regexSubscriptionMode);
+        return this;
+    }
+
+    /**
+     * Set a topic range generator for Key_Shared subscription.
+     *
+     * @param rangeGenerator A generator which would generate a set of {@link TopicRange} for given
+     *     topic.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setRangeGenerator(RangeGenerator rangeGenerator) {
+        this.rangeGenerator = rangeGenerator;
+        if (configuration.contains(PULSAR_SUBSCRIPTION_TYPE)) {
+            SubscriptionType subscriptionType = configuration.get(PULSAR_SUBSCRIPTION_TYPE);
+            if (subscriptionType != SubscriptionType.Key_Shared) {
+                LOG.warn(
+                        "Key_Shared subscription should be used for custom rangeGenerator instead of {}",
+                        subscriptionType);
+            }
+        }
+
+        return this;
+    }
+
+    /**
+     * Specify from which offsets the PulsarSource should start consume from by providing an {@link
+     * StartCursor}.
+     *
+     * @param startCursorSupplier the supplier providing a {@link StartCursor} which set the
+     *     starting offsets for the Source.
+     * @return this PulsarSourceBuilder.
+     */
+    public PulsarSourceBuilder<IN, OUT> setStartCursorSupplier(
+            SerializableSupplier<StartCursor> startCursorSupplier) {
+        this.startCursorSupplier = startCursorSupplier;
+        return this;
+    }
+
+    /**
+     * By default the PulsarSource is set to run in {@link Boundedness#CONTINUOUS_UNBOUNDED} manner
+     * and thus never stops until the Flink job fails or is canceled. To let the PulsarSource run as
+     * a streaming source but still stops at some point, one can set an {@link StartCursor} to
+     * specify the stopping offsets for each partition. When all the partitions have reached their
+     * stopping offsets, the PulsarSource will then exit.
+     *
+     * <p>This method is different from {@link #setBounded(SerializableSupplier)} that after setting
+     * the stopping offsets with this method, {@link PulsarSource#getBoundedness()} will still
+     * return {@link Boundedness#CONTINUOUS_UNBOUNDED} even though it will stop at the stopping
+     * offsets specified by the stopping offsets {@link StartCursor}.
+     *
+     * @param stopCursorSupplier The {@link StopCursor} to specify the stopping offset.
+     * @return this PulsarSourceBuilder.
+     * @see #setBounded(SerializableSupplier)
+     */
+    public PulsarSourceBuilder<IN, OUT> setUnbounded(
+            SerializableSupplier<StopCursor> stopCursorSupplier) {
+        this.boundedness = Boundedness.CONTINUOUS_UNBOUNDED;
+        this.stopCursorSupplier = stopCursorSupplier;
+        return this;
+    }
+
+    /**
+     * By default the PulsarSource is set to run in {@link Boundedness#CONTINUOUS_UNBOUNDED} manner
+     * and thus never stops until the Flink job fails or is canceled. To let the PulsarSource run in
+     * {@link Boundedness#BOUNDED} manner and stops at some point, one can set an {@link
+     * StartCursor} to specify the stopping offsets for each partition. When all the partitions have
+     * reached their stopping offsets, the PulsarSource will then exit.
+     *
+     * <p>This method is different from {@link #setUnbounded(SerializableSupplier)} that after
+     * setting the stopping offsets with this method, {@link PulsarSource#getBoundedness()} will
+     * return {@link Boundedness#BOUNDED} instead of {@link Boundedness#CONTINUOUS_UNBOUNDED}.
+     *
+     * @param stopCursorSupplier the {@link StopCursor} to specify the stopping offsets.
+     * @return this PulsarSourceBuilder.
+     * @see #setUnbounded(SerializableSupplier)
+     */
+    public PulsarSourceBuilder<IN, OUT> setBounded(

Review comment:
       `setStopCursor`?

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumerator.java
##########
@@ -0,0 +1,237 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source.enumerator;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.exception.PulsarRuntimeException;
+import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+import static java.util.Collections.singletonList;
+import static org.apache.flink.connector.pulsar.common.utils.PulsarClosingUtils.closeWithTimeout;
+
+/** The enumerator class for pulsar source. */
+@Internal
+public class PulsarSourceEnumerator
+        implements SplitEnumerator<PulsarPartitionSplit, PulsarSourceEnumState> {
+
+    private static final Logger LOG = LoggerFactory.getLogger(PulsarSourceEnumerator.class);
+
+    /** The admin interface for pulsar. */
+    private final PulsarAdmin pulsarAdmin;
+
+    private final PulsarSubscriber subscriber;
+    private final RangeGenerator rangeGenerator;
+    private final Configuration configuration;
+    private final SourceConfiguration sourceConfiguration;
+    private final SplitEnumeratorContext<PulsarPartitionSplit> context;
+    private final SplitsAssignmentState assignmentState;
+
+    public PulsarSourceEnumerator(
+            PulsarAdmin pulsarAdmin,
+            PulsarSubscriber subscriber,
+            RangeGenerator rangeGenerator,
+            SerializableSupplier<StartCursor> startCursorSupplier,
+            SerializableSupplier<StopCursor> stopCursorSupplier,
+            Configuration configuration,
+            SourceConfiguration sourceConfiguration,
+            SplitEnumeratorContext<PulsarPartitionSplit> context) {
+        this.pulsarAdmin = pulsarAdmin;
+        this.subscriber = subscriber;
+        this.rangeGenerator = rangeGenerator;
+        this.configuration = configuration;
+        this.sourceConfiguration = sourceConfiguration;
+        this.context = context;
+        this.assignmentState =
+                new SplitsAssignmentState(
+                        startCursorSupplier, stopCursorSupplier, sourceConfiguration);
+    }
+
+    public PulsarSourceEnumerator(
+            PulsarAdmin pulsarAdmin,
+            PulsarSubscriber subscriber,
+            RangeGenerator rangeGenerator,
+            SerializableSupplier<StartCursor> startCursorSupplier,
+            SerializableSupplier<StopCursor> stopCursorSupplier,
+            Configuration configuration,
+            SourceConfiguration sourceConfiguration,
+            SplitEnumeratorContext<PulsarPartitionSplit> context,
+            PulsarSourceEnumState sourceEnumState) {
+        this.pulsarAdmin = pulsarAdmin;
+        this.subscriber = subscriber;
+        this.rangeGenerator = rangeGenerator;
+        this.configuration = configuration;
+        this.sourceConfiguration = sourceConfiguration;
+        this.context = context;
+        this.assignmentState =
+                new SplitsAssignmentState(
+                        startCursorSupplier,
+                        stopCursorSupplier,
+                        sourceConfiguration,
+                        sourceEnumState);
+    }
+

Review comment:
       Please merge ctors for example by adding `SplitsAssignmentState` as a parameter and filling that in at call-site.

##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumStateSerializer.java
##########
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.source.enumerator;
+
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
+import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitSerializer;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.flink.connector.pulsar.common.utils.PulsarSerdeUtils.deserializeMap;
+import static org.apache.flink.connector.pulsar.common.utils.PulsarSerdeUtils.deserializeSet;
+import static org.apache.flink.connector.pulsar.common.utils.PulsarSerdeUtils.serializeMap;
+import static org.apache.flink.connector.pulsar.common.utils.PulsarSerdeUtils.serializeSet;
+
+/** The {@link SimpleVersionedSerializer Serializer} for the enumerator state of Pulsar source. */
+public class PulsarSourceEnumStateSerializer
+        implements SimpleVersionedSerializer<PulsarSourceEnumState> {
+
+    private static final PulsarPartitionSplitSerializer SPLIT_SERIALIZER =
+            new PulsarPartitionSplitSerializer();

Review comment:
       Maybe use singleton pattern in `PulsarPartitionSplitSerializer` and just fetch instance here.
   This `PulsarSourceEnumStateSerializer` could also be a singleton.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org