From 998390990684f7086c4dce25b30868daa456c37e Mon Sep 17 00:00:00 2001 From: Sean Busbey Date: Wed, 11 Feb 2015 17:02:57 -0600 Subject: [PATCH] HBASE-13045 Tests to validate our example use of TableInputFormatBase * move mapreduce version of TableInputFormat tests out of mapred * add ability to get runnable job via MR test shims * add tests the run a job based on the extending TableInputFormatBase (as given in the javadocs) * correct the javadoc example for current APIs. --- .../hadoop/hbase/mapred/TableInputFormatBase.java | 32 +- .../hbase/mapreduce/TableInputFormatBase.java | 39 +- .../hadoop/hbase/mapred/TestTableInputFormat.java | 206 +++++----- .../hbase/mapreduce/MapreduceTestingShim.java | 27 +- .../hbase/mapreduce/TestTableInputFormat.java | 408 ++++++++++++++++++++ 5 files changed, 580 insertions(+), 132 deletions(-) create mode 100644 hbase-server/src/test/java/org/apache/hadoop/hbase/mapreduce/TestTableInputFormat.java diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/mapred/TableInputFormatBase.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/mapred/TableInputFormatBase.java index 85c920b06d5d61292713cdc147734d701c05b6cc..3b7453115f09e4456d4b6f2475d98b284d26249c 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/mapred/TableInputFormatBase.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/mapred/TableInputFormatBase.java @@ -45,23 +45,25 @@ import org.apache.hadoop.mapred.Reporter; *
  *   class ExampleTIF extends TableInputFormatBase implements JobConfigurable {
  *
+ *     @Override
  *     public void configure(JobConf job) {
- *       HTable exampleTable = new HTable(HBaseConfiguration.create(job),
- *         Bytes.toBytes("exampleTable"));
- *       // mandatory
- *       setHTable(exampleTable);
- *       Text[] inputColumns = new byte [][] { Bytes.toBytes("columnA"),
- *         Bytes.toBytes("columnB") };
- *       // mandatory
- *       setInputColumns(inputColumns);
- *       RowFilterInterface exampleFilter = new RegExpRowFilter("keyPrefix.*");
- *       // optional
- *       setRowFilter(exampleFilter);
+ *       try {
+ *         HTable exampleTable = new HTable(HBaseConfiguration.create(job),
+ *           Bytes.toBytes("exampleTable"));
+ *         // mandatory
+ *         setHTable(exampleTable);
+ *         byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"),
+ *           Bytes.toBytes("columnB") };
+ *         // mandatory
+ *         setInputColumns(inputColumns);
+ *         // optional
+ *         Filter exampleFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator("aa.*"));
+ *         setRowFilter(exampleFilter);
+ *       } catch (IOException exception) {
+ *         throw new RuntimeException("Failed to configure for job.", exception);
+ *       }
  *     }
- *
- *     public void validateInput(JobConf job) throws IOException {
- *     }
- *  }
+ *   }
  * 
*/ diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/TableInputFormatBase.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/TableInputFormatBase.java index 268d4de33fafb583c821928b2823e0ab2863aca0..acd28b6103ba94a69e39ba7261c3795ffdd363a5 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/TableInputFormatBase.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/TableInputFormatBase.java @@ -58,25 +58,30 @@ import org.apache.hadoop.util.StringUtils; *

* An example of a subclass: *

- *   class ExampleTIF extends TableInputFormatBase implements JobConfigurable {
- *
+ *   public static class ExampleTIF extends TableInputFormatBase implements JobConfigurable {
+ *  
+ *     @Override
  *     public void configure(JobConf job) {
- *       HTable exampleTable = new HTable(HBaseConfiguration.create(job),
- *         Bytes.toBytes("exampleTable"));
- *       // mandatory
- *       setHTable(exampleTable);
- *       Text[] inputColumns = new byte [][] { Bytes.toBytes("cf1:columnA"),
- *         Bytes.toBytes("cf2") };
- *       // mandatory
- *       setInputColumns(inputColumns);
- *       RowFilterInterface exampleFilter = new RegExpRowFilter("keyPrefix.*");
- *       // optional
- *       setRowFilter(exampleFilter);
- *     }
- *
- *     public void validateInput(JobConf job) throws IOException {
+ *       try {
+ *         HTable exampleTable = new HTable(HBaseConfiguration.create(job),
+ *           Bytes.toBytes("exampleTable"));
+ *         // mandatory
+ *         setHTable(exampleTable);
+ *         byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"),
+ *           Bytes.toBytes("columnB") };
+ *         // optional
+ *         Scan scan = new Scan();
+ *         for (byte[] family : inputColumns) {
+ *           scan.addFamily(family);
+ *         }
+ *         Filter exampleFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator("aa.*"));
+ *         scan.setFilter(exampleFilter);
+ *         setScan(scan);
+ *       } catch (IOException exception) {
+ *         throw new RuntimeException("Failed to configure for job.", exception);
+ *       }
  *     }
