Allow a KeyValue to have an empty value.

The code was erroneously refusing to de-serialize a KeyValue with an
empty key, even though this is permitted by HBase.

Optimization: don't allocate new empty byte arrays.

In the code parsing META entries: properly handle `info:server' entries
with an empty value, as can happen during an NSRE caused by a split.

Change-Id: Idfb32815871d1812a075cfcd84b040f868b74a24

src/HBaseClient.java

@@ -1280,7 +1280,8 @@ private RegionClient discoverRegion(final ArrayList<KeyValue> meta_row) {
           return null;
         }
         start_key = tmp[0];
-      } else if (Arrays.equals(SERVER, qualifier)) {
+      } else if (Arrays.equals(SERVER, qualifier)
+                 && kv.value() != EMPTY_ARRAY) {  // Empty during NSRE.
         final byte[] hostport = kv.value();
         int colon = hostport.length - 1;
         for (/**/; colon > 0 /* Can't be at the beginning */; colon--) {
@@ -1303,6 +1304,10 @@ private RegionClient discoverRegion(final ArrayList<KeyValue> meta_row) {
             + Bytes.pretty(hostport), kv);
         }
       }
+      // TODO(tsuna): If this is the parent of a split region, there are two
+      // other KVs that could be useful: `info:splitA' and `info:splitB'.
+      // Need to investigate whether we can use those as a hint to update our
+      // regions_cache with the daughter regions of the split.
     }
     if (start_key == null) {
       throw new BrokenMetaException(null, "It didn't contain any"

src/KeyValue.java

@@ -198,7 +198,7 @@ public static KeyValue fromBuffer(final ChannelBuffer buf,
     HBaseRpc.checkNonEmptyArrayLength(buf, rowkey_length);
     final int value_length = buf.readInt();
     //LOG.debug("value_length="+value_length);
-    HBaseRpc.checkNonEmptyArrayLength(buf, value_length);
+    HBaseRpc.checkArrayLength(buf, value_length);
     final short key_length = buf.readShort();
     //LOG.debug("key_length="+key_length);
     HBaseRpc.checkArrayLength(buf, value_length);
@@ -217,11 +217,13 @@ public static KeyValue fromBuffer(final ChannelBuffer buf,
     final int qual_length = (rowkey_length - key_length - family_length
                              - 2 - 1 - 8 - 1);
     HBaseRpc.checkArrayLength(buf, qual_length);
-    final byte[] qualifier = new byte[qual_length];
+    final byte[] qualifier = (qual_length > 0 ? new byte[qual_length]
+                              : HBaseClient.EMPTY_ARRAY);
     buf.readBytes(qualifier);
     final long timestamp = buf.readLong();
     final byte key_type = buf.readByte();
-    final byte[] value = new byte[value_length];
+    final byte[] value = (value_length > 0 ? new byte[value_length]
+                          : HBaseClient.EMPTY_ARRAY);
     buf.readBytes(value);
     if (2 + key_length + 1 + family_length + qual_length + 8 + 1
         != rowkey_length) {  // XXX TMP DEBUG