[Bug][Sort] Fix Pulsar connecotr deserialize complex format with multiple data will cause data loss

[thesumery]

What happened

when using pulsar connecor to read data with inlong msg, some data will loss.

What you expected to happen

All data exactly equal with what inlong msg includes.

How to reproduce

1. prepare a flink environment(flink 1.13.6)
2. put all dependency into flink lib
3. start flink standalone service
4. execute flink sql client

CREATE TABLE `test`(
    `__content__` STRING)
    WITH (
    'properties.auth-params' = 'xxxxxxxxxxxxxxxxxx',
    'properties.auth-plugin-classname' = 'org.apache.pulsar.client.impl.auth.AuthenticationToken',
    'inlong.metric' = 's5825b746-225f-40ff-afaf-eb273c64dcf3&b_s5825b746-225f-40ff-afaf-eb273c64dcf3&b_s5825b746-225f-40ff-afaf-eb273c64dcf3',
    'connector' = 'pulsar-inlong',
    'inlong-msg.inner.format' = 'raw',
    'format' = 'inlong-msg',
    'inlong-msg.raw.charset' = 'UTF-8',
    'inlong-msg.raw.endianness' = 'big-endian',
    'inlong-msg.ignore-parse-errors' = 'false',
    'generic' = 'true',
    'service-url' = 'http://pulsar-op488nga8meo.tdmq.ap-bj.public.tencenttdmq.com:8080',
    'topic' = 'pulsar-op488nga8meo/namespace_1257305158_20220826111254509798/b_s5825b746-225f-40ff-afaf-eb273c64dcf3',
    'scan.startup.mode' = 'latest'
);

select * from test;

5. start pulsar service and produce data into pulsar
   produce inlong msg data into pulsar.

Environment

No response

InLong version

master

InLong Component

InLong Sort

Are you willing to submit PR?

• Yes, I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct