From 676ee3810adc773a421e4883cb159211d61ace3f Mon Sep 17 00:00:00 2001 From: Neng Lu Date: Thu, 20 Feb 2020 11:10:50 -0800 Subject: [PATCH 1/2] encode null key/value in KeyValue --- .../apache/pulsar/common/schema/KeyValue.java | 59 +++++++++++++++---- .../client/impl/TypedMessageBuilderImpl.java | 8 ++- 2 files changed, 51 insertions(+), 16 deletions(-) diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/KeyValue.java b/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/KeyValue.java index 9d867582ccd6f..e038225863b74 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/KeyValue.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/KeyValue.java @@ -60,11 +60,19 @@ public boolean equals(Object obj) { @Override public String toString() { StringBuilder sb = new StringBuilder(); - sb.append("(key = \"") - .append(key) - .append("\", value = \"") - .append(value) - .append("\")"); + sb.append("("); + if (null == key) { + sb.append("key is null, "); + } else { + sb.append("key = \"").append(key).append("\", "); + } + + if (null == value) { + sb.append("value is null"); + } else { + sb.append("value = \"").append(value).append("\""); + } + sb.append(")"); return sb.toString(); } @@ -98,8 +106,24 @@ public static byte[] encode(K key, Schema keyWriter, V value, Schema valueWriter) { byte [] keyBytes = keyWriter.encode(key); byte [] valueBytes = valueWriter.encode(value); - ByteBuffer byteBuffer = ByteBuffer.allocate(4 + keyBytes.length + 4 + valueBytes.length); - byteBuffer.putInt(keyBytes.length).put(keyBytes).putInt(valueBytes.length).put(valueBytes); + + int keyLength = 1 + (null == keyBytes ? 0 : 4 + keyBytes.length); + int valueLength = 1 + (null == valueBytes ? 0 : 4 + valueBytes.length); + + ByteBuffer byteBuffer = ByteBuffer.allocate(keyLength + valueLength); + + if (null == keyBytes) { + byteBuffer.put((byte) 1); + } else { + byteBuffer.put((byte) 0).putInt(keyLength).put(keyBytes); + } + + if (null == valueBytes) { + byteBuffer.put((byte) 1); + } else { + byteBuffer.put((byte) 0).putInt(valueLength).put(valueBytes); + } + return byteBuffer.array(); } @@ -112,13 +136,22 @@ public static byte[] encode(K key, Schema keyWriter, */ public static KeyValue decode(byte[] data, KeyValueDecoder decoder) { ByteBuffer byteBuffer = ByteBuffer.wrap(data); - int keyLength = byteBuffer.getInt(); - byte[] keyBytes = new byte[keyLength]; - byteBuffer.get(keyBytes); - int valueLength = byteBuffer.getInt(); - byte[] valueBytes = new byte[valueLength]; - byteBuffer.get(valueBytes); + byte[] keyBytes = null; + byte isKeyNull = byteBuffer.get(); + if (isKeyNull == (byte) 0) { + int keyLength = byteBuffer.getInt(); + keyBytes = new byte[keyLength]; + byteBuffer.get(keyBytes); + } + + byte[] valueBytes = null; + byte isValueNull = byteBuffer.get(); + if (isValueNull == (byte) 0) { + int valueLength = byteBuffer.getInt(); + valueBytes = new byte[valueLength]; + byteBuffer.get(valueBytes); + } return decoder.decode(keyBytes, valueBytes); } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java index 8d7884954bd9a..396075cf6cbbe 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java @@ -144,9 +144,11 @@ public TypedMessageBuilder value(T value) { org.apache.pulsar.common.schema.KeyValue kv = (org.apache.pulsar.common.schema.KeyValue) value; if (kvSchema.getKeyValueEncodingType() == KeyValueEncodingType.SEPARATED) { // set key as the message key - msgMetadataBuilder.setPartitionKey( - Base64.getEncoder().encodeToString(kvSchema.getKeySchema().encode(kv.getKey()))); - msgMetadataBuilder.setPartitionKeyB64Encoded(true); + if (kv.getKey() != null) { + msgMetadataBuilder.setPartitionKey( + Base64.getEncoder().encodeToString(kvSchema.getKeySchema().encode(kv.getKey()))); + msgMetadataBuilder.setPartitionKeyB64Encoded(true); + } // set value as the payload this.content = ByteBuffer.wrap(kvSchema.getValueSchema().encode(kv.getValue())); return this; From 970fcbf7ffe65072cd8195ee4643eaf0316910ea Mon Sep 17 00:00:00 2001 From: Neng Lu Date: Fri, 21 Feb 2020 21:58:06 -0800 Subject: [PATCH 2/2] fix key/value length when encoding --- .../apache/pulsar/common/schema/KeyValue.java | 25 +++++++++---------- 1 file changed, 12 insertions(+), 13 deletions(-) diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/KeyValue.java b/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/KeyValue.java index e038225863b74..f17d909e8da3c 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/KeyValue.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/KeyValue.java @@ -107,21 +107,20 @@ public static byte[] encode(K key, Schema keyWriter, byte [] keyBytes = keyWriter.encode(key); byte [] valueBytes = valueWriter.encode(value); - int keyLength = 1 + (null == keyBytes ? 0 : 4 + keyBytes.length); - int valueLength = 1 + (null == valueBytes ? 0 : 4 + valueBytes.length); - - ByteBuffer byteBuffer = ByteBuffer.allocate(keyLength + valueLength); + int keyEncodeLength = 1 + (null == keyBytes ? 0 : 4 + keyBytes.length); + int valueEncodeLength = 1 + (null == valueBytes ? 0 : 4 + valueBytes.length); + ByteBuffer byteBuffer = ByteBuffer.allocate(keyEncodeLength + valueEncodeLength); if (null == keyBytes) { - byteBuffer.put((byte) 1); + byteBuffer.put((byte) 0); } else { - byteBuffer.put((byte) 0).putInt(keyLength).put(keyBytes); + byteBuffer.put((byte) 1).putInt(keyBytes.length).put(keyBytes); } if (null == valueBytes) { - byteBuffer.put((byte) 1); + byteBuffer.put((byte) 0); } else { - byteBuffer.put((byte) 0).putInt(valueLength).put(valueBytes); + byteBuffer.put((byte) 1).putInt(valueBytes.length).put(valueBytes); } return byteBuffer.array(); @@ -138,16 +137,16 @@ public static KeyValue decode(byte[] data, KeyValueDecoder de ByteBuffer byteBuffer = ByteBuffer.wrap(data); byte[] keyBytes = null; - byte isKeyNull = byteBuffer.get(); - if (isKeyNull == (byte) 0) { + byte isKeySet = byteBuffer.get(); + if (isKeySet == (byte) 1) { int keyLength = byteBuffer.getInt(); keyBytes = new byte[keyLength]; byteBuffer.get(keyBytes); } byte[] valueBytes = null; - byte isValueNull = byteBuffer.get(); - if (isValueNull == (byte) 0) { + byte isValueSet = byteBuffer.get(); + if (isValueSet == (byte) 1) { int valueLength = byteBuffer.getInt(); valueBytes = new byte[valueLength]; byteBuffer.get(valueBytes); @@ -155,4 +154,4 @@ public static KeyValue decode(byte[] data, KeyValueDecoder de return decoder.decode(keyBytes, valueBytes); } -} +} \ No newline at end of file