Problem with protobuf deserializer

[Welkin-Y]

Problem description
Hello, I am currently working on the flink-connector project trying to add protobuf support. I've noticed that after serializing a message to a ByteBuffer, if I convert it to a byte[] using .array(), and then convert it back to a ByteBuffer using ByteBuffer.wrap(), there will be an exception when passing it to the deserialize function. I noticed that in the test cases, there are no scenarios covering conversions between ByteBuffers and byte arrays. However, such a convertion is needed for flink-connector interfaces.

Problem location

To reproduce the exception, you can modify this file by changing line 84 as follows:, you can regenerate the exception by changing line 84 as follows:

The error message is like this:

Suggestions for an improvement
I'm not sure if it's related to the deserialize method. From my experiments, I found that the issue arises because, after converting a ByteBuffer to a byte[] and back, the limit information of the ByteBuffer is lost, leading to a parsing error in protobuf.

[crazyzhou]

@Welkin-Y When parsing bytebuffer into byte[], do you have the information of the limit? Directly using array() may lose such infomation, there are some usages like: https://github.com/pravega/flink-connectors/blob/053269788245cccd53257cd70178097d4b61bf6c/src/main/java/io/pravega/connectors/flink/util/FlinkPravegaUtils.java#L204 Maybe you can have a try.

@shshashwat Please also help to see if there are anything can be improvement from this side.