[schema] Add ByteBuf schema and fix ByteBuffer schema (apache#2624)

*Motivation*

ByteBuffer is a variant of `bytes`. so it should be `SchemaType.BYTES`, not `SchemaType.BYTEBUFFER`

*Changes*

- Fix bytebuffer schema type
- Add bytebuf schema

pulsar-client-schema/src/main/java/org/apache/pulsar/client/impl/schema/ByteBufSchema.java

@@ -0,0 +1,65 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl.schema;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.buffer.Unpooled;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.schema.SchemaInfo;
+import org.apache.pulsar.common.schema.SchemaType;
+
+/**
+ * A variant `Bytes` schema that takes {@link io.netty.buffer.ByteBuf}.
+ */
+public class ByteBufSchema implements Schema<ByteBuf> {
+
+    public static ByteBufSchema of() {
+        return INSTANCE;
+    }
+
+    private static final ByteBufSchema INSTANCE = new ByteBufSchema();
+    private static final SchemaInfo SCHEMA_INFO = new SchemaInfo()
+        .setName("ByteBuf")
+        .setType(SchemaType.BYTES)
+        .setSchema(new byte[0]);
+
+    @Override
+    public byte[] encode(ByteBuf message) {
+        if (message == null) {
+            return null;
+        }
+
+        return ByteBufUtil.getBytes(message);
+    }
+
+    @Override
+    public ByteBuf decode(byte[] bytes) {
+        if (null == bytes) {
+            return null;
+        } else {
+            return Unpooled.wrappedBuffer(bytes);
+        }
+    }
+
+    @Override
+    public SchemaInfo getSchemaInfo() {
+        return SCHEMA_INFO;
+    }
+}

pulsar-client-schema/src/main/java/org/apache/pulsar/client/impl/schema/ByteBufferSchema.java

@@ -24,7 +24,7 @@
 import org.apache.pulsar.common.schema.SchemaType;
 
 /**
- * A bytebuffer schema.
+ * A bytebuffer schema is effectively a `BYTES` schema.
  */
 public class ByteBufferSchema implements Schema<ByteBuffer> {
 
@@ -35,7 +35,7 @@ public static ByteBufferSchema of() {
     private static final ByteBufferSchema INSTANCE = new ByteBufferSchema();
     private static final SchemaInfo SCHEMA_INFO = new SchemaInfo()
         .setName("ByteBuffer")
-        .setType(SchemaType.BYTEBUFFER)
+        .setType(SchemaType.BYTES)
         .setSchema(new byte[0]);
 
     @Override

pulsar-client-schema/src/test/java/org/apache/pulsar/client/schema/PrimitiveSchemaTest.java

@@ -18,16 +18,19 @@
  */
 package org.apache.pulsar.client.schema;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNull;
 
+import io.netty.buffer.Unpooled;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.schema.ByteBufSchema;
 import org.apache.pulsar.client.impl.schema.ByteBufferSchema;
 import org.apache.pulsar.client.impl.schema.ByteSchema;
 import org.apache.pulsar.client.impl.schema.BytesSchema;
@@ -55,8 +58,9 @@ public class PrimitiveSchemaTest {
             put(LongSchema.of(), Arrays.asList(922337203685477580L, -922337203685477581L));
             put(FloatSchema.of(), Arrays.asList(5678567.12312f, -5678567.12341f));
             put(DoubleSchema.of(), Arrays.asList(5678567.12312d, -5678567.12341d));
-            put(BytesSchema.of(), Arrays.asList("my string".getBytes()));
-            put(ByteBufferSchema.of(), Arrays.asList(ByteBuffer.allocate(10).put("my string".getBytes())));
+            put(BytesSchema.of(), Arrays.asList("my string".getBytes(UTF_8)));
+            put(ByteBufferSchema.of(), Arrays.asList(ByteBuffer.allocate(10).put("my string".getBytes(UTF_8))));
+            put(ByteBufSchema.of(), Arrays.asList(Unpooled.wrappedBuffer("my string".getBytes(UTF_8))));
         }
     };
 
@@ -94,8 +98,8 @@ public void allSchemasShouldHaveSchemaType() {
         assertEquals(SchemaType.DOUBLE, DoubleSchema.of().getSchemaInfo().getType());
         assertEquals(SchemaType.STRING, StringSchema.utf8().getSchemaInfo().getType());
         assertEquals(SchemaType.BYTES, BytesSchema.of().getSchemaInfo().getType());
-        assertEquals(SchemaType.BYTEBUFFER, ByteBufferSchema.of().getSchemaInfo().getType());
-
+        assertEquals(SchemaType.BYTES, ByteBufferSchema.of().getSchemaInfo().getType());
+        assertEquals(SchemaType.BYTES, ByteBufSchema.of().getSchemaInfo().getType());
     }
 
 

pulsar-common/src/main/java/org/apache/pulsar/common/schema/SchemaType.java

@@ -67,11 +67,6 @@ public enum SchemaType {
      */
     BYTES,
 
-    /**
-     * A bytebuffer.
-     */
-    BYTEBUFFER,
-
     /**
      * JSON object encoding and validation
      */

tests/scripts/pre-integ-tests.sh

@@ -24,7 +24,7 @@ ulimit -a
 pwd
 df -h
 ps -eo euser,pid,ppid,pgid,start,pcpu,pmem,cmd
-docker network prune -f --filter name=pulsarnet_*
+docker system prune -f
 docker system events > docker.debug-info & echo $! > docker-log.pid
 docker pull apachepulsar/s3mock:latest
 docker pull alpine/socat:latest