@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();
  }
}