[kafka] branch trunk updated: Fix ByteBufferSerializer#serialize(String, ByteBuffer) not roundtrip input with ByteBufferDeserializer#deserialize(String, byte[]) (#12704)

[guozhang]

This is an automated email from the ASF dual-hosted git repository.

guozhang pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 496ae054c2d Fix ByteBufferSerializer#serialize(String, ByteBuffer) not 
roundtrip input with ByteBufferDeserializer#deserialize(String, byte[]) (#12704)
496ae054c2d is described below

commit 496ae054c2d43c0905167745bfb2f4a0725e9fc2
Author: LinShunKang <linshunkang....@gmail.com>
AuthorDate: Fri Sep 30 21:45:18 2022 +0800

    Fix ByteBufferSerializer#serialize(String, ByteBuffer) not roundtrip input 
with ByteBufferDeserializer#deserialize(String, byte[]) (#12704)
    
    Reviewers: Guozhang Wang <wangg...@gmail.com>
---
 .../apache/kafka/common/serialization/ByteBufferSerializer.java   | 8 +++-----
 .../org/apache/kafka/common/serialization/SerializationTest.java  | 4 +++-
 2 files changed, 6 insertions(+), 6 deletions(-)

diff --git 
a/clients/src/main/java/org/apache/kafka/common/serialization/ByteBufferSerializer.java
 
b/clients/src/main/java/org/apache/kafka/common/serialization/ByteBufferSerializer.java
index 5987688759e..06b66a62cb0 100644
--- 
a/clients/src/main/java/org/apache/kafka/common/serialization/ByteBufferSerializer.java
+++ 
b/clients/src/main/java/org/apache/kafka/common/serialization/ByteBufferSerializer.java
@@ -21,8 +21,7 @@ import org.apache.kafka.common.utils.Utils;
 import java.nio.ByteBuffer;
 
 /**
- * ByteBufferSerializer will not change ByteBuffer's mark, position and limit.
- * And do not need to flip before call <i>serialize(String, ByteBuffer)</i>. 
For example:
+ * Do not need to flip before call <i>serialize(String, ByteBuffer)</i>. For 
example:
  *
  * <blockquote>
  * <pre>
@@ -48,8 +47,7 @@ public class ByteBufferSerializer implements 
Serializer<ByteBuffer> {
             }
         }
 
-        final ByteBuffer copyData = data.asReadOnlyBuffer();
-        copyData.flip();
-        return Utils.toArray(copyData);
+        data.flip();
+        return Utils.toArray(data);
     }
 }
diff --git 
a/clients/src/test/java/org/apache/kafka/common/serialization/SerializationTest.java
 
b/clients/src/test/java/org/apache/kafka/common/serialization/SerializationTest.java
index eb1fee3943f..a0b67a03d42 100644
--- 
a/clients/src/test/java/org/apache/kafka/common/serialization/SerializationTest.java
+++ 
b/clients/src/test/java/org/apache/kafka/common/serialization/SerializationTest.java
@@ -50,7 +50,9 @@ public class SerializationTest {
             put(Float.class, Arrays.asList(5678567.12312f, -5678567.12341f));
             put(Double.class, Arrays.asList(5678567.12312d, -5678567.12341d));
             put(byte[].class, Arrays.asList("my string".getBytes()));
-            put(ByteBuffer.class, 
Arrays.asList(ByteBuffer.allocate(10).put("my string".getBytes())));
+            put(ByteBuffer.class, Arrays.asList(ByteBuffer.wrap("my 
string".getBytes()),
+                    ByteBuffer.allocate(10).put("my string".getBytes()),
+                    ByteBuffer.allocateDirect(10).put("my 
string".getBytes())));
             put(Bytes.class, Arrays.asList(new Bytes("my string".getBytes())));
             put(UUID.class, Arrays.asList(UUID.randomUUID()));
         }