commit 496ae054c2d43c0905167745bfb2f4a0725e9fc2
author LinShunKang <linshunkang.chn@gmail.com> Fri Sep 30 21:45:18 2022 +0800
committer GitHub <noreply@github.com> Fri Sep 30 06:45:18 2022 -0700
tree 8a276dd06ffe67fc06e42af9dd3cc16d037efe2a
parent 3a9efc77b243c720bd1d1b6da4df4128f24a68f7

Fix ByteBufferSerializer#serialize(String, ByteBuffer) not roundtrip input with ByteBufferDeserializer#deserialize(String, byte[]) (#12704)

Reviewers: Guozhang Wang <wangguoz@gmail.com>

clients/src/main/java/org/apache/kafka/common/serialization/ByteBufferSerializer.java

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 5987688..06b66a6 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 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 @@
             }
         }
 
-        final ByteBuffer copyData = data.asReadOnlyBuffer();
-        copyData.flip();
-        return Utils.toArray(copyData);
+        data.flip();
+        return Utils.toArray(data);
     }
 }

clients/src/test/java/org/apache/kafka/common/serialization/SerializationTest.java

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 eb1fee3..a0b67a0 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 @@
             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()));
         }