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);
        }
    }
Get topic partitions all. If the topic does not exist, it is created automatically

@return allTopicPartitions
@throws PulsarAdminException pulsarAdminException
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/PulsarMetadataReader.java