HBaseSink extends EventSink.Base { @Override synchronized public void close() throws IOException { if (table != null) { table.close(); table = null; LOG.info("HBase sink successfully closed"); } else { LOG.warn("Double close of HBase sink"); } } HBaseSink(String tableName, String rowkey, List<QualifierSpec> spec);  HBaseSink(String tableName, String rowkey, List<QualifierSpec> spec,
      long writeBufferSize, boolean writeToWal, Configuration config); @Override void append(Event e); @Override synchronized void close(); @Override synchronized void open(); static SinkBuilder builder(); @SuppressWarnings("unchecked") static List<Pair<String, SinkFactory.SinkBuilder>> getSinkBuilders(); static final String KW_BUFFER_SIZE; static final String KW_USE_WAL; static final String USAGE; }
@Test public void testSink() throws IOException, InterruptedException { final String tableName = "testSink"; final String tableFamily1 = "family1"; final String tableFamily2 = "family2"; HTableDescriptor desc = new HTableDescriptor(tableName); desc.addFamily(new HColumnDescriptor(tableFamily1)); desc.addFamily(new HColumnDescriptor(tableFamily2)); HBaseAdmin admin = new HBaseAdmin(hbaseEnv.conf); admin.createTable(desc); 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); 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(); } }
@Test public void testSinkEmptyCol() throws IOException, InterruptedException { final String tableName = "testSinkEmptyCol"; final String tableFamily1 = "family1"; final String tableFamily2 = "family2"; HTableDescriptor desc = new HTableDescriptor(tableName); desc.addFamily(new HColumnDescriptor(tableFamily1)); desc.addFamily(new HColumnDescriptor(tableFamily2)); HBaseAdmin admin = new HBaseAdmin(hbaseEnv.conf); admin.createTable(desc); 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); 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(); } }
@Test public void testSinkEscaping() throws IOException, InterruptedException { final String tableName = "testSinkEscaping"; final String tableFamily1 = "family1"; final String tableFamily2 = "family2"; HTableDescriptor desc = new HTableDescriptor(tableName); desc.addFamily(new HColumnDescriptor(tableFamily1)); desc.addFamily(new HColumnDescriptor(tableFamily2)); HBaseAdmin admin = new HBaseAdmin(hbaseEnv.conf); admin.createTable(desc); 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); 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())); 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(); } }