[INLONG-7940][Sort] Use pulsar subscriptions and specify subscription offset in pulsar connector

inlong-sort/sort-common/src/test/java/org/apache/inlong/sort/protocol/node/extract/PulsarExtractNodeTest.java

@@ -50,7 +50,7 @@ public Node getTestObject() {
                 format,
                 "earliest",
                 null,
-                null,
-                null);
+                "subscription",
+                "earliest");
     }
 }

inlong-sort/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/FlinkPulsarSource.java

@@ -1006,6 +1006,16 @@ public Map<TopicRange, MessageId> offsetForEachTopic(
                     }
                 }
                 return specificOffsets;
+            case EXTERNAL_SUBSCRIPTION:
+                Map<TopicRange, MessageId> offsetsFromSubs = new HashMap<>();
+                for (TopicRange topic : topics) {
+                    offsetsFromSubs.put(
+                            topic,
+                            metadataReader.getPositionFromSubscription(
+                                    topic, subscriptionPosition));
+                }
+                log.info("offset for each topic: {}", offsetsFromSubs);
+                return offsetsFromSubs;
         }
         return null;
     }

inlong-sort/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/PulsarMetadataReader.java

@@ -17,6 +17,9 @@
 
 package org.apache.inlong.sort.pulsar.internal;
 
+import java.io.UnsupportedEncodingException;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.ConfigOptions;
 import org.apache.flink.streaming.connectors.pulsar.internal.PulsarClientUtils;
@@ -33,9 +36,13 @@
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Range;
+import org.apache.pulsar.client.impl.MessageIdImpl;
 import org.apache.pulsar.client.impl.PulsarClientImpl;
 import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
 import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
+import org.apache.pulsar.common.policies.data.SubscriptionStats;
+import org.apache.pulsar.common.policies.data.TopicStats;
 import org.apache.pulsar.shade.com.google.common.collect.Sets;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -260,6 +267,44 @@ public void commitOffsetToCursor(Map<TopicRange, MessageId> offset) {
         }
     }
 
+    public MessageId getPositionFromSubscription(TopicRange topic, MessageId defaultPosition) {
+        try {
+            String subscriptionName = subscriptionNameFrom(topic);
+            TopicStats topicStats = admin.topics().getStats(topic.getTopic());
+            if (topicStats.getSubscriptions().containsKey(subscriptionName)) {
+                SubscriptionStats subStats = topicStats.getSubscriptions().get(subscriptionName);
+                if (subStats.getConsumers().size() != 0) {
+                    throw new IllegalStateException(
+                            "Subscription been actively used by other consumers, "
+                                    + "in this situation, the exactly-once semantics cannot be guaranteed.");
+                } else {
+                    String encodedSubName =
+                            URLEncoder.encode(subscriptionName, StandardCharsets.UTF_8.toString());
+                    PersistentTopicInternalStats.CursorStats c =
+                            admin.topics()
+                                    .getInternalStats(topic.getTopic()).cursors
+                                            .get(encodedSubName);
+                    String[] ids = c.markDeletePosition.split(":", 2);
+                    long ledgerId = Long.parseLong(ids[0]);
+                    long entryIdInMarkDelete = Long.parseLong(ids[1]);
+                    // we are getting the next mid from sub position, if the entryId is -1,
+                    // it denotes we haven't read data from the ledger before,
+                    // therefore no need to skip the current entry for the next position
+                    long entryId = entryIdInMarkDelete == -1 ? -1 : entryIdInMarkDelete + 1;
+                    int partitionIdx = TopicName.getPartitionIndex(topic.getTopic());
+                    return new MessageIdImpl(ledgerId, entryId, partitionIdx);
+                }
+            } else {
+                // create sub on topic
+                admin.topics()
+                        .createSubscription(topic.getTopic(), subscriptionName, defaultPosition);
+                return defaultPosition;
+            }
+        } catch (PulsarAdminException | UnsupportedEncodingException e) {
+            throw new IllegalStateException("Failed to get stats for topic " + topic, e);
+        }
+    }
+
     /**
      * Designate the close of the metadata reader.
      */

inlong-sort/sort-core/src/test/java/org/apache/inlong/sort/parser/PulsarSqlParserTest.java

@@ -74,8 +74,8 @@ public PulsarExtractNode buildPulsarExtractNode() {
                 format,
                 "earliest",
                 null,
-                null,
-                null);
+                "test",
+                "earliest");
     }
 
     private NodeRelation buildNodeRelation(List<Node> inputs, List<Node> outputs) {