[Bug] when trying to sync data from pulsar to hdfs using flink-pulsar-connector, it always occurs an exception of :

### Search before asking

- [X] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.

### Version

3.0.0

### Minimal reproduce step

1. I have code as following to sync data from pulsar into hdfs
`
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// env.socketTextStream("10.180.85.1", 8081);
env.socketTextStream("localhost", 8081);
// String serviceUrl = "pulsar://10.180.85.1:6650";
// String adminUrl = "http://10.180.85.1:8080";
String serviceUrl = "pulsar://localhost:6650";
String adminUrl = "http://localhost:8080";
// String topic = "public/my-namespace/my-topic";
// String path = "hdfs://10.180.85.1:9000/user/root/paimon/files";
String path = "E:/workspace/algorithum/pulsar17/out";

PulsarSource source = PulsarSource.builder()
.setServiceUrl(serviceUrl)
.setAdminUrl(adminUrl)
.setStartCursor(StartCursor.earliest())
.setTopics("t1/np/to")
// .setSubscriptionType(SubscriptionType.Shared)
.setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
.setSubscriptionName("my-subscription")
.build();

DataStreamSource stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
stream.print();
DataStream shuffle = stream.shuffle();

final FileSink sink = FileSink
.forRowFormat(new Path(path),
new SimpleStringEncoder("UTF-8"))
.withRollingPolicy(
DefaultRollingPolicy.builder()
.withRolloverInterval(Duration.ofMinutes(15))
.withInactivityInterval(Duration.ofMinutes(5))
.withMaxPartSize(MemorySize.ofMebiBytes(1024))
.build())
.build();
shuffle.sinkTo(sink);
env.execute();
}`
2. I debug the code and find stack will occur this exception all the time:
3. I can promise that there's no other comsumers connect this topic:

### What did you expect to see?

i want to sync data from pulsar to hdfs well

### What did you see instead?

see output of the data

### Anything else?

_No response_

### Are you willing to submit a PR?

- [X] I'm willing to submit a PR!