private List<PulsarTopicState<T>> createPartitionStateHolders(
            Map<TopicRange, MessageId> partitionsToInitialOffsets,
            int timestampWatermarkMode,
            SerializedValue<WatermarkStrategy<T>> watermarkStrategy,
            ClassLoader userCodeClassLoader) throws IOException, ClassNotFoundException {

        // CopyOnWrite as adding discovered partitions could happen in parallel
        // while different threads iterate the partitions list
        List<PulsarTopicState<T>> partitionStates = new CopyOnWriteArrayList<>();

        switch (timestampWatermarkMode) {
            case NO_TIMESTAMPS_WATERMARKS: {
                for (Map.Entry<TopicRange, MessageId> partitionEntry : partitionsToInitialOffsets.entrySet()) {
                    PulsarTopicState<T> state = new PulsarTopicState(partitionEntry.getKey());
                    state.setOffset(partitionEntry.getValue());
                    partitionStates.add(state);
                }
                return partitionStates;
            }

            case WITH_WATERMARK_GENERATOR: {
                for (Map.Entry<TopicRange, MessageId> partitionEntry : partitionsToInitialOffsets.entrySet()) {
                    final TopicRange topicRange = partitionEntry.getKey();

                    PulsarTopicState<T> state = new PulsarTopicState(partitionEntry.getKey());
                    WatermarkStrategy<T> deserializedWatermarkStrategy = watermarkStrategy.deserializeValue(
                            userCodeClassLoader);

                    // the format of the ID does not matter, as long as it is unique
                    final String partitionId = state.getTopicRange().toString();
                    watermarkOutputMultiplexer.registerNewOutput(partitionId);
                    WatermarkOutput immediateOutput =
                            watermarkOutputMultiplexer.getImmediateOutput(partitionId);
                    WatermarkOutput deferredOutput =
                            watermarkOutputMultiplexer.getDeferredOutput(partitionId);

                    PulsarTopicPartitionStateWithWatermarkGenerator<T> partitionState =
                            new PulsarTopicPartitionStateWithWatermarkGenerator<>(
                                    topicRange,
                                    state,
                                    deserializedWatermarkStrategy.createTimestampAssigner(() -> consumerMetricGroup),
                                    deserializedWatermarkStrategy.createWatermarkGenerator(() -> consumerMetricGroup),
                                    immediateOutput,
                                    deferredOutput);

                    partitionState.setOffset(partitionEntry.getValue());

                    partitionStates.add(partitionState);

                }

                return partitionStates;
            }

            default:
                // cannot happen, add this as a guard for the future
                throw new RuntimeException();
        }
    }
Utility method that takes the topic partitions and creates the topic partition state
holders, depending on the timestamp / watermark mode.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/PulsarFetcher.java