Fixes bug resolving lock w/ empty col fam. Reported in #660 (#1123)

There was a bug where if a column family was empty and the qual was not
empty this would cause lock recovery to fail.  The underlying cause was
a bug in the Column class.  This class has an isFamilySet() method that
was returning false when the family was set to the empty string.  This
cause caused lock recovery code to create an incorrect range.

The Column class was relying on internal behavior of the Bytes class
that probably changed and caused this bug.

This commit adds a new IT that recreates this bug.  If the new IT is run
w/o the fix to the Column class then it would fail as follows.

```
Running org.apache.fluo.integration.impl.FailureIT
Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 8.011 sec <<< FAILURE! - in org.apache.fluo.integration.impl.FailureIT
testRecoverEmptyColumn(org.apache.fluo.integration.impl.FailureIT)  Time elapsed: 7.096 sec  <<< ERROR!
java.lang.IllegalStateException: can not abort : bob  bal  5 (UNKNOWN)
	at org.apache.fluo.integration.impl.FailureIT.testRecoverEmptyColumn(FailureIT.java:688)
```

modules/api/src/main/java/org/apache/fluo/api/data/Column.java

@@ -28,26 +28,42 @@
 public final class Column implements Comparable<Column>, Serializable {
 
   private static final long serialVersionUID = 1L;
-  private static final Bytes UNSET = Bytes.of(new byte[0]);
 
-  private Bytes family = UNSET;
-  private Bytes qualifier = UNSET;
-  private Bytes visibility = UNSET;
+  private final Bytes family;
+  private final Bytes qualifier;
+  private final Bytes visibility;
+
+  private final boolean isFamilySet;
+  private final boolean isQualifierSet;
+  private final boolean isVisibilitySet;
+
   private int hashCode = 0;
 
   public static final Column EMPTY = new Column();
 
   /**
    * Creates an empty Column where family, qualifier and visibility are not set
    */
-  public Column() {}
+  public Column() {
+    this.family = Bytes.EMPTY;
+    this.isFamilySet = false;
+    this.qualifier = Bytes.EMPTY;
+    this.isQualifierSet = false;
+    this.visibility = Bytes.EMPTY;
+    this.isVisibilitySet = false;
+  }
 
   /**
    * Creates Column with only a family.
    */
   public Column(Bytes family) {
     Objects.requireNonNull(family, "Family must not be null");
     this.family = family;
+    this.isFamilySet = true;
+    this.qualifier = Bytes.EMPTY;
+    this.isQualifierSet = false;
+    this.visibility = Bytes.EMPTY;
+    this.isVisibilitySet = false;
   }
 
   /**
@@ -64,7 +80,11 @@ public Column(Bytes family, Bytes qualifier) {
     Objects.requireNonNull(family, "Family must not be null");
     Objects.requireNonNull(qualifier, "Qualifier must not be null");
     this.family = family;
+    this.isFamilySet = true;
     this.qualifier = qualifier;
+    this.isQualifierSet = true;
+    this.visibility = Bytes.EMPTY;
+    this.isVisibilitySet = false;
   }
 
   /**
@@ -82,8 +102,11 @@ public Column(Bytes family, Bytes qualifier, Bytes visibility) {
     Objects.requireNonNull(qualifier, "Qualifier must not be null");
     Objects.requireNonNull(visibility, "Visibility must not be null");
     this.family = family;
+    this.isFamilySet = true;
     this.qualifier = qualifier;
+    this.isQualifierSet = true;
     this.visibility = visibility;
+    this.isVisibilitySet = true;
   }
 
   /**
@@ -98,16 +121,13 @@ public Column(CharSequence family, CharSequence qualifier, CharSequence visibili
    * Returns true if family is set
    */
   public boolean isFamilySet() {
-    return family != UNSET;
+    return isFamilySet;
   }
 
   /**
    * Retrieves Column Family (in Bytes). Returns Bytes.EMPTY if not set.
    */
   public Bytes getFamily() {
-    if (!isFamilySet()) {
-      return Bytes.EMPTY;
-    }
     return family;
   }
 
@@ -122,16 +142,13 @@ public String getsFamily() {
    * Returns true if qualifier is set
    */
   public boolean isQualifierSet() {
-    return qualifier != UNSET;
+    return isQualifierSet;
   }
 
   /**
    * Retrieves Column Qualifier (in Bytes). Returns Bytes.EMPTY if not set.
    */
   public Bytes getQualifier() {
-    if (!isQualifierSet()) {
-      return Bytes.EMPTY;
-    }
     return qualifier;
   }
 
@@ -146,16 +163,13 @@ public String getsQualifier() {
    * Returns true if visibility is set.
    */
   public boolean isVisibilitySet() {
-    return visibility != UNSET;
+    return isVisibilitySet;
   }
 
   /**
    * Retrieves Column Visibility (in Bytes). Returns Bytes.EMPTY if not set.
    */
   public Bytes getVisibility() {
-    if (!isVisibilitySet()) {
-      return Bytes.EMPTY;
-    }
     return visibility;
   }
 

modules/api/src/test/java/org/apache/fluo/api/data/ColumnTest.java

@@ -15,6 +15,9 @@
 
 package org.apache.fluo.api.data;
 
+import java.util.Arrays;
+import java.util.List;
+
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -73,6 +76,23 @@ public void testCreation() {
     Assert.assertEquals(Bytes.of("cq3"), col.getQualifier());
     Assert.assertEquals(Bytes.of("cv3"), col.getVisibility());
     Assert.assertEquals(new Column("cf3", "cq3", "cv3"), col);
+
+    // an empty string should always be considered as set, try diff combos of empty and non empty
+    // strings.
+    for (String fam : Arrays.asList("", "f")) {
+      for (String qual : Arrays.asList("", "q")) {
+        for (String vis : Arrays.asList("", "v")) {
+          col = new Column(fam, qual, vis);
+          Assert.assertTrue(col.isFamilySet());
+          Assert.assertTrue(col.isQualifierSet());
+          Assert.assertTrue(col.isVisibilitySet());
+          Assert.assertEquals(Bytes.of(fam), col.getFamily());
+          Assert.assertEquals(Bytes.of(qual), col.getQualifier());
+          Assert.assertEquals(Bytes.of(vis), col.getVisibility());
+          Assert.assertEquals(new Column(fam, qual, vis), col);
+        }
+      }
+    }
   }
 
   @Test

modules/integration-tests/src/main/java/org/apache/fluo/integration/impl/FailureIT.java

@@ -20,10 +20,12 @@
 
 import org.apache.accumulo.core.client.Scanner;
 import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.admin.CompactionConfig;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.security.Authorizations;
 import org.apache.curator.framework.CuratorFramework;
+import org.apache.fluo.accumulo.format.FluoFormatter;
 import org.apache.fluo.accumulo.util.ColumnConstants;
 import org.apache.fluo.accumulo.util.ColumnType;
 import org.apache.fluo.accumulo.util.LongUtil;
@@ -32,6 +34,7 @@
 import org.apache.fluo.api.client.TransactionBase;
 import org.apache.fluo.api.data.Bytes;
 import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.RowColumn;
 import org.apache.fluo.api.exceptions.CommitException;
 import org.apache.fluo.api.exceptions.FluoException;
 import org.apache.fluo.api.observer.Observer;
@@ -652,4 +655,40 @@ private boolean wasRolledBackPrimary(long startTs, String rolledBackRow)
     return sawExpected;
   }
 
+  /*
+   * There was a bug where a locked column with an empty family could not be recovered.
+   */
+  @Test
+  public void testRecoverEmptyColumn() {
+    Column ecol = new Column("", "bal");
+
+    TestTransaction tx1 = new TestTransaction(env);
+
+    tx1.set("bob", ecol, "10");
+    tx1.set("joe", ecol, "20");
+    tx1.set("jill", ecol, "60");
+
+    tx1.done();
+
+    // partially commit a transaction, leaving the row 'joe' with a lock.
+    TestTransaction tx2 = new TestTransaction(env);
+    TestUtil.increment(tx2, "bob", ecol, 5);
+    TestUtil.increment(tx2, "joe", ecol, -5);
+
+    CommitData cd = tx2.createCommitData();
+    RowColumn primary = new RowColumn("bob", ecol);
+    Assert.assertTrue(tx2.preCommit(cd, primary));
+    Stamp commitTs = env.getSharedResources().getOracleClient().getStamp();
+    tx2.commitPrimaryColumn(cd, commitTs);
+    tx2.close();
+
+    // this transaction should roll forward the above transaction
+    TestTransaction tx3 = new TestTransaction(env);
+    Assert.assertEquals("15", tx3.gets("bob", ecol));
+    Assert.assertEquals("15", tx3.gets("joe", ecol));
+    Assert.assertEquals("60", tx3.gets("jill", ecol));
+    tx3.close();
+
+
+  }
 }