Fix KeyValueSchema

apache · May 10, 2021 · b585bb2 · b585bb2
1 parent 2787101
commit b585bb2
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 9 deletions.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java
@@ -835,11 +835,10 @@ public void testAutoKeyValueConsumeGenericObject() throws Exception {
                     assertEquals(data.getValue().getField("i"), i * 1000);
                     assertEquals(data.getKey().getField("j"), i);
                     assertEquals(data.getValue().getField("j"), i * 20);
+                    KeyValueSchema keyValueSchema = (KeyValueSchema) data.getReaderSchema().get();
+                    assertNotNull(keyValueSchema.getKeySchema());
+                    assertNotNull(keyValueSchema.getValueSchema());
                 }
-
-                KeyValueSchema keyValueSchema = (KeyValueSchema) data.getReaderSchema().get();
-                assertNotNull(keyValueSchema.getKeySchema());
-                assertNotNull(keyValueSchema.getValueSchema());
             }
         }
     }

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/KeyValueSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/KeyValueSchema.java
@@ -142,16 +142,27 @@ public byte[] encode(KeyValue<K, V> message) {
         }
     }
 
-    public KeyValue<K, V> decode(ByteBuf byteBuf) {
-        return decode(byteBuf, null);
+    @Override
+    public KeyValue<K, V> decode(byte[] bytes) {
+        return decode(bytes, null);
     }
 
-    public KeyValue<K, V> decode(ByteBuf bytes, byte[] schemaVersion) {
+    @Override
+    public KeyValue<K, V> decode(byte[] bytes, byte[] schemaVersion) {
         if (this.keyValueEncodingType == KeyValueEncodingType.SEPARATED) {
             throw new SchemaSerializationException("This method cannot be used under this SEPARATED encoding type");
         }
-        byte[] array = ByteBufUtil.getBytes(bytes);
-        return KeyValue.decode(array, (keyBytes, valueBytes) -> decode(keyBytes, valueBytes, schemaVersion));
+        return KeyValue.decode(bytes, (keyBytes, valueBytes) -> decode(keyBytes, valueBytes, schemaVersion));
+    }
+
+    @Override
+    public KeyValue<K, V> decode(ByteBuf byteBuf) {
+        return decode(ByteBufUtil.getBytes(byteBuf));
+    }
+
+    @Override
+    public T decode(ByteBuf byteBuf, byte[] schemaVersion) {
+        return decode(ByteBufUtil.getBytes(byteBuf), schemaVersion);
     }
 
     public KeyValue<K, V> decode(byte[] keyBytes, byte[] valueBytes, byte[] schemaVersion) {