/**
 * 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.hcatalog.hbase;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;

import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hive.cli.CliSessionState;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
import org.apache.hadoop.hive.metastore.MetaStoreUtils;
import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hcatalog.cli.HCatDriver;
import org.apache.hcatalog.cli.SemanticAnalysis.HCatSemanticAnalyzer;
import org.apache.hcatalog.common.HCatConstants;
import org.apache.hcatalog.common.HCatException;
import org.apache.hcatalog.common.HCatUtil;
import org.apache.hcatalog.data.HCatRecord;
import org.apache.hcatalog.data.schema.HCatFieldSchema;
import org.apache.hcatalog.data.schema.HCatSchema;
import org.apache.hcatalog.hbase.snapshot.RevisionManager;
import org.apache.hcatalog.hbase.snapshot.RevisionManagerConfiguration;
import org.apache.hcatalog.hbase.snapshot.Transaction;
import org.apache.hcatalog.mapreduce.HCatInputFormat;
import org.apache.hcatalog.mapreduce.InputJobInfo;
import org.apache.hcatalog.mapreduce.PartInfo;
import org.junit.Test;

public class TestHBaseInputFormat extends SkeletonHBaseTest {

    private static HiveConf hcatConf;
    private static HCatDriver hcatDriver;
    private final byte[] FAMILY = Bytes.toBytes("testFamily");
    private final byte[] QUALIFIER1 = Bytes.toBytes("testQualifier1");
    private final byte[] QUALIFIER2 = Bytes.toBytes("testQualifier2");

    public TestHBaseInputFormat() throws Exception {
        hcatConf = getHiveConf();
        hcatConf.set(ConfVars.SEMANTIC_ANALYZER_HOOK.varname,
            HCatSemanticAnalyzer.class.getName());
        URI fsuri = getFileSystem().getUri();
        Path whPath = new Path(fsuri.getScheme(), fsuri.getAuthority(),
            getTestDir());
        hcatConf.set(HiveConf.ConfVars.HADOOPFS.varname, fsuri.toString());
        hcatConf.set(ConfVars.METASTOREWAREHOUSE.varname, whPath.toString());

        //Add hbase properties

        for (Map.Entry<String, String> el : getHbaseConf()) {
            if (el.getKey().startsWith("hbase.")) {
                hcatConf.set(el.getKey(), el.getValue());
            }
        }
        HBaseConfiguration.merge(hcatConf,
            RevisionManagerConfiguration.create());


        SessionState.start(new CliSessionState(hcatConf));
        hcatDriver = new HCatDriver();

    }

    private List<Put> generatePuts(int num, String tableName) throws IOException {

        List<String> columnFamilies = Arrays.asList("testFamily");
        RevisionManager rm = null;
        List<Put> myPuts;
        try {
            rm = HBaseRevisionManagerUtil.getOpenedRevisionManager(hcatConf);
            rm.open();
            myPuts = new ArrayList<Put>();
            for (int i = 1; i <= num; i++) {
                Put put = new Put(Bytes.toBytes("testRow"));
                put.add(FAMILY, QUALIFIER1, i, Bytes.toBytes("textValue-" + i));
                put.add(FAMILY, QUALIFIER2, i, Bytes.toBytes("textValue-" + i));
                myPuts.add(put);
                Transaction tsx = rm.beginWriteTransaction(tableName,
                    columnFamilies);
                rm.commitWriteTransaction(tsx);
            }
        } finally {
            if (rm != null)
                rm.close();
        }

        return myPuts;
    }

    private void populateHBaseTable(String tName, int revisions) throws IOException {
        List<Put> myPuts = generatePuts(revisions, tName);
        HTable table = new HTable(getHbaseConf(), Bytes.toBytes(tName));
        table.put(myPuts);
    }

    private long populateHBaseTableQualifier1(String tName, int value, Boolean commit)
        throws IOException {
        List<String> columnFamilies = Arrays.asList("testFamily");
        RevisionManager rm = null;
        List<Put> myPuts = new ArrayList<Put>();
        long revision;
        try {
            rm = HBaseRevisionManagerUtil.getOpenedRevisionManager(hcatConf);
            rm.open();
            Transaction tsx = rm.beginWriteTransaction(tName, columnFamilies);

            Put put = new Put(Bytes.toBytes("testRow"));
            revision = tsx.getRevisionNumber();
            put.add(FAMILY, QUALIFIER1, revision,
                Bytes.toBytes("textValue-" + value));
            myPuts.add(put);

            // If commit is null it is left as a running transaction
            if (commit != null) {
                if (commit) {
                    rm.commitWriteTransaction(tsx);
                } else {
                    rm.abortWriteTransaction(tsx);
                }
            }
        } finally {
            if (rm != null)
                rm.close();
        }
        HTable table = new HTable(getHbaseConf(), Bytes.toBytes(tName));
        table.put(myPuts);
        return revision;
    }

    @Test
    public void TestHBaseTableReadMR() throws Exception {
        String tableName = newTableName("MyTable");
        String databaseName = newTableName("MyDatabase");
        //Table name will be lower case unless specified by hbase.table.name property
        String hbaseTableName = (databaseName + "." + tableName).toLowerCase();
        String db_dir = getTestDir() + "/hbasedb";

        String dbquery = "CREATE DATABASE IF NOT EXISTS " + databaseName + " LOCATION '"
            + db_dir + "'";
        String tableQuery = "CREATE TABLE " + databaseName + "." + tableName
            + "(key string, testqualifier1 string, testqualifier2 string) STORED BY " +
            "'org.apache.hcatalog.hbase.HBaseHCatStorageHandler'"
            + "TBLPROPERTIES ('hbase.columns.mapping'=':key,testFamily:testQualifier1,testFamily:testQualifier2')";

        CommandProcessorResponse responseOne = hcatDriver.run(dbquery);
        assertEquals(0, responseOne.getResponseCode());
        CommandProcessorResponse responseTwo = hcatDriver.run(tableQuery);
        assertEquals(0, responseTwo.getResponseCode());

        HBaseAdmin hAdmin = new HBaseAdmin(getHbaseConf());
        boolean doesTableExist = hAdmin.tableExists(hbaseTableName);
        assertTrue(doesTableExist);

        populateHBaseTable(hbaseTableName, 5);
        Configuration conf = new Configuration(hcatConf);
        conf.set(HCatConstants.HCAT_KEY_HIVE_CONF,
            HCatUtil.serialize(getHiveConf().getAllProperties()));

        // output settings
        Path outputDir = new Path(getTestDir(), "mapred/testHbaseTableMRRead");
        FileSystem fs = getFileSystem();
        if (fs.exists(outputDir)) {
            fs.delete(outputDir, true);
        }
        // create job
        Job job = new Job(conf, "hbase-mr-read-test");
        job.setJarByClass(this.getClass());
        job.setMapperClass(MapReadHTable.class);
        MapReadHTable.resetCounters();

        job.setInputFormatClass(HCatInputFormat.class);
        InputJobInfo inputJobInfo = InputJobInfo.create(databaseName, tableName,
            null);
        HCatInputFormat.setInput(job, inputJobInfo);
        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job, outputDir);
        job.setMapOutputKeyClass(BytesWritable.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(BytesWritable.class);
        job.setOutputValueClass(Text.class);
        job.setNumReduceTasks(0);
        assertTrue(job.waitForCompletion(true));
        // Note: These asserts only works in case of LocalJobRunner as they run in same jvm.
        // If using MiniMRCluster, the tests will have to be modified.
        assertFalse(MapReadHTable.error);
        assertEquals(MapReadHTable.count, 1);

        String dropTableQuery = "DROP TABLE " + hbaseTableName;
        CommandProcessorResponse responseThree = hcatDriver.run(dropTableQuery);
        assertEquals(0, responseThree.getResponseCode());

        boolean isHbaseTableThere = hAdmin.tableExists(hbaseTableName);
        assertFalse(isHbaseTableThere);

        String dropDB = "DROP DATABASE " + databaseName;
        CommandProcessorResponse responseFour = hcatDriver.run(dropDB);
        assertEquals(0, responseFour.getResponseCode());
    }

    @Test
    public void TestHBaseTableProjectionReadMR() throws Exception {

        String tableName = newTableName("MyTable");
        //Table name as specified by hbase.table.name property
        String hbaseTableName = "MyDB_" + tableName;
        String tableQuery = "CREATE TABLE " + tableName
            + "(key string, testqualifier1 string, testqualifier2 string) STORED BY "
            + "'org.apache.hcatalog.hbase.HBaseHCatStorageHandler'"
            + "TBLPROPERTIES ('hbase.columns.mapping'="
            + "':key,testFamily:testQualifier1,testFamily:testQualifier2',"
            + "'hbase.table.name'='" + hbaseTableName + "')";

        CommandProcessorResponse responseTwo = hcatDriver.run(tableQuery);
        assertEquals(0, responseTwo.getResponseCode());

        HBaseAdmin hAdmin = new HBaseAdmin(getHbaseConf());
        boolean doesTableExist = hAdmin.tableExists(hbaseTableName);
        assertTrue(doesTableExist);

        populateHBaseTable(hbaseTableName, 5);

        Configuration conf = new Configuration(hcatConf);
        conf.set(HCatConstants.HCAT_KEY_HIVE_CONF,
            HCatUtil.serialize(getHiveConf().getAllProperties()));

        // output settings
        Path outputDir = new Path(getTestDir(), "mapred/testHBaseTableProjectionReadMR");
        FileSystem fs = getFileSystem();
        if (fs.exists(outputDir)) {
            fs.delete(outputDir, true);
        }
        // create job
        Job job = new Job(conf, "hbase-column-projection");
        job.setJarByClass(this.getClass());
        job.setMapperClass(MapReadProjHTable.class);
        job.setInputFormatClass(HCatInputFormat.class);
        InputJobInfo inputJobInfo = InputJobInfo.create(
            MetaStoreUtils.DEFAULT_DATABASE_NAME, tableName, null);
        HCatInputFormat.setOutputSchema(job, getProjectionSchema());
        HCatInputFormat.setInput(job, inputJobInfo);
        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job, outputDir);
        job.setMapOutputKeyClass(BytesWritable.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(BytesWritable.class);
        job.setOutputValueClass(Text.class);
        job.setNumReduceTasks(0);
        assertTrue(job.waitForCompletion(true));
        assertFalse(MapReadProjHTable.error);
        assertEquals(MapReadProjHTable.count, 1);

        String dropTableQuery = "DROP TABLE " + tableName;
        CommandProcessorResponse responseThree = hcatDriver.run(dropTableQuery);
        assertEquals(0, responseThree.getResponseCode());

        boolean isHbaseTableThere = hAdmin.tableExists(hbaseTableName);
        assertFalse(isHbaseTableThere);
    }

    @Test
    public void TestHBaseInputFormatProjectionReadMR() throws Exception {

        String tableName = newTableName("mytable");
        String tableQuery = "CREATE TABLE " + tableName
            + "(key string, testqualifier1 string, testqualifier2 string) STORED BY " +
            "'org.apache.hcatalog.hbase.HBaseHCatStorageHandler'"
            + "TBLPROPERTIES ('hbase.columns.mapping'=':key," +
            "testFamily:testQualifier1,testFamily:testQualifier2')";

        CommandProcessorResponse responseTwo = hcatDriver.run(tableQuery);
        assertEquals(0, responseTwo.getResponseCode());

        HBaseAdmin hAdmin = new HBaseAdmin(getHbaseConf());
        boolean doesTableExist = hAdmin.tableExists(tableName);
        assertTrue(doesTableExist);

        populateHBaseTable(tableName, 5);

        Configuration conf = new Configuration(hcatConf);
        conf.set(HCatConstants.HCAT_KEY_HIVE_CONF,
            HCatUtil.serialize(getHiveConf().getAllProperties()));

        // output settings
        Path outputDir = new Path(getTestDir(), "mapred/testHBaseTableProjectionReadMR");
        FileSystem fs = getFileSystem();
        if (fs.exists(outputDir)) {
            fs.delete(outputDir, true);
        }
        // create job
        JobConf job = new JobConf(conf);
        job.setJobName("hbase-scan-column");
        job.setJarByClass(this.getClass());
        job.setMapperClass(MapReadProjectionHTable.class);
        job.setInputFormat(HBaseInputFormat.class);

        InputJobInfo inputJobInfo = InputJobInfo.create(
            MetaStoreUtils.DEFAULT_DATABASE_NAME, tableName, null);
        //Configure projection schema
        job.set(HCatConstants.HCAT_KEY_OUTPUT_SCHEMA, HCatUtil.serialize(getProjectionSchema()));
        Job newJob = new Job(job);
        HCatInputFormat.setInput(newJob, inputJobInfo);
        String inputJobString = newJob.getConfiguration().get(HCatConstants.HCAT_KEY_JOB_INFO);
        InputJobInfo info = (InputJobInfo) HCatUtil.deserialize(inputJobString);
        job.set(HCatConstants.HCAT_KEY_JOB_INFO, inputJobString);
        for (PartInfo partinfo : info.getPartitions()) {
            for (Entry<String, String> entry : partinfo.getJobProperties().entrySet())
                job.set(entry.getKey(), entry.getValue());
        }
        assertEquals("testFamily:testQualifier1", job.get(TableInputFormat.SCAN_COLUMNS));

        job.setOutputFormat(org.apache.hadoop.mapred.TextOutputFormat.class);
        org.apache.hadoop.mapred.TextOutputFormat.setOutputPath(job, outputDir);
        job.setMapOutputKeyClass(BytesWritable.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(BytesWritable.class);
        job.setOutputValueClass(Text.class);
        job.setNumReduceTasks(0);

        RunningJob runJob = JobClient.runJob(job);
        runJob.waitForCompletion();
        assertTrue(runJob.isSuccessful());
        assertFalse(MapReadProjHTable.error);
        assertEquals(MapReadProjHTable.count, 1);

        String dropTableQuery = "DROP TABLE " + tableName;
        CommandProcessorResponse responseThree = hcatDriver.run(dropTableQuery);
        assertEquals(0, responseThree.getResponseCode());

        boolean isHbaseTableThere = hAdmin.tableExists(tableName);
        assertFalse(isHbaseTableThere);
    }

    @Test
    public void TestHBaseTableIgnoreAbortedTransactions() throws Exception {
        String tableName = newTableName("mytable");
        String tableQuery = "CREATE TABLE " + tableName
            + "(key string, testqualifier1 string, testqualifier2 string) STORED BY " +
            "'org.apache.hcatalog.hbase.HBaseHCatStorageHandler'"
            + "TBLPROPERTIES ('hbase.columns.mapping'=':key," +
            "testFamily:testQualifier1,testFamily:testQualifier2')";

        CommandProcessorResponse responseTwo = hcatDriver.run(tableQuery);
        assertEquals(0, responseTwo.getResponseCode());

        HBaseAdmin hAdmin = new HBaseAdmin(getHbaseConf());
        boolean doesTableExist = hAdmin.tableExists(tableName);
        assertTrue(doesTableExist);

        populateHBaseTable(tableName, 5);
        populateHBaseTableQualifier1(tableName, 6, false);
        populateHBaseTableQualifier1(tableName, 7, false);

        Configuration conf = new Configuration(hcatConf);
        conf.set(HCatConstants.HCAT_KEY_HIVE_CONF,
            HCatUtil.serialize(getHiveConf().getAllProperties()));

        Path outputDir = new Path(getTestDir(), "mapred/testHBaseTableIgnoreAbortedTransactions");
        FileSystem fs = getFileSystem();
        if (fs.exists(outputDir)) {
            fs.delete(outputDir, true);
        }
        Job job = new Job(conf, "hbase-aborted-transaction");
        job.setJarByClass(this.getClass());
        job.setMapperClass(MapReadHTable.class);
        MapReadHTable.resetCounters();
        job.setInputFormatClass(HCatInputFormat.class);
        InputJobInfo inputJobInfo = InputJobInfo.create(
            MetaStoreUtils.DEFAULT_DATABASE_NAME, tableName, null);
        HCatInputFormat.setInput(job, inputJobInfo);
        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job, outputDir);
        job.setMapOutputKeyClass(BytesWritable.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(BytesWritable.class);
        job.setOutputValueClass(Text.class);
        job.setNumReduceTasks(0);
        assertTrue(job.waitForCompletion(true));
        // Verify that the records do not contain aborted transaction
        // revisions 6 and 7 for testFamily:testQualifier1 and
        // fetches revision 5 for both testQualifier1 and testQualifier2
        assertFalse(MapReadHTable.error);
        assertEquals(1, MapReadHTable.count);

        String dropTableQuery = "DROP TABLE " + tableName;
        CommandProcessorResponse responseThree = hcatDriver.run(dropTableQuery);
        assertEquals(0, responseThree.getResponseCode());

        boolean isHbaseTableThere = hAdmin.tableExists(tableName);
        assertFalse(isHbaseTableThere);
    }

    @Test
    public void TestHBaseTableIgnoreAbortedAndRunningTransactions() throws Exception {
        String tableName = newTableName("mytable");
        String tableQuery = "CREATE TABLE " + tableName
            + "(key string, testqualifier1 string, testqualifier2 string) STORED BY " +
            "'org.apache.hcatalog.hbase.HBaseHCatStorageHandler'"
            + "TBLPROPERTIES ('hbase.columns.mapping'=':key," +
            "testFamily:testQualifier1,testFamily:testQualifier2')";

        CommandProcessorResponse responseTwo = hcatDriver.run(tableQuery);
        assertEquals(0, responseTwo.getResponseCode());

        HBaseAdmin hAdmin = new HBaseAdmin(getHbaseConf());
        boolean doesTableExist = hAdmin.tableExists(tableName);
        assertTrue(doesTableExist);

        populateHBaseTable(tableName, 2);
        populateHBaseTableQualifier1(tableName, 3, Boolean.TRUE); //Committed transaction
        populateHBaseTableQualifier1(tableName, 4, null); //Running transaction
        populateHBaseTableQualifier1(tableName, 5, Boolean.FALSE);  //Aborted transaction
        populateHBaseTableQualifier1(tableName, 6, Boolean.TRUE); //Committed transaction
        populateHBaseTableQualifier1(tableName, 7, null); //Running Transaction
        populateHBaseTableQualifier1(tableName, 8, Boolean.FALSE); //Aborted Transaction

        Configuration conf = new Configuration(hcatConf);
        conf.set(HCatConstants.HCAT_KEY_HIVE_CONF,
            HCatUtil.serialize(getHiveConf().getAllProperties()));

        Path outputDir = new Path(getTestDir(), "mapred/testHBaseTableIgnoreAbortedTransactions");
        FileSystem fs = getFileSystem();
        if (fs.exists(outputDir)) {
            fs.delete(outputDir, true);
        }
        Job job = new Job(conf, "hbase-running-aborted-transaction");
        job.setJarByClass(this.getClass());
        job.setMapperClass(MapReadHTableRunningAbort.class);
        job.setInputFormatClass(HCatInputFormat.class);
        InputJobInfo inputJobInfo = InputJobInfo.create(
            MetaStoreUtils.DEFAULT_DATABASE_NAME, tableName, null);
        HCatInputFormat.setInput(job, inputJobInfo);
        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job, outputDir);
        job.setMapOutputKeyClass(BytesWritable.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(BytesWritable.class);
        job.setOutputValueClass(Text.class);
        job.setNumReduceTasks(0);
        assertTrue(job.waitForCompletion(true));
        // Verify that the records do not contain running and aborted transaction
        // and it fetches revision 2 for testQualifier1 and testQualifier2
        assertFalse(MapReadHTableRunningAbort.error);
        assertEquals(1, MapReadHTableRunningAbort.count);

        String dropTableQuery = "DROP TABLE " + tableName;
        CommandProcessorResponse responseThree = hcatDriver.run(dropTableQuery);
        assertEquals(0, responseThree.getResponseCode());

        boolean isHbaseTableThere = hAdmin.tableExists(tableName);
        assertFalse(isHbaseTableThere);
    }


    static class MapReadHTable
        extends
        Mapper<ImmutableBytesWritable, HCatRecord, WritableComparable<?>, Text> {

        static boolean error = false;
        static int count = 0;

        @Override
        public void map(ImmutableBytesWritable key, HCatRecord value,
                        Context context) throws IOException, InterruptedException {
            System.out.println("HCat record value" + value.toString());
            boolean correctValues = (value.size() == 3)
                && (value.get(0).toString()).equalsIgnoreCase("testRow")
                && (value.get(1).toString()).equalsIgnoreCase("textValue-5")
                && (value.get(2).toString()).equalsIgnoreCase("textValue-5");

            if (correctValues == false) {
                error = true;
            }
            count++;
        }

        public static void resetCounters() {
            error = false;
            count = 0;
        }
    }

    static class MapReadProjHTable
        extends
        Mapper<ImmutableBytesWritable, HCatRecord, WritableComparable<?>, Text> {

        static boolean error = false;
        static int count = 0;

        @Override
        public void map(ImmutableBytesWritable key, HCatRecord value,
                        Context context) throws IOException, InterruptedException {
            System.out.println("HCat record value" + value.toString());
            boolean correctValues = (value.size() == 2)
                && (value.get(0).toString()).equalsIgnoreCase("testRow")
                && (value.get(1).toString()).equalsIgnoreCase("textValue-5");

            if (correctValues == false) {
                error = true;
            }
            count++;
        }
    }

    static class MapReadProjectionHTable
        implements org.apache.hadoop.mapred.Mapper<ImmutableBytesWritable, Result, WritableComparable<?>, Text> {

        static boolean error = false;
        static int count = 0;

        @Override
        public void configure(JobConf job) {
        }

        @Override
        public void close() throws IOException {
        }

        @Override
        public void map(ImmutableBytesWritable key, Result result,
                        OutputCollector<WritableComparable<?>, Text> output, Reporter reporter)
            throws IOException {
            System.out.println("Result " + result.toString());
            List<KeyValue> list = result.list();
            boolean correctValues = (list.size() == 1)
                && (Bytes.toString(list.get(0).getRow())).equalsIgnoreCase("testRow")
                && (Bytes.toString(list.get(0).getValue())).equalsIgnoreCase("textValue-5")
                && (Bytes.toString(list.get(0).getFamily())).equalsIgnoreCase("testFamily")
                && (Bytes.toString(list.get(0).getQualifier())).equalsIgnoreCase("testQualifier1");

            if (correctValues == false) {
                error = true;
            }
            count++;
        }
    }

    static class MapReadHTableRunningAbort
        extends
        Mapper<ImmutableBytesWritable, HCatRecord, WritableComparable<?>, Text> {

        static boolean error = false;
        static int count = 0;

        @Override
        public void map(ImmutableBytesWritable key, HCatRecord value,
                        Context context) throws IOException, InterruptedException {
            System.out.println("HCat record value" + value.toString());
            boolean correctValues = (value.size() == 3)
                && (value.get(0).toString()).equalsIgnoreCase("testRow")
                && (value.get(1).toString()).equalsIgnoreCase("textValue-3")
                && (value.get(2).toString()).equalsIgnoreCase("textValue-2");

            if (correctValues == false) {
                error = true;
            }
            count++;
        }
    }

    private HCatSchema getProjectionSchema() throws HCatException {

        HCatSchema schema = new HCatSchema(new ArrayList<HCatFieldSchema>());
        schema.append(new HCatFieldSchema("key", HCatFieldSchema.Type.STRING,
            ""));
        schema.append(new HCatFieldSchema("testqualifier1",
            HCatFieldSchema.Type.STRING, ""));
        return schema;
    }


}