content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF 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 org.apache.hadoop.chukwa.datacollection.writer;
  
  import java.util.ArrayList;
  
  import junit.framework.Assert;
  import junit.framework.TestCase;
  
  import org.apache.hadoop.chukwa.Chunk;
  import org.apache.hadoop.chukwa.ChunkImpl;
  import org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
  import org.apache.hadoop.chukwa.datacollection.writer.hbase.HBaseWriter;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.hbase.HBaseTestingUtility;
  import org.apache.hadoop.hbase.HColumnDescriptor;
  import org.apache.hadoop.hbase.HTableDescriptor;
  import org.apache.hadoop.hbase.TableName;
  import org.apache.hadoop.hbase.client.Result;
  import org.apache.hadoop.hbase.client.ResultScanner;
  import org.apache.hadoop.hbase.client.Table;
  import org.apache.hadoop.hbase.util.Bytes;
  import org.apache.log4j.Logger;
  
  
  public class TestHBaseWriter extends TestCase{
    static Logger log = Logger.getLogger(TestHBaseWriter.class);
    private HBaseTestingUtility util;
    private HBaseWriter hbw;
    private Configuration conf;
    private byte[] columnFamily = Bytes.toBytes("TestColumnFamily");
    private byte[] qualifier = Bytes.toBytes("Key");
    private byte[] expectedValue = Bytes.toBytes("Value");
  
    private byte[] table = Bytes.toBytes("Test");
    private byte[] test = Bytes.toBytes("1234567890 Key Value");
    private ChukwaConfiguration cc;
    long timestamp = 1234567890;
    
    public TestHBaseWriter() {
      cc = new ChukwaConfiguration();
    }
    
    public void setUp() {
      try {
        util = new HBaseTestingUtility();
        util.startMiniZKCluster();
        util.startMiniCluster();
        conf = util.getConfiguration();
        HTableDescriptor desc = new HTableDescriptor(TableName.valueOf("Test"));
        HColumnDescriptor family = new HColumnDescriptor(columnFamily);
        desc.addFamily(family);
        util.getHBaseAdmin().createTable(desc);
  
      } catch (Exception e) {
        e.printStackTrace();
        Assert.fail(e.getMessage());
      }    
    }
    
    public void tearDown() throws Exception {
      util.shutdownMiniCluster();
      util.shutdownMiniZKCluster();
    }
    
    public void testWriters() {
      ArrayList<Chunk> chunks = new ArrayList<Chunk>();
      chunks.add(new ChunkImpl("TextParser", "name", timestamp, test, null));      
      try {      
        cc.set("hbase.demux.package", "org.apache.chukwa.datacollection.writer.test.demux");
        cc.set("TextParser","org.apache.hadoop.chukwa.datacollection.writer.test.demux.TextParser");
        hbw = new HBaseWriter(cc, conf);
        hbw.init(cc);
        if(hbw.add(chunks)!=ChukwaWriter.COMMIT_OK) {
          Assert.fail("Commit status is not OK.");
        }
        Table testTable = util.getConnection().getTable(TableName.valueOf(table));
        ResultScanner scanner = testTable.getScanner(columnFamily, qualifier);
        for(Result res : scanner) {
          Assert.assertEquals(new String(expectedValue), new String(res.getValue(columnFamily, qualifier)));
        }
        // Cleanup and return
        scanner.close();
        testTable.close();
        // Compare data in Hbase with generated chunks
      } catch (Exception e) {
        e.printStackTrace();
        Assert.fail(e.getMessage());
      }
    }
  }
  