我想启用Flink从Pulsar读取数据

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-pulsar_2.11</artifactId>
    <version>1.13.2</version>
</dependency>

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.pulsar.PulsarSourceBuilder;
import org.apache.flink.streaming.connectors.pulsar.config.StartupMode;
import org.apache.flink.streaming.connectors.pulsar.internal.PulsarDeserializationSchemaWrapper;
import org.apache.flink.api.common.serialization.SimpleStringSchema;

public class PulsarFlinkExample {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        PulsarSourceBuilder<String> builder = PulsarSourceBuilder
            .builder(new PulsarDeserializationSchemaWrapper<>(new SimpleStringSchema()))
            .serviceUrl("pulsar://localhost:6650")
            .topic("persistent://public/default/my-topic")
            .subscriptionName("my-subscription")
            .subscriptionType(SubscriptionType.Exclusive)
            .startupMode(StartupMode.Latest);

        env.addSource(builder.build()).print();

        env.execute("Flink Pulsar Example");
    }
}