From 8d43375e80ec011526f622f96ff9d8e8c2123dcf Mon Sep 17 00:00:00 2001
From: Jonathan Hsieh <jon@cloudera.com>
Date: Thu, 14 Apr 2011 10:47:14 -0700
Subject: [PATCH] FLUME-414: Improved syntax HBase Sink

---
 .../java/com/cloudera/flume/hbase/HBaseSink.java   |  207 +++++++++++++++
 .../com/cloudera/flume/hbase/TestHBaseSink.java    |  268 ++++++++++++++++++++
 .../cloudera/flume/hbase/TestHBaseSinkBuilder.java |  106 ++++++++
 3 files changed, 581 insertions(+), 0 deletions(-)
 create mode 100644 plugins/hbasesink/src/java/com/cloudera/flume/hbase/HBaseSink.java
 create mode 100644 plugins/hbasesink/src/javatest/com/cloudera/flume/hbase/TestHBaseSink.java
 create mode 100644 plugins/hbasesink/src/javatest/com/cloudera/flume/hbase/TestHBaseSinkBuilder.java

diff --git a/plugins/hbasesink/src/java/com/cloudera/flume/hbase/HBaseSink.java b/plugins/hbasesink/src/java/com/cloudera/flume/hbase/HBaseSink.java
new file mode 100644
index 0000000..50cd30d
--- /dev/null
+++ b/plugins/hbasesink/src/java/com/cloudera/flume/hbase/HBaseSink.java
@@ -0,0 +1,207 @@
+/**
+ * Licensed to Cloudera, Inc. under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  Cloudera, Inc. licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.cloudera.flume.hbase;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Put;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.cloudera.flume.conf.Context;
+import com.cloudera.flume.conf.SinkFactory;
+import com.cloudera.flume.conf.SinkFactory.SinkBuilder;
+import com.cloudera.flume.core.Event;
+import com.cloudera.flume.core.EventSink;
+import com.cloudera.util.Pair;
+import com.google.common.base.Preconditions;
+
+/**
+ * This is a straightforward and completely explicit hbase sink.
+ * 
+ * "writeBufferSize" - If provided, autoFlush for the HTable set to "false", and
+ * writeBufferSize is set to its value. If not provided, by default autoFlush is
+ * set to "true" (default HTable setting). This setting is valuable to boost
+ * HBase write speed. The default is 2MB.
+ * 
+ * "writeToWal" - Determines whether WAL should be used during writing to HBase.
+ * If not provided Puts are written to WAL by default This setting is valuable
+ * to boost HBase write speed, but decreases reliability level. Use it if you
+ * know what it does.
+ * 
+ * The Sink also implements method getSinkBuilders(), so it can be used as
+ * Flume's extension plugin (see flume.plugin.classes property of flume-site.xml
+ * config details)
+ */
+public class HBaseSink extends EventSink.Base {
+  private static final Logger LOG = LoggerFactory.getLogger(HBaseSink.class);
+  public static final String KW_BUFFER_SIZE = "writeBufferSize";
+  public static final String KW_USE_WAL = "writeToWal";
+  public static final String USAGE = "usage: hbase(\"table\", \"rowkey\", "
+      + "\"cf1\"," + " \"c1\", \"val1\"[,\"cf2\", \"c2\", \"val2\", ....]{, "
+      + KW_BUFFER_SIZE + "=int, " + KW_USE_WAL + "=true|false})";
+
+  // triples for what values to write
+  public static class QualifierSpec {
+    String colFam;
+    String col;
+    String value;
+
+    QualifierSpec() {
+    }
+
+    QualifierSpec(String cf, String c, String v) {
+      this.colFam = cf;
+      this.col = c;
+      this.value = v;
+    }
+  };
+
+  final String tableName; // not escapable
+  final String rowkey; // flume escapable string
+  final List<QualifierSpec> spec;
+  final long writeBufferSize;
+  final boolean writeToWal;
+  final Configuration config;
+
+  private HTable table;
+
+  public HBaseSink(String tableName, String rowkey, List<QualifierSpec> spec) {
+    this(tableName, rowkey, spec, 0L, true, HBaseConfiguration.create());
+  }
+
+  public HBaseSink(String tableName, String rowkey, List<QualifierSpec> spec,
+      long writeBufferSize, boolean writeToWal, Configuration config) {
+    Preconditions.checkNotNull(tableName, "Must specify table name.");
+    Preconditions.checkNotNull(spec, "Must specify cols and values to write. ");
+    this.tableName = tableName;
+    this.rowkey = rowkey;
+    this.spec = spec;
+    this.writeBufferSize = writeBufferSize;
+    this.writeToWal = writeToWal;
+    this.config = config;
+  }
+
+  @Override
+  public void append(Event e) throws IOException {
+    String rowVal = e.escapeString(rowkey);
+    Put p = new Put(rowVal.getBytes());
+
+    for (QualifierSpec q : spec) {
+      String cf = q.colFam;
+      String c = e.escapeString(q.col);
+      String val = e.escapeString(q.value);
+      p.add(cf.getBytes(), c.getBytes(), val.getBytes());
+    }
+
+    p.setWriteToWAL(writeToWal);
+    table.put(p);
+  }
+
+  @Override
+  synchronized public void close() throws IOException {
+    if (table != null) {
+      table.close(); // performs flushCommits() internally, so we are good when
+                     // autoFlush=false
+      table = null;
+      LOG.info("HBase sink successfully closed");
+    } else {
+      LOG.warn("Double close of HBase sink");
+    }
+
+  }
+
+  @Override
+  synchronized public void open() throws IOException {
+    if (table != null) {
+      throw new IllegalStateException(
+          "HTable is already initialized. Looks like sink close() hasn't been proceeded properly.");
+    }
+    // This instantiates an HTable object that connects you to
+    // the tableName table.
+    table = new HTable(config, tableName);
+    if (writeBufferSize > 0) {
+      table.setAutoFlush(false);
+      table.setWriteBufferSize(writeBufferSize);
+    }
+    validateColFams(table);
+    LOG.info("HBase sink successfully opened");
+  }
+
+  /**
+   * Column family validity check happens in open(), so we through an
+   * IOException (ideally invalid column families would be a
+   * IllegalArgumentException but this exn doesn't make sense on open)
+   */
+  void validateColFams(HTable ht) throws IOException {
+    for (QualifierSpec q : spec) {
+      String cf = q.colFam;
+      HColumnDescriptor hcd = ht.getTableDescriptor().getFamily(cf.getBytes());
+      // TODO check hbase semantics
+      if (hcd == null) {
+        throw new IOException("The column familiy '" + cf
+            + "' does not exist in table '" + tableName + "'");
+      }
+    }
+  }
+
+  public static SinkBuilder builder() {
+    return new SinkBuilder() {
+
+      @Override
+      public EventSink build(Context context, String... argv) {
+        // at least table, row, and one (cf,c,val)
+        Preconditions.checkArgument(argv.length >= 2 + 3, USAGE);
+        // guarantee table, row plus triples of (cf,c,val)
+        Preconditions.checkArgument((argv.length % 3) == 2, USAGE);
+
+        String tableName = argv[0];
+        String rowKey = argv[1];
+
+        List<QualifierSpec> spec = new ArrayList<QualifierSpec>();
+        for (int i = 2; i < argv.length; i += 3) {
+          QualifierSpec qs = new QualifierSpec();
+          qs.colFam = argv[i];
+          qs.col = argv[i + 1];
+          qs.value = argv[i + 2];
+          spec.add(qs);
+        }
+
+        String bufSzStr = context.getValue(KW_BUFFER_SIZE);
+        String isWriteToWal = context.getValue(KW_USE_WAL);
+        long bufSz = (bufSzStr == null ? 0 : Long.parseLong(bufSzStr));
+
+        return new HBaseSink(tableName, rowKey, spec, bufSz,
+            Boolean.parseBoolean(isWriteToWal), HBaseConfiguration.create());
+      }
+    };
+  }
+
+  @SuppressWarnings("unchecked")
+  public static List<Pair<String, SinkFactory.SinkBuilder>> getSinkBuilders() {
+    return Arrays.asList(new Pair<String, SinkFactory.SinkBuilder>("hbase",
+        builder()));
+  }
+}
diff --git a/plugins/hbasesink/src/javatest/com/cloudera/flume/hbase/TestHBaseSink.java b/plugins/hbasesink/src/javatest/com/cloudera/flume/hbase/TestHBaseSink.java
new file mode 100644
index 0000000..95c0e3b
--- /dev/null
+++ b/plugins/hbasesink/src/javatest/com/cloudera/flume/hbase/TestHBaseSink.java
@@ -0,0 +1,268 @@
+/**
+ * Licensed to Cloudera, Inc. under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  Cloudera, Inc. licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.cloudera.flume.hbase;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import junit.framework.Assert;
+
+import org.apache.hadoop.hbase.HBaseTestCase;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.TableNotFoundException;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.HBaseAdmin;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.cloudera.flume.core.Event;
+import com.cloudera.flume.core.Event.Priority;
+import com.cloudera.flume.core.EventImpl;
+import com.cloudera.flume.hbase.HBaseSink.QualifierSpec;
+import com.cloudera.util.Clock;
+
+/**
+ * Test the hbase sink writes events to a table/family properly
+ */
+public class TestHBaseSink {
+  private static HBaseTestEnv hbaseEnv;
+  public static Logger LOG = LoggerFactory.getLogger(TestHBaseSink.class);
+  public static final String DEFAULT_HOST = "qwigibo";
+
+  @BeforeClass
+  public static void setup() throws Exception {
+    // expensive, so just do it once for all tests, just make sure
+    // that tests don't overlap (use diff tables for each test)
+    hbaseEnv = new HBaseTestEnv();
+    hbaseEnv.conf.set(HBaseTestCase.TEST_DIRECTORY_KEY, "build/test/data");
+    hbaseEnv.setUp();
+  }
+
+  @AfterClass
+  public static void teardown() throws Exception {
+    hbaseEnv.tearDown();
+  }
+
+  void shipThreeEvents(HBaseSink snk) throws IOException {
+    snk.open();
+    try {
+      Event e1 = new EventImpl("message0".getBytes(), Clock.unixTime(),
+          Priority.INFO, 0, DEFAULT_HOST);
+      e1.set("rowkey", Bytes.toBytes("row-key0"));
+      e1.set("attr1", Bytes.toBytes("attr1_val0"));
+      e1.set("attr2", Bytes.toBytes("attr2_val0"));
+      e1.set("other", Bytes.toBytes("other_val0"));
+      snk.append(e1);
+
+      Event e2 = new EventImpl("message1".getBytes(), Clock.unixTime(),
+          Priority.INFO, 1, DEFAULT_HOST);
+      e2.set("rowkey", Bytes.toBytes("row-key1"));
+      e2.set("attr1", Bytes.toBytes("attr1_val1"));
+      e2.set("attr2", Bytes.toBytes("attr2_val1"));
+      e2.set("other", Bytes.toBytes("other_val1"));
+      snk.append(e2);
+
+      Event e3 = new EventImpl("message2".getBytes(), Clock.unixTime(),
+          Priority.INFO, 2, DEFAULT_HOST);
+      e3.set("rowkey", Bytes.toBytes("row-key2"));
+      e3.set("attr1", Bytes.toBytes("attr1_val2"));
+      e3.set("attr2", Bytes.toBytes("attr2_val2"));
+      e3.set("other", Bytes.toBytes("other_val2"));
+      snk.append(e3);
+    } finally {
+      snk.close();
+    }
+  }
+
+  /**
+   * Write events to a sink directly, verify by scanning HBase table. x
+   */
+  @Test
+  public void testSink() throws IOException, InterruptedException {
+    final String tableName = "testSink";
+    final String tableFamily1 = "family1";
+    final String tableFamily2 = "family2";
+
+    // create the table and column family to be used by sink
+    HTableDescriptor desc = new HTableDescriptor(tableName);
+    desc.addFamily(new HColumnDescriptor(tableFamily1));
+    desc.addFamily(new HColumnDescriptor(tableFamily2));
+    HBaseAdmin admin = new HBaseAdmin(hbaseEnv.conf);
+    admin.createTable(desc);
+
+    // explicit constructor rather than builder - we want to control the conf
+    List<QualifierSpec> spec = new ArrayList<QualifierSpec>();
+    spec.add(new QualifierSpec(tableFamily1, "col1", "%{attr1}"));
+    spec.add(new QualifierSpec(tableFamily2, "col2", "%{attr2}"));
+    HBaseSink snk = new HBaseSink(tableName, "%{rowkey}", spec, 0L, false,
+        hbaseEnv.conf);
+    shipThreeEvents(snk);
+
+    // verify that the events made it into hbase
+    HTable table = new HTable(hbaseEnv.conf, tableName);
+    try {
+      for (long i = 0; i <= 2; i++) {
+        Result r = table.get(new Get(Bytes.toBytes("row-key" + i)));
+        LOG.info("result " + r);
+
+        byte[] v1 = r.getValue(Bytes.toBytes(tableFamily1),
+            Bytes.toBytes("col1"));
+        Assert.assertEquals("Matching value 1", "attr1_val" + i,
+            Bytes.toString(v1));
+        byte[] v2 = r.getValue(Bytes.toBytes(tableFamily2),
+            Bytes.toBytes("col2"));
+        Assert.assertEquals("Matching value 2", "attr2_val" + i,
+            Bytes.toString(v2));
+      }
+    } finally {
+      table.close();
+    }
+  }
+
+  /**
+   * Write events to a sink directly, verify by scanning HBase table. x
+   */
+  @Test
+  public void testSinkEmptyCol() throws IOException, InterruptedException {
+    final String tableName = "testSinkEmptyCol";
+    final String tableFamily1 = "family1";
+    final String tableFamily2 = "family2";
+
+    // create the table and column family to be used by sink
+    HTableDescriptor desc = new HTableDescriptor(tableName);
+    desc.addFamily(new HColumnDescriptor(tableFamily1));
+    desc.addFamily(new HColumnDescriptor(tableFamily2));
+    HBaseAdmin admin = new HBaseAdmin(hbaseEnv.conf);
+    admin.createTable(desc);
+
+    // explicit constructor rather than builder - we want to control the conf
+    List<QualifierSpec> spec = new ArrayList<QualifierSpec>();
+    spec.add(new QualifierSpec(tableFamily1, "", "%{attr1}"));
+    spec.add(new QualifierSpec(tableFamily2, "", "%{attr2}"));
+    HBaseSink snk = new HBaseSink(tableName, "%{rowkey}", spec, 0L, false,
+        hbaseEnv.conf);
+    shipThreeEvents(snk);
+
+    // verify that the events made it into hbase
+    HTable table = new HTable(hbaseEnv.conf, tableName);
+    try {
+      for (long i = 0; i <= 2; i++) {
+        Result r = table.get(new Get(Bytes.toBytes("row-key" + i)));
+        LOG.info("result " + r);
+
+        byte[] v1 = r.getValue(Bytes.toBytes(tableFamily1), Bytes.toBytes(""));
+        Assert.assertEquals("Matching value 1", "attr1_val" + i,
+            Bytes.toString(v1));
+        byte[] v2 = r.getValue(Bytes.toBytes(tableFamily2), Bytes.toBytes(""));
+        Assert.assertEquals("Matching value 2", "attr2_val" + i,
+            Bytes.toString(v2));
+      }
+    } finally {
+      table.close();
+    }
+  }
+
+  /**
+   * Write events to a sink directly, verify by scanning HBase table. x
+   */
+  @Test
+  public void testSinkEscaping() throws IOException, InterruptedException {
+    final String tableName = "testSinkEscaping";
+    final String tableFamily1 = "family1";
+    final String tableFamily2 = "family2";
+
+    // create the table and column family to be used by sink
+    HTableDescriptor desc = new HTableDescriptor(tableName);
+    desc.addFamily(new HColumnDescriptor(tableFamily1));
+    desc.addFamily(new HColumnDescriptor(tableFamily2));
+    HBaseAdmin admin = new HBaseAdmin(hbaseEnv.conf);
+    admin.createTable(desc);
+
+    // explicit constructor rather than builder - we want to control the conf
+    List<QualifierSpec> spec = new ArrayList<QualifierSpec>();
+    spec.add(new QualifierSpec(tableFamily1, "%{priority}", "%{body}"));
+    spec.add(new QualifierSpec(tableFamily2, "col2", "%{badescape}"));
+    HBaseSink snk = new HBaseSink(tableName, "%{host}-%{rowkey}", spec, 0L,
+        false, hbaseEnv.conf);
+    shipThreeEvents(snk);
+
+    // verify that the events made it into hbase
+    HTable table = new HTable(hbaseEnv.conf, tableName);
+    try {
+      for (long i = 0; i <= 2; i++) {
+        Result r = table.get(new Get(Bytes.toBytes(DEFAULT_HOST + "-row-key"
+            + i)));
+        LOG.info("result " + r);
+
+        byte[] v1 = r.getValue(Bytes.toBytes(tableFamily1),
+            Bytes.toBytes(Event.Priority.INFO.toString())); // default prio
+        Assert
+            .assertEquals("Matching val 1", "message" + i, Bytes.toString(v1));
+        byte[] v2 = r.getValue(Bytes.toBytes(tableFamily2),
+            Bytes.toBytes("col2"));
+        Assert.assertEquals("Matching val 2", "", Bytes.toString(v2));
+      }
+    } finally {
+      table.close();
+    }
+  }
+
+  @Test(expected = TableNotFoundException.class)
+  public void testOpenFailBadTable() throws IOException {
+    final String tableFamily1 = "family1";
+    final String tableFamily2 = "family2";
+
+    // explicit constructor rather than builder - we want to control the conf
+    List<QualifierSpec> spec = new ArrayList<QualifierSpec>();
+    spec.add(new QualifierSpec(tableFamily1, "col1", "%{attr1}"));
+    spec.add(new QualifierSpec(tableFamily2, "col2", "%{attr2}")); // invalid
+    HBaseSink snk = new HBaseSink("bogus table name", "%{rowkey}", spec, 0L,
+        false, hbaseEnv.conf);
+    shipThreeEvents(snk);
+  }
+
+  @Test(expected = IOException.class)
+  public void testOpenFailBadColFam() throws IOException {
+    final String tableName = "testSinkFailBadColFamf";
+    final String tableFamily1 = "family1";
+    final String tableFamily2 = "family2";
+
+    // create the table and column family to be used by sink
+    HTableDescriptor desc = new HTableDescriptor(tableName);
+    desc.addFamily(new HColumnDescriptor(tableFamily1));
+    HBaseAdmin admin = new HBaseAdmin(hbaseEnv.conf);
+    admin.createTable(desc);
+
+    // explicit constructor rather than builder - we want to control the conf
+    List<QualifierSpec> spec = new ArrayList<QualifierSpec>();
+    spec.add(new QualifierSpec(tableFamily1, "col1", "%{attr1}"));
+    spec.add(new QualifierSpec(tableFamily2, "col2", "%{attr2}")); // invalid
+    HBaseSink snk = new HBaseSink(tableName, "%{rowkey}", spec, 0L, false,
+        hbaseEnv.conf);
+    shipThreeEvents(snk);
+  }
+
+}
diff --git a/plugins/hbasesink/src/javatest/com/cloudera/flume/hbase/TestHBaseSinkBuilder.java b/plugins/hbasesink/src/javatest/com/cloudera/flume/hbase/TestHBaseSinkBuilder.java
new file mode 100644
index 0000000..fb3cd52
--- /dev/null
+++ b/plugins/hbasesink/src/javatest/com/cloudera/flume/hbase/TestHBaseSinkBuilder.java
@@ -0,0 +1,106 @@
+/**
+ * Licensed to Cloudera, Inc. under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  Cloudera, Inc. licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.cloudera.flume.hbase;
+
+import static org.junit.Assert.assertEquals;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.cloudera.flume.conf.FlumeBuilder;
+import com.cloudera.flume.conf.FlumeSpecException;
+import com.cloudera.flume.conf.LogicalNodeContext;
+import com.cloudera.flume.conf.SinkFactory.SinkBuilder;
+import com.cloudera.flume.conf.SinkFactoryImpl;
+import com.cloudera.flume.hbase.HBaseSink.QualifierSpec;
+
+/**
+ * Test the hbase sink writes events to a table/family properly
+ */
+public class TestHBaseSinkBuilder {
+  public static Logger LOG = LoggerFactory
+      .getLogger(TestHBaseSinkBuilder.class);
+
+  @BeforeClass
+  public static void setup() throws Exception {
+    SinkFactoryImpl sf = new SinkFactoryImpl();
+    sf.setSink("hbase", HBaseSink.builder());
+    FlumeBuilder.setSinkFactory(sf);
+  }
+
+  /**
+   * Expect no failures
+   */
+  @Test
+  public void testBuilder() {
+    SinkBuilder esb = HBaseSink.builder();
+    HBaseSink snk = (HBaseSink) esb.build(LogicalNodeContext.testingContext(),
+        "table", "%{rowkey}", "cf1", "col1", "%{attr1}", "cf2", "col2",
+        "%{attr2}");
+    assertEquals(snk.tableName, "table");
+    assertEquals(snk.rowkey, "%{rowkey}");
+
+    QualifierSpec s0 = snk.spec.get(0);
+    assertEquals(s0.colFam, "cf1");
+    assertEquals(s0.col, "col1");
+    assertEquals(s0.value, "%{attr1}");
+
+    QualifierSpec s1 = snk.spec.get(1);
+    assertEquals(s1.colFam, "cf2");
+    assertEquals(s1.col, "col2");
+    assertEquals(s1.value, "%{attr2}");
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testFailNotTriples() {
+    SinkBuilder esb = HBaseSink.builder();
+    HBaseSink snk = (HBaseSink) esb.build(LogicalNodeContext.testingContext(),
+        "table", "%{rowkey}", "cf1", "col1", "%{attr1}", "cf2", "col2");
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testFailNotEnough() {
+    SinkBuilder esb = HBaseSink.builder();
+    HBaseSink snk = (HBaseSink) esb.build(LogicalNodeContext.testingContext(),
+        "table");
+  }
+
+  @Test
+  public void testKwArgs() throws FlumeSpecException {
+    HBaseSink s = (HBaseSink) FlumeBuilder.buildSink(
+        LogicalNodeContext.testingContext(), "hbase(\"table\","
+            + "\"%{rowkey}\"," + "\"cf1\", \"col1\", \"%{attr1}\","
+            + "writeBufferSize=10000, writeToWal=true)");
+    assertEquals(s.writeBufferSize, 10000);
+    assertEquals(s.writeToWal, true);
+
+    s = (HBaseSink) FlumeBuilder.buildSink(LogicalNodeContext.testingContext(),
+        "hbase(\"table\"," + "\"%{rowkey}\","
+            + "\"cf1\", \"col1\", \"%{attr1}\"," + "writeToWal=false)");
+    assertEquals(s.writeBufferSize, 0);
+    assertEquals(s.writeToWal, false);
+
+    s = (HBaseSink) FlumeBuilder.buildSink(LogicalNodeContext.testingContext(),
+        "hbase(\"table\"," + "\"%{rowkey}\","
+            + "\"cf1\", \"col1\", \"%{attr1}\"," + "writeBufferSize=10000)");
+    assertEquals(s.writeBufferSize, 10000);
+    assertEquals(s.writeToWal, false);
+  }
+}
-- 
1.7.1

