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