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..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
@@ -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,23 @@ public static <K, V> byte[] encode(K key, Schema<K> keyWriter,
                                        V value, Schema<V> 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 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) 0);
+        } else {
+            byteBuffer.put((byte) 1).putInt(keyBytes.length).put(keyBytes);
+        }
+
+        if (null == valueBytes) {
+            byteBuffer.put((byte) 0);
+        } else {
+            byteBuffer.put((byte) 1).putInt(valueBytes.length).put(valueBytes);
+        }
+
         return byteBuffer.array();
     }
 
@@ -112,14 +135,23 @@ public static <K, V> byte[] encode(K key, Schema<K> keyWriter,
      */
     public static <K, V> KeyValue<K, V> decode(byte[] data, KeyValueDecoder<K, V> 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 isKeySet = byteBuffer.get();
+        if (isKeySet == (byte) 1) {
+            int keyLength = byteBuffer.getInt();
+            keyBytes = new byte[keyLength];
+            byteBuffer.get(keyBytes);
+        }
+
+        byte[] valueBytes = null;
+        byte isValueSet = byteBuffer.get();
+        if (isValueSet == (byte) 1) {
+            int valueLength = byteBuffer.getInt();
+            valueBytes = new byte[valueLength];
+            byteBuffer.get(valueBytes);
+        }
 
         return decoder.decode(keyBytes, valueBytes);
     }
-}
+}
\ No newline at end of file
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<T> 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;