[BUG] schema incompatible while use PulsarPrimitiveSchema(String.class) to consume pulsar topic

[KevinyhZou]

Describe the bug
I use FlinkPulsarSource consume a pulsar topic, the code as below

FlinkPulsarSource<String> flinkPulsarSource =
                new FlinkPulsarSource<String>(
                        serviceUrl,
                        adminUrl,
                        new PulsarPrimitiveSchema(String.class),
                        properties).setStartFromSubscription(groupid_antisdk, MessageId.latest);

while submit the job to flink cluster, the exception throws as blow

To Reproduce
do it as the description above.

Expected behavior
the flink job run normally, and the data can deserialize to string by this schema.