- *  }
+ *   }
  * 
*/ @InterfaceAudience.Public diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mapred/TestTableInputFormat.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mapred/TestTableInputFormat.java index 93cb29a4764ccd5b87a8d800f29a812cd4b1f6c0..ddefce6ece122c2b779794be19a85ddaf70b2ec1 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/mapred/TestTableInputFormat.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/mapred/TestTableInputFormat.java @@ -35,14 +35,29 @@ import java.util.Map; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.hbase.*; +import org.apache.hadoop.hbase.Cell; 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.client.ResultScanner; import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp; +import org.apache.hadoop.hbase.filter.Filter; +import org.apache.hadoop.hbase.filter.RegexStringComparator; +import org.apache.hadoop.hbase.filter.RowFilter; +import org.apache.hadoop.hbase.mapreduce.MapreduceTestingShim; import org.apache.hadoop.hbase.io.ImmutableBytesWritable; import org.apache.hadoop.hbase.testclassification.LargeTests; import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.io.NullWritable; +import org.apache.hadoop.mapred.JobClient; +import org.apache.hadoop.mapred.JobConf; +import org.apache.hadoop.mapred.JobConfigurable; +import org.apache.hadoop.mapred.MiniMRCluster; +import org.apache.hadoop.mapred.OutputCollector; +import org.apache.hadoop.mapred.Reporter; +import org.apache.hadoop.mapred.RunningJob; +import org.apache.hadoop.mapred.lib.NullOutputFormat; import org.junit.AfterClass; import org.junit.Before; import org.junit.BeforeClass; @@ -61,6 +76,7 @@ public class TestTableInputFormat { private static final Log LOG = LogFactory.getLog(TestTableInputFormat.class); private final static HBaseTestingUtility UTIL = new HBaseTestingUtility(); + private static MiniMRCluster mrCluster; static final byte[] FAMILY = Bytes.toBytes("family"); private static final byte[][] columns = new byte[][] { FAMILY }; @@ -68,10 +84,12 @@ public class TestTableInputFormat { @BeforeClass public static void beforeClass() throws Exception { UTIL.startMiniCluster(); + mrCluster = UTIL.startMiniMapReduceCluster(); } @AfterClass public static void afterClass() throws Exception { + UTIL.shutdownMiniMapReduceCluster(); UTIL.shutdownMiniCluster(); } @@ -90,12 +108,27 @@ public class TestTableInputFormat { * @throws IOException */ public static HTable createTable(byte[] tableName) throws IOException { - HTable table = UTIL.createTable(tableName, FAMILY); + return createTable(tableName, new byte[][] { FAMILY }); + } + + /** + * Setup a table with two rows and values per column family. + * + * @param tableName + * @return + * @throws IOException + */ + public static HTable createTable(byte[] tableName, byte[][] families) throws IOException { + HTable table = UTIL.createTable(tableName, families); Put p = new Put("aaa".getBytes()); - p.add(FAMILY, null, "value aaa".getBytes()); + for (byte[] family : families) { + p.add(family, null, "value aaa".getBytes()); + } table.put(p); p = new Put("bbb".getBytes()); - p.add(FAMILY, null, "value bbb".getBytes()); + for (byte[] family : families) { + p.add(family, null, "value bbb".getBytes()); + } table.put(p); return table; } @@ -151,46 +184,6 @@ public class TestTableInputFormat { } /** - * Create table data and run tests on specified htable using the - * o.a.h.hbase.mapreduce API. - * - * @param table - * @throws IOException - * @throws InterruptedException - */ - static void runTestMapreduce(HTable table) throws IOException, - InterruptedException { - org.apache.hadoop.hbase.mapreduce.TableRecordReaderImpl trr = - new org.apache.hadoop.hbase.mapreduce.TableRecordReaderImpl(); - Scan s = new Scan(); - s.setStartRow("aaa".getBytes()); - s.setStopRow("zzz".getBytes()); - s.addFamily(FAMILY); - trr.setScan(s); - trr.setHTable(table); - - trr.initialize(null, null); - Result r = new Result(); - ImmutableBytesWritable key = new ImmutableBytesWritable(); - - boolean more = trr.nextKeyValue(); - assertTrue(more); - key = trr.getCurrentKey(); - r = trr.getCurrentValue(); - checkResult(r, key, "aaa".getBytes(), "value aaa".getBytes()); - - more = trr.nextKeyValue(); - assertTrue(more); - key = trr.getCurrentKey(); - r = trr.getCurrentValue(); - checkResult(r, key, "bbb".getBytes(), "value bbb".getBytes()); - - // no more data - more = trr.nextKeyValue(); - assertFalse(more); - } - - /** * Create a table that IOE's on first scanner next call * * @throws IOException @@ -320,70 +313,85 @@ public class TestTableInputFormat { } /** - * Run test assuming no errors using newer mapreduce api - * - * @throws IOException - * @throws InterruptedException + * Verify the example we present in javadocs on TableInputFormatBase */ @Test - public void testTableRecordReaderMapreduce() throws IOException, - InterruptedException { - HTable table = createTable("table1-mr".getBytes()); - runTestMapreduce(table); + public void testExtensionOfTableInputFormatBase() throws IOException { + LOG.info("testing use of an InputFormat taht extends InputFormatBase"); + final HTable htable = createTable(Bytes.toBytes("exampleTable"), + new byte[][] { Bytes.toBytes("columnA"), Bytes.toBytes("columnB") }); + final JobConf job = MapreduceTestingShim.getJobConf(mrCluster); + job.setInputFormat(ExampleTIF.class); + job.setOutputFormat(NullOutputFormat.class); + job.setMapperClass(ExampleVerifier.class); + job.setNumReduceTasks(0); + LOG.debug("submitting job."); + final RunningJob run = JobClient.runJob(job); + assertTrue("job failed!", run.isSuccessful()); + assertEquals("Saw the wrong number of instances of the filtered-for row.", 2, run.getCounters() + .findCounter(TestTableInputFormat.class.getName() + ":row", "aaa").getCounter()); + assertEquals("Saw any instances of the filtered out row.", 0, run.getCounters() + .findCounter(TestTableInputFormat.class.getName() + ":row", "bbb").getCounter()); + assertEquals("Saw the wrong number of instances of columnA.", 1, run.getCounters() + .findCounter(TestTableInputFormat.class.getName() + ":family", "columnA").getCounter()); + assertEquals("Saw the wrong number of instances of columnB.", 1, run.getCounters() + .findCounter(TestTableInputFormat.class.getName() + ":family", "columnB").getCounter()); + assertEquals("Saw the wrong count of values for the filtered-for row.", 2, run.getCounters() + .findCounter(TestTableInputFormat.class.getName() + ":value", "value aaa").getCounter()); + assertEquals("Saw the wrong count of values for the filtered-out row.", 0, run.getCounters() + .findCounter(TestTableInputFormat.class.getName() + ":value", "value bbb").getCounter()); } - /** - * Run test assuming Scanner IOException failure using newer mapreduce api - * - * @throws IOException - * @throws InterruptedException - */ - @Test - public void testTableRecordReaderScannerFailMapreduce() throws IOException, - InterruptedException { - HTable htable = createIOEScannerTable("table2-mr".getBytes(), 1); - runTestMapreduce(htable); - } + public static class ExampleVerifier implements TableMap { + + @Override + public void configure(JobConf conf) { + } + + @Override + public void map(ImmutableBytesWritable key, Result value, + OutputCollector output, + Reporter reporter) throws IOException { + for (Cell cell : value.listCells()) { + reporter.getCounter(TestTableInputFormat.class.getName() + ":row", + Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength())) + .increment(1l); + reporter.getCounter(TestTableInputFormat.class.getName() + ":family", + Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength())) + .increment(1l); + reporter.getCounter(TestTableInputFormat.class.getName() + ":value", + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength())) + .increment(1l); + } + } - /** - * Run test assuming Scanner IOException failure using newer mapreduce api - * - * @throws IOException - * @throws InterruptedException - */ - @Test(expected = IOException.class) - public void testTableRecordReaderScannerFailMapreduceTwice() throws IOException, - InterruptedException { - HTable htable = createIOEScannerTable("table3-mr".getBytes(), 2); - runTestMapreduce(htable); - } + @Override + public void close() { + } - /** - * Run test assuming UnknownScannerException (which is a type of - * DoNotRetryIOException) using newer mapreduce api - * - * @throws InterruptedException - * @throws org.apache.hadoop.hbase.DoNotRetryIOException - */ - @Test - public void testTableRecordReaderScannerTimeoutMapreduce() - throws IOException, InterruptedException { - HTable htable = createDNRIOEScannerTable("table4-mr".getBytes(), 1); - runTestMapreduce(htable); } - /** - * Run test assuming UnknownScannerException (which is a type of - * DoNotRetryIOException) using newer mapreduce api - * - * @throws InterruptedException - * @throws org.apache.hadoop.hbase.DoNotRetryIOException - */ - @Test(expected = org.apache.hadoop.hbase.DoNotRetryIOException.class) - public void testTableRecordReaderScannerTimeoutMapreduceTwice() - throws IOException, InterruptedException { - HTable htable = createDNRIOEScannerTable("table5-mr".getBytes(), 2); - runTestMapreduce(htable); + public static class ExampleTIF extends TableInputFormatBase implements JobConfigurable { + + @Override + public void configure(JobConf job) { + try { + HTable exampleTable = new HTable(HBaseConfiguration.create(job), + Bytes.toBytes("exampleTable")); + // mandatory + setHTable(exampleTable); + byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"), + Bytes.toBytes("columnB") }; + // mandatory + setInputColumns(inputColumns); + Filter exampleFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator("aa.*")); + // optional + setRowFilter(exampleFilter); + } catch (IOException exception) { + throw new RuntimeException("Failed to configure for job.", exception); + } + } + } } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mapreduce/MapreduceTestingShim.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mapreduce/MapreduceTestingShim.java index dee42771ade056be3714d316b0b120258b7c8896..b080d7f426418ae4b5827650c7f5fba0af921bb7 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/mapreduce/MapreduceTestingShim.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/mapreduce/MapreduceTestingShim.java @@ -52,6 +52,8 @@ abstract public class MapreduceTestingShim { abstract public JobContext newJobContext(Configuration jobConf) throws IOException; + + abstract public Job newJob(Configuration conf) throws IOException; abstract public JobConf obtainJobConf(MiniMRCluster cluster); @@ -66,6 +68,10 @@ abstract public class MapreduceTestingShim { return instance.obtainJobConf(cluster); } + public static Job createJob(Configuration conf) throws IOException { + return instance.newJob(conf); + } + public static String getMROutputDirProp() { return instance.obtainMROutputDirProp(); } @@ -84,6 +90,20 @@ abstract public class MapreduceTestingShim { "Failed to instantiate new JobContext(jobConf, new JobID())", e); } } + + @Override + public Job newJob(Configuration conf) throws IOException { + // Implementing: + // return new Job(conf); + Constructor c; + try { + c = Job.class.getConstructor(Configuration.class); + return c.newInstance(conf); + } catch (Exception e) { + throw new IllegalStateException( + "Failed to instantiate new Job(conf)", e); + } + } public JobConf obtainJobConf(MiniMRCluster cluster) { if (cluster == null) return null; @@ -110,11 +130,16 @@ abstract public class MapreduceTestingShim { private static class MapreduceV2Shim extends MapreduceTestingShim { public JobContext newJobContext(Configuration jobConf) { + return newJob(jobConf); + } + + @Override + public Job newJob(Configuration jobConf) { // Implementing: // return Job.getInstance(jobConf); try { Method m = Job.class.getMethod("getInstance", Configuration.class); - return (JobContext) m.invoke(null, jobConf); // static method, then arg + return (Job) m.invoke(null, jobConf); // static method, then arg } catch (Exception e) { e.printStackTrace(); throw new IllegalStateException( diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mapreduce/TestTableInputFormat.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mapreduce/TestTableInputFormat.java new file mode 100644 index 0000000000000000000000000000000000000000..0356100218a87ffd206df8cd17306d43267c7709 --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/mapreduce/TestTableInputFormat.java @@ -0,0 +1,408 @@ +/** + * + * 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.hbase.mapreduce; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.Matchers.anyObject; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Map; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.hadoop.hbase.*; +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.client.ResultScanner; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp; +import org.apache.hadoop.hbase.filter.Filter; +import org.apache.hadoop.hbase.filter.RegexStringComparator; +import org.apache.hadoop.hbase.filter.RowFilter; +import org.apache.hadoop.hbase.io.ImmutableBytesWritable; +import org.apache.hadoop.hbase.testclassification.LargeTests; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.io.NullWritable; +import org.apache.hadoop.mapred.JobConf; +import org.apache.hadoop.mapred.JobConfigurable; +import org.apache.hadoop.mapred.MiniMRCluster; +import org.apache.hadoop.mapreduce.Job; +import org.apache.hadoop.mapreduce.Mapper.Context; +import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; + +/** + * This tests the TableInputFormat and its recovery semantics + * + */ +@Category(LargeTests.class) +public class TestTableInputFormat { + + private static final Log LOG = LogFactory.getLog(TestTableInputFormat.class); + + private final static HBaseTestingUtility UTIL = new HBaseTestingUtility(); + private static MiniMRCluster mrCluster; + static final byte[] FAMILY = Bytes.toBytes("family"); + + private static final byte[][] columns = new byte[][] { FAMILY }; + + @BeforeClass + public static void beforeClass() throws Exception { + UTIL.startMiniCluster(); + mrCluster = UTIL.startMiniMapReduceCluster(); + } + + @AfterClass + public static void afterClass() throws Exception { + UTIL.shutdownMiniMapReduceCluster(); + UTIL.shutdownMiniCluster(); + } + + @Before + public void before() throws IOException { + LOG.info("before"); + UTIL.ensureSomeRegionServersAvailable(1); + LOG.info("before done"); + } + + /** + * Setup a table with two rows and values. + * + * @param tableName + * @return + * @throws IOException + */ + public static HTable createTable(byte[] tableName) throws IOException { + return createTable(tableName, new byte[][] { FAMILY }); + } + + /** + * Setup a table with two rows and values per column family. + * + * @param tableName + * @return + * @throws IOException + */ + public static HTable createTable(byte[] tableName, byte[][] families) throws IOException { + HTable table = UTIL.createTable(tableName, families); + Put p = new Put("aaa".getBytes()); + for (byte[] family : families) { + p.add(family, null, "value aaa".getBytes()); + } + table.put(p); + p = new Put("bbb".getBytes()); + for (byte[] family : families) { + p.add(family, null, "value bbb".getBytes()); + } + table.put(p); + return table; + } + + /** + * Verify that the result and key have expected values. + * + * @param r + * @param key + * @param expectedKey + * @param expectedValue + * @return + */ + static boolean checkResult(Result r, ImmutableBytesWritable key, + byte[] expectedKey, byte[] expectedValue) { + assertEquals(0, key.compareTo(expectedKey)); + Map vals = r.getFamilyMap(FAMILY); + byte[] value = vals.values().iterator().next(); + assertTrue(Arrays.equals(value, expectedValue)); + return true; // if succeed + } + + /** + * Create table data and run tests on specified htable using the + * o.a.h.hbase.mapreduce API. + * + * @param table + * @throws IOException + * @throws InterruptedException + */ + static void runTestMapreduce(HTable table) throws IOException, + InterruptedException { + org.apache.hadoop.hbase.mapreduce.TableRecordReaderImpl trr = + new org.apache.hadoop.hbase.mapreduce.TableRecordReaderImpl(); + Scan s = new Scan(); + s.setStartRow("aaa".getBytes()); + s.setStopRow("zzz".getBytes()); + s.addFamily(FAMILY); + trr.setScan(s); + trr.setHTable(table); + + trr.initialize(null, null); + Result r = new Result(); + ImmutableBytesWritable key = new ImmutableBytesWritable(); + + boolean more = trr.nextKeyValue(); + assertTrue(more); + key = trr.getCurrentKey(); + r = trr.getCurrentValue(); + checkResult(r, key, "aaa".getBytes(), "value aaa".getBytes()); + + more = trr.nextKeyValue(); + assertTrue(more); + key = trr.getCurrentKey(); + r = trr.getCurrentValue(); + checkResult(r, key, "bbb".getBytes(), "value bbb".getBytes()); + + // no more data + more = trr.nextKeyValue(); + assertFalse(more); + } + + /** + * Create a table that IOE's on first scanner next call + * + * @throws IOException + */ + static HTable createIOEScannerTable(byte[] name, final int failCnt) + throws IOException { + // build up a mock scanner stuff to fail the first time + Answer a = new Answer() { + int cnt = 0; + + @Override + public ResultScanner answer(InvocationOnMock invocation) throws Throwable { + // first invocation return the busted mock scanner + if (cnt++ < failCnt) { + // create mock ResultScanner that always fails. + Scan scan = mock(Scan.class); + doReturn("bogus".getBytes()).when(scan).getStartRow(); // avoid npe + ResultScanner scanner = mock(ResultScanner.class); + // simulate TimeoutException / IOException + doThrow(new IOException("Injected exception")).when(scanner).next(); + return scanner; + } + + // otherwise return the real scanner. + return (ResultScanner) invocation.callRealMethod(); + } + }; + + HTable htable = spy(createTable(name)); + doAnswer(a).when(htable).getScanner((Scan) anyObject()); + return htable; + } + + /** + * Create a table that throws a DoNoRetryIOException on first scanner next + * call + * + * @throws IOException + */ + static HTable createDNRIOEScannerTable(byte[] name, final int failCnt) + throws IOException { + // build up a mock scanner stuff to fail the first time + Answer a = new Answer() { + int cnt = 0; + + @Override + public ResultScanner answer(InvocationOnMock invocation) throws Throwable { + // first invocation return the busted mock scanner + if (cnt++ < failCnt) { + // create mock ResultScanner that always fails. + Scan scan = mock(Scan.class); + doReturn("bogus".getBytes()).when(scan).getStartRow(); // avoid npe + ResultScanner scanner = mock(ResultScanner.class); + + invocation.callRealMethod(); // simulate UnknownScannerException + doThrow( + new UnknownScannerException("Injected simulated TimeoutException")) + .when(scanner).next(); + return scanner; + } + + // otherwise return the real scanner. + return (ResultScanner) invocation.callRealMethod(); + } + }; + + HTable htable = spy(createTable(name)); + doAnswer(a).when(htable).getScanner((Scan) anyObject()); + return htable; + } + + /** + * Run test assuming no errors using newer mapreduce api + * + * @throws IOException + * @throws InterruptedException + */ + @Test + public void testTableRecordReaderMapreduce() throws IOException, + InterruptedException { + HTable table = createTable("table1-mr".getBytes()); + runTestMapreduce(table); + } + + /** + * Run test assuming Scanner IOException failure using newer mapreduce api + * + * @throws IOException + * @throws InterruptedException + */ + @Test + public void testTableRecordReaderScannerFailMapreduce() throws IOException, + InterruptedException { + HTable htable = createIOEScannerTable("table2-mr".getBytes(), 1); + runTestMapreduce(htable); + } + + /** + * Run test assuming Scanner IOException failure using newer mapreduce api + * + * @throws IOException + * @throws InterruptedException + */ + @Test(expected = IOException.class) + public void testTableRecordReaderScannerFailMapreduceTwice() throws IOException, + InterruptedException { + HTable htable = createIOEScannerTable("table3-mr".getBytes(), 2); + runTestMapreduce(htable); + } + + /** + * Run test assuming UnknownScannerException (which is a type of + * DoNotRetryIOException) using newer mapreduce api + * + * @throws InterruptedException + * @throws org.apache.hadoop.hbase.DoNotRetryIOException + */ + @Test + public void testTableRecordReaderScannerTimeoutMapreduce() + throws IOException, InterruptedException { + HTable htable = createDNRIOEScannerTable("table4-mr".getBytes(), 1); + runTestMapreduce(htable); + } + + /** + * Run test assuming UnknownScannerException (which is a type of + * DoNotRetryIOException) using newer mapreduce api + * + * @throws InterruptedException + * @throws org.apache.hadoop.hbase.DoNotRetryIOException + */ + @Test(expected = org.apache.hadoop.hbase.DoNotRetryIOException.class) + public void testTableRecordReaderScannerTimeoutMapreduceTwice() + throws IOException, InterruptedException { + HTable htable = createDNRIOEScannerTable("table5-mr".getBytes(), 2); + runTestMapreduce(htable); + } + + /** + * Verify the example we present in javadocs on TableInputFormatBase + */ + @Test + public void testExtensionOfTableInputFormatBase() + throws IOException, InterruptedException, ClassNotFoundException { + LOG.info("testing use of an InputFormat taht extends InputFormatBase"); + final HTable htable = createTable(Bytes.toBytes("exampleTable"), + new byte[][] { Bytes.toBytes("columnA"), Bytes.toBytes("columnB") }); + + final Job job = MapreduceTestingShim.createJob(UTIL.getConfiguration()); + job.setInputFormatClass(ExampleTIF.class); + job.setOutputFormatClass(NullOutputFormat.class); + job.setMapperClass(ExampleVerifier.class); + job.setNumReduceTasks(0); + + LOG.debug("submitting job."); + assertTrue("job failed!", job.waitForCompletion(true)); + assertEquals("Saw the wrong number of instances of the filtered-for row.", 2, job.getCounters() + .findCounter(TestTableInputFormat.class.getName() + ":row", "aaa").getValue()); + assertEquals("Saw any instances of the filtered out row.", 0, job.getCounters() + .findCounter(TestTableInputFormat.class.getName() + ":row", "bbb").getValue()); + assertEquals("Saw the wrong number of instances of columnA.", 1, job.getCounters() + .findCounter(TestTableInputFormat.class.getName() + ":family", "columnA").getValue()); + assertEquals("Saw the wrong number of instances of columnB.", 1, job.getCounters() + .findCounter(TestTableInputFormat.class.getName() + ":family", "columnB").getValue()); + assertEquals("Saw the wrong count of values for the filtered-for row.", 2, job.getCounters() + .findCounter(TestTableInputFormat.class.getName() + ":value", "value aaa").getValue()); + assertEquals("Saw the wrong count of values for the filtered-out row.", 0, job.getCounters() + .findCounter(TestTableInputFormat.class.getName() + ":value", "value bbb").getValue()); + } + + public static class ExampleVerifier extends TableMapper { + + @Override + public void map(ImmutableBytesWritable key, Result value, Context context) + throws IOException { + for (Cell cell : value.listCells()) { + context.getCounter(TestTableInputFormat.class.getName() + ":row", + Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength())) + .increment(1l); + context.getCounter(TestTableInputFormat.class.getName() + ":family", + Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength())) + .increment(1l); + context.getCounter(TestTableInputFormat.class.getName() + ":value", + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength())) + .increment(1l); + } + } + + } + + public static class ExampleTIF extends TableInputFormatBase implements JobConfigurable { + + @Override + public void configure(JobConf job) { + try { + HTable exampleTable = new HTable(HBaseConfiguration.create(job), + Bytes.toBytes("exampleTable")); + // mandatory + setHTable(exampleTable); + byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"), + Bytes.toBytes("columnB") }; + // optional + Scan scan = new Scan(); + for (byte[] family : inputColumns) { + scan.addFamily(family); + } + Filter exampleFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator("aa.*")); + scan.setFilter(exampleFilter); + setScan(scan); + } catch (IOException exception) { + throw new RuntimeException("Failed to configure for job.", exception); + } + } + + } + +} + -- 1.7.10.2 (Apple Git-33)