[FLINK-27399][Connector/Pulsar] Change initial consuming position setting logic for better handle the checkpoint. (apache#19972)

* Change the initial start cursor and stop cursor to better handle the consuming behaviors.
* Create the initial subscription instead seek every time. This should fix the wrong position setting.
* Fix the wrong stop cursor, make sure it stops at the correct space
* Drop Consumer.seek() for apache/pulsar#16171

Author: syhily

docs/content.zh/docs/connectors/datastream/pulsar.md

@@ -322,6 +322,7 @@ Pulsar Source 使用 `setStartCursor(StartCursor)` 方法给定开始消费的
   ```
   {{< /tab >}}
   {{< /tabs >}}
+
 - 与前者不同的是，给定的消息可以跳过，再进行消费。
   {{< tabs "pulsar-starting-position-from-message-id-bool" >}}
   {{< tab "Java" >}}
@@ -335,7 +336,8 @@ Pulsar Source 使用 `setStartCursor(StartCursor)` 方法给定开始消费的
   ```
   {{< /tab >}}
   {{< /tabs >}}
-- 从给定的消息时间开始消费。
+
+- 从给定的消息发布时间开始消费，这个方法因为名称容易导致误解现在已经不建议使用。你可以使用方法 `StartCursor.fromPublishTime(long)`。
   {{< tabs "pulsar-starting-position-message-time" >}}
   {{< tab "Java" >}}
   ```java
@@ -349,6 +351,11 @@ Pulsar Source 使用 `setStartCursor(StartCursor)` 方法给定开始消费的
   {{< /tab >}}
   {{< /tabs >}}
 
+- 从给定的消息发布时间开始消费。
+  ```java
+  StartCursor.fromPublishTime(long);
+  ```
+
 {{< hint info >}}
 每条消息都有一个固定的序列号，这个序列号在 Pulsar 上有序排列，其包含了 ledger、entry、partition 等原始信息，用于在 Pulsar 底层存储上查找到具体的消息。
 
@@ -404,6 +411,7 @@ Pulsar Source 默认情况下使用流的方式消费数据。除非任务失败
   ```
   {{< /tab >}}
   {{< /tabs >}}
+
 - 停止于某条消息之后，结果里包含此消息。
   {{< tabs "pulsar-boundedness-after-message-id" >}}
   {{< tab "Java" >}}
@@ -417,7 +425,18 @@ Pulsar Source 默认情况下使用流的方式消费数据。除非任务失败
   ```
   {{< /tab >}}
   {{< /tabs >}}
-- 停止于某个给定的消息发布时间戳，比如 `Message<byte[]>.getPublishTime()`。
+
+- 停止于某个给定的消息事件时间戳，比如 `Message<byte[]>.getEventTime()`，消费结果里不包含此时间戳的消息。
+  ```java
+  StopCursor.atEventTime(long);
+  ```
+
+- 停止于某个给定的消息事件时间戳，比如 `Message<byte[]>.getEventTime()`，消费结果里包含此时间戳的消息。
+  ```java
+  StopCursor.afterEventTime(long);
+  ```
+
+- 停止于某个给定的消息发布时间戳，比如 `Message<byte[]>.getPublishTime()`，消费结果里不包含此时间戳的消息。
   {{< tabs "pulsar-boundedness-publish-time" >}}
   {{< tab "Java" >}}
   ```java
@@ -431,9 +450,10 @@ Pulsar Source 默认情况下使用流的方式消费数据。除非任务失败
   {{< /tab >}}
   {{< /tabs >}}
 
-  {{< hint warning >}}
-  StopCursor.atEventTime(long) 目前已经处于弃用状态。
-  {{< /hint >}}
+- 停止于某个给定的消息发布时间戳，比如 `Message<byte[]>.getPublishTime()`，消费结果里包含此时间戳的消息。
+  ```java
+  StopCursor.afterPublishTime(long);
+  ```
 
 ### Source 配置项
 

docs/content/docs/connectors/datastream/pulsar.md

@@ -356,6 +356,7 @@ The Pulsar connector consumes from the latest available message if the message I
   ```
   {{< /tab >}}
   {{< /tabs >}}
+
 - Start from a specified message between the earliest and the latest.
 The Pulsar connector consumes from the latest available message if the message ID doesn't exist.
 
@@ -373,7 +374,10 @@ The Pulsar connector consumes from the latest available message if the message I
   {{< /tab >}}
   {{< /tabs >}}
 
-- Start from the specified message time by `Message<byte[]>.getPublishTime()`.
+- Start from the specified message publish time by `Message<byte[]>.getPublishTime()`.
+This method is deprecated because the name is totally wrong which may cause confuse.
+You can use `StartCursor.fromPublishTime(long)` instead.
+
   {{< tabs "pulsar-starting-position-message-time" >}}
   {{< tab "Java" >}}
   ```java
@@ -387,6 +391,11 @@ The Pulsar connector consumes from the latest available message if the message I
   {{< /tab >}}
   {{< /tabs >}}
 
+- Start from the specified message publish time by `Message<byte[]>.getPublishTime()`.
+  ```java
+  StartCursor.fromPublishTime(long);
+  ```
+
 {{< hint info >}}
 Each Pulsar message belongs to an ordered sequence on its topic.
 The sequence ID (`MessageId`) of the message is ordered in that sequence.
@@ -420,7 +429,7 @@ Built-in stop cursors include:
   {{< /tab >}}
   {{< /tabs >}}
 
-- Stop at the latest available message when the  Pulsar source starts consuming messages.
+- Stop at the latest available message when the Pulsar source starts consuming messages.
   {{< tabs "pulsar-boundedness-latest" >}}
   {{< tab "Java" >}}
   ```java
@@ -447,6 +456,7 @@ Built-in stop cursors include:
   ```
   {{< /tab >}}
   {{< /tabs >}}
+
 - Stop but include the given message in the consuming result.
   {{< tabs "pulsar-boundedness-after-message-id" >}}
   {{< tab "Java" >}}
@@ -461,7 +471,20 @@ Built-in stop cursors include:
   {{< /tab >}}
   {{< /tabs >}}
 
-- Stop at the specified message time by `Message<byte[]>.getPublishTime()`.
+- Stop at the specified event time by `Message<byte[]>.getEventTime()`. The message with the
+given event time won't be included in the consuming result.
+  ```java
+  StopCursor.atEventTime(long);
+  ```
+
+- Stop after the specified event time by `Message<byte[]>.getEventTime()`. The message with the
+given event time will be included in the consuming result.
+  ```java
+  StopCursor.afterEventTime(long);
+  ```
+
+- Stop at the specified publish time by `Message<byte[]>.getPublishTime()`. The message with the
+given publish time won't be included in the consuming result.
   {{< tabs "pulsar-boundedness-publish-time" >}}
   {{< tab "Java" >}}
   ```java
@@ -475,9 +498,11 @@ Built-in stop cursors include:
   {{< /tab >}}
   {{< /tabs >}}
 
-  {{< hint warning >}}
-  StopCursor.atEventTime(long) is now deprecated.
-  {{< /hint >}}
+- Stop after the specified publish time by `Message<byte[]>.getPublishTime()`. The message with the
+given publish time will be included in the consuming result.
+  ```java
+  StopCursor.afterPublishTime(long);
+  ```
 
 ### Source Configurable Options
 

docs/layouts/shortcodes/generated/pulsar_consumer_configuration.html

@@ -140,12 +140,6 @@
             <td>Boolean</td>
             <td>If enabled, the consumer will automatically retry messages.</td>
         </tr>
-        <tr>
-            <td><h5>pulsar.consumer.subscriptionInitialPosition</h5></td>
-            <td style="word-wrap: break-word;">Latest</td>
-            <td><p>Enum</p></td>
-            <td>Initial position at which to set cursor when subscribing to a topic at first time.<br /><br />Possible values:<ul><li>"Latest"</li><li>"Earliest"</li></ul></td>
-        </tr>
         <tr>
             <td><h5>pulsar.consumer.subscriptionMode</h5></td>
             <td style="word-wrap: break-word;">Durable</td>

flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSource.java

@@ -32,7 +32,8 @@
 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.SplitsAssignmentState;
+import org.apache.flink.connector.pulsar.source.enumerator.assigner.SplitAssigner;
+import org.apache.flink.connector.pulsar.source.enumerator.assigner.SplitAssignerFactory;
 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;
@@ -142,31 +143,30 @@ public SourceReader<OUT, PulsarPartitionSplit> createReader(SourceReaderContext
     @Override
     public SplitEnumerator<PulsarPartitionSplit, PulsarSourceEnumState> createEnumerator(
             SplitEnumeratorContext<PulsarPartitionSplit> enumContext) {
-        SplitsAssignmentState assignmentState =
-                new SplitsAssignmentState(stopCursor, sourceConfiguration);
+        SplitAssigner splitAssigner = SplitAssignerFactory.create(stopCursor, sourceConfiguration);
         return new PulsarSourceEnumerator(
                 subscriber,
                 startCursor,
                 rangeGenerator,
                 sourceConfiguration,
                 enumContext,
-                assignmentState);
+                splitAssigner);
     }
 
     @Internal
     @Override
     public SplitEnumerator<PulsarPartitionSplit, PulsarSourceEnumState> restoreEnumerator(
             SplitEnumeratorContext<PulsarPartitionSplit> enumContext,
             PulsarSourceEnumState checkpoint) {
-        SplitsAssignmentState assignmentState =
-                new SplitsAssignmentState(stopCursor, sourceConfiguration, checkpoint);
+        SplitAssigner splitAssigner =
+                SplitAssignerFactory.create(stopCursor, sourceConfiguration, checkpoint);
         return new PulsarSourceEnumerator(
                 subscriber,
                 startCursor,
                 rangeGenerator,
                 sourceConfiguration,
                 enumContext,
-                assignmentState);
+                splitAssigner);
     }
 
     @Internal

flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java

@@ -97,7 +97,7 @@
  * <p>To stop the connector user has to disable the auto partition discovery. As auto partition
  * discovery always expected new splits to come and not exiting. To disable auto partition
  * discovery, use builder.setConfig({@link
- * PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS}, -1).
+ * PulsarSourceOptions#PULSAR_PARTITION_DISCOVERY_INTERVAL_MS}, -1).
  *
  * <pre>{@code
  * PulsarSource<String> source = PulsarSource
@@ -266,7 +266,7 @@ public PulsarSourceBuilder<OUT> setTopicPattern(
     }
 
     /**
-     * The consumer name is informative and it can be used to identify a particular consumer
+     * The consumer name is informative, and it can be used to identify a particular consumer
      * instance from the topic stats.
      */
     public PulsarSourceBuilder<OUT> setConsumerName(String consumerName) {
@@ -321,7 +321,7 @@ public PulsarSourceBuilder<OUT> setStartCursor(StartCursor startCursor) {
      * <p>To stop the connector user has to disable the auto partition discovery. As auto partition
      * discovery always expected new splits to come and not exiting. To disable auto partition
      * discovery, use builder.setConfig({@link
-     * PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS}, -1).
+     * PulsarSourceOptions#PULSAR_PARTITION_DISCOVERY_INTERVAL_MS}, -1).
      *
      * @param stopCursor The {@link StopCursor} to specify the stopping offset.
      * @return this PulsarSourceBuilder.
@@ -334,7 +334,7 @@ public PulsarSourceBuilder<OUT> setUnboundedStopCursor(StopCursor stopCursor) {
     }
 
     /**
-     * By default the PulsarSource is set to run in {@link Boundedness#CONTINUOUS_UNBOUNDED} manner
+     * 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 StopCursor}
      * to specify the stopping offsets for each partition. When all the partitions have reached

flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java

@@ -26,6 +26,7 @@
 import org.apache.flink.configuration.description.Description;
 import org.apache.flink.connector.pulsar.common.config.PulsarOptions;
 import org.apache.flink.connector.pulsar.source.config.CursorVerification;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
 
 import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
 import org.apache.pulsar.client.api.SubscriptionInitialPosition;
@@ -503,6 +504,12 @@ private PulsarSourceOptions() {
                                             code("PulsarClientException"))
                                     .build());
 
+    /**
+     * @deprecated This option would be reset by {@link StartCursor}, no need to use it anymore.
+     *     Pulsar didn't support this config option before 1.10.1, so we have to remove this config
+     *     option.
+     */
+    @Deprecated
     public static final ConfigOption<SubscriptionInitialPosition>
             PULSAR_SUBSCRIPTION_INITIAL_POSITION =
                     ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "subscriptionInitialPosition")

flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java

@@ -59,7 +59,6 @@
 import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_REPLICATE_SUBSCRIPTION_STATE;
 import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_RETRY_ENABLE;
 import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_RETRY_LETTER_TOPIC;
-import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_INITIAL_POSITION;
 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;
@@ -113,8 +112,6 @@ public static <T> ConsumerBuilder<T> createConsumerBuilder(
                 builder::consumerName);
         configuration.useOption(PULSAR_READ_COMPACTED, builder::readCompacted);
         configuration.useOption(PULSAR_PRIORITY_LEVEL, builder::priorityLevel);
-        configuration.useOption(
-                PULSAR_SUBSCRIPTION_INITIAL_POSITION, builder::subscriptionInitialPosition);
         createDeadLetterPolicy(configuration).ifPresent(builder::deadLetterPolicy);
         configuration.useOption(
                 PULSAR_AUTO_UPDATE_PARTITIONS_INTERVAL_SECONDS,

flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/SourceConfiguration.java

@@ -83,6 +83,10 @@ public int getMessageQueueCapacity() {
         return messageQueueCapacity;
     }
 
+    /**
+     * We would override the interval into a negative number when we set the connector with bounded
+     * stop cursor.
+     */
     public boolean isEnablePartitionDiscovery() {
         return getPartitionDiscoveryIntervalMs() > 0;
     }

flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumState.java

@@ -18,15 +18,18 @@
 
 package org.apache.flink.connector.pulsar.source.enumerator;
 
+import org.apache.flink.connector.pulsar.source.enumerator.assigner.SplitAssigner;
 import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
 import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
 
+import java.util.HashMap;
+import java.util.HashSet;
 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}.
+ * managed and controlled by {@link SplitAssigner}.
  */
 public class PulsarSourceEnumState {
 
@@ -46,11 +49,12 @@ public class PulsarSourceEnumState {
     private final Map<Integer, Set<PulsarPartitionSplit>> sharedPendingPartitionSplits;
 
     /**
-     * A {@link PulsarPartitionSplit} should be assigned for all flink readers. Using this map for
-     * recording assign status.
+     * It is used for Shared subscription. Every {@link PulsarPartitionSplit} should be assigned for
+     * all flink readers. Using this map for recording assign status.
      */
     private final Map<Integer, Set<String>> readerAssignedSplits;
 
+    /** The pipeline has been triggered and topic partitions have been assigned to readers. */
     private final boolean initialized;
 
     public PulsarSourceEnumState(
@@ -85,4 +89,10 @@ public Map<Integer, Set<String>> getReaderAssignedSplits() {
     public boolean isInitialized() {
         return initialized;
     }
+
+    /** The initial assignment state for Pulsar. */
+    public static PulsarSourceEnumState initialState() {
+        return new PulsarSourceEnumState(
+                new HashSet<>(), new HashSet<>(), new HashMap<>(), new HashMap<>(), false);
+    }
 }