KAFKA-15602: revert KAFKA-4852 (apache#14617)

This PR reverts
 - apache@51dbd17
 - apache@496ae05

Reviewers:  Philip Nee <pnee@confluent.io>, Guozhang Wang <guozhang@confluent.io>

Author: mjsax

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

@@ -16,38 +16,34 @@
  */
 package org.apache.kafka.common.serialization;
 
-import org.apache.kafka.common.utils.Utils;
-
 import java.nio.ByteBuffer;
 
 /**
- * Do not need to flip before call <i>serialize(String, ByteBuffer)</i>. For example:
- *
- * <blockquote>
- * <pre>
- * ByteBufferSerializer serializer = ...; // Create Serializer
- * ByteBuffer buffer = ...;               // Allocate ByteBuffer
- * buffer.put(data);                      // Put data into buffer, do not need to flip
- * serializer.serialize(topic, buffer);   // Serialize buffer
- * </pre>
- * </blockquote>
+ * {@code ByteBufferSerializer} always {@link ByteBuffer#rewind() rewinds} the position of the input buffer to zero for
+ * serialization. A manual rewind is not necessary.
+ * <p>
+ * Note: any existing buffer position is ignored.
+ * <p>
+ * The position is also rewound back to zero before {@link #serialize(String, ByteBuffer)}
+ * returns.
  */
 public class ByteBufferSerializer implements Serializer<ByteBuffer> {
-
-    @Override
     public byte[] serialize(String topic, ByteBuffer data) {
-        if (data == null) {
+        if (data == null)
             return null;
-        }
+
+        data.rewind();
 
         if (data.hasArray()) {
-            final byte[] arr = data.array();
+            byte[] arr = data.array();
             if (data.arrayOffset() == 0 && arr.length == data.remaining()) {
                 return arr;
             }
         }
 
-        data.flip();
-        return Utils.toArray(data);
+        byte[] ret = new byte[data.remaining()];
+        data.get(ret, 0, ret.length);
+        data.rewind();
+        return ret;
     }
 }

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

@@ -406,23 +406,6 @@ private Serde<String> getStringSerde(String encoder) {
         return Serdes.serdeFrom(serializer, deserializer);
     }
 
-    @Test
-    public void testByteBufferSerializer() {
-        final byte[] bytes = "Hello".getBytes(UTF_8);
-        final ByteBuffer heapBuffer0 = ByteBuffer.allocate(bytes.length + 1).put(bytes);
-        final ByteBuffer heapBuffer1 = ByteBuffer.allocate(bytes.length).put(bytes);
-        final ByteBuffer heapBuffer2 = ByteBuffer.wrap(bytes);
-        final ByteBuffer directBuffer0 = ByteBuffer.allocateDirect(bytes.length + 1).put(bytes);
-        final ByteBuffer directBuffer1 = ByteBuffer.allocateDirect(bytes.length).put(bytes);
-        try (final ByteBufferSerializer serializer = new ByteBufferSerializer()) {
-            assertArrayEquals(bytes, serializer.serialize(topic, heapBuffer0));
-            assertArrayEquals(bytes, serializer.serialize(topic, heapBuffer1));
-            assertArrayEquals(bytes, serializer.serialize(topic, heapBuffer2));
-            assertArrayEquals(bytes, serializer.serialize(topic, directBuffer0));
-            assertArrayEquals(bytes, serializer.serialize(topic, directBuffer1));
-        }
-    }
-
     @ParameterizedTest
     @ValueSource(booleans = { true, false })
     public void testBooleanSerializer(Boolean dataToSerialize) {