001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.mapred;
019
020import static org.apache.hadoop.hbase.mapreduce.TableSnapshotInputFormatImpl.SNAPSHOT_INPUTFORMAT_LOCALITY_ENABLED_DEFAULT;
021import static org.mockito.Mockito.mock;
022
023import java.io.IOException;
024import java.util.Iterator;
025import org.apache.hadoop.fs.Path;
026import org.apache.hadoop.hbase.HBaseClassTestRule;
027import org.apache.hadoop.hbase.HBaseTestingUtility;
028import org.apache.hadoop.hbase.HConstants;
029import org.apache.hadoop.hbase.TableName;
030import org.apache.hadoop.hbase.client.Result;
031import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
032import org.apache.hadoop.hbase.mapreduce.TableSnapshotInputFormatTestBase;
033import org.apache.hadoop.hbase.testclassification.LargeTests;
034import org.apache.hadoop.hbase.testclassification.VerySlowMapReduceTests;
035import org.apache.hadoop.hbase.util.Bytes;
036import org.apache.hadoop.hbase.util.RegionSplitter;
037import org.apache.hadoop.io.NullWritable;
038import org.apache.hadoop.mapred.InputSplit;
039import org.apache.hadoop.mapred.JobClient;
040import org.apache.hadoop.mapred.JobConf;
041import org.apache.hadoop.mapred.MapReduceBase;
042import org.apache.hadoop.mapred.OutputCollector;
043import org.apache.hadoop.mapred.RecordReader;
044import org.apache.hadoop.mapred.Reducer;
045import org.apache.hadoop.mapred.Reporter;
046import org.apache.hadoop.mapred.RunningJob;
047import org.apache.hadoop.mapred.lib.NullOutputFormat;
048import org.junit.Assert;
049import org.junit.ClassRule;
050import org.junit.Rule;
051import org.junit.Test;
052import org.junit.experimental.categories.Category;
053import org.junit.rules.TestName;
054
055@Category({VerySlowMapReduceTests.class, LargeTests.class})
056public class TestTableSnapshotInputFormat extends TableSnapshotInputFormatTestBase {
057
058  @ClassRule
059  public static final HBaseClassTestRule CLASS_RULE =
060      HBaseClassTestRule.forClass(TestTableSnapshotInputFormat.class);
061
062  private static final byte[] aaa = Bytes.toBytes("aaa");
063  private static final byte[] after_zzz = Bytes.toBytes("zz{"); // 'z' + 1 => '{'
064  private static final String COLUMNS =
065    Bytes.toString(FAMILIES[0]) + " " + Bytes.toString(FAMILIES[1]);
066
067  @Rule
068  public TestName name = new TestName();
069
070  @Override
071  protected byte[] getStartRow() {
072    return aaa;
073  }
074
075  @Override
076  protected byte[] getEndRow() {
077    return after_zzz;
078  }
079
080  static class TestTableSnapshotMapper extends MapReduceBase
081      implements TableMap<ImmutableBytesWritable, NullWritable> {
082    @Override
083    public void map(ImmutableBytesWritable key, Result value,
084        OutputCollector<ImmutableBytesWritable, NullWritable> collector, Reporter reporter)
085        throws IOException {
086      verifyRowFromMap(key, value);
087      collector.collect(key, NullWritable.get());
088    }
089  }
090
091  public static class TestTableSnapshotReducer extends MapReduceBase
092      implements Reducer<ImmutableBytesWritable, NullWritable, NullWritable, NullWritable> {
093    HBaseTestingUtility.SeenRowTracker rowTracker =
094      new HBaseTestingUtility.SeenRowTracker(aaa, after_zzz);
095
096    @Override
097    public void reduce(ImmutableBytesWritable key, Iterator<NullWritable> values,
098        OutputCollector<NullWritable, NullWritable> collector, Reporter reporter)
099        throws IOException {
100      rowTracker.addRow(key.get());
101    }
102
103    @Override
104    public void close() {
105      rowTracker.validate();
106    }
107  }
108
109  @Test
110  public void testInitTableSnapshotMapperJobConfig() throws Exception {
111    setupCluster();
112    final TableName tableName = TableName.valueOf(name.getMethodName());
113    String snapshotName = "foo";
114
115    try {
116      createTableAndSnapshot(UTIL, tableName, snapshotName, getStartRow(), getEndRow(), 1);
117      JobConf job = new JobConf(UTIL.getConfiguration());
118      Path tmpTableDir = UTIL.getDataTestDirOnTestFS(snapshotName);
119
120      TableMapReduceUtil.initTableSnapshotMapJob(snapshotName,
121        COLUMNS, TestTableSnapshotMapper.class, ImmutableBytesWritable.class,
122        NullWritable.class, job, false, tmpTableDir);
123
124      // TODO: would be better to examine directly the cache instance that results from this
125      // config. Currently this is not possible because BlockCache initialization is static.
126      Assert.assertEquals(
127        "Snapshot job should be configured for default LruBlockCache.",
128        HConstants.HFILE_BLOCK_CACHE_SIZE_DEFAULT,
129        job.getFloat(HConstants.HFILE_BLOCK_CACHE_SIZE_KEY, -1), 0.01);
130      Assert.assertEquals(
131        "Snapshot job should not use BucketCache.",
132        0, job.getFloat("hbase.bucketcache.size", -1), 0.01);
133    } finally {
134      UTIL.getAdmin().deleteSnapshot(snapshotName);
135      UTIL.deleteTable(tableName);
136      tearDownCluster();
137    }
138  }
139
140  // TODO: mapred does not support limiting input range by startrow, endrow.
141  // Thus the following tests must override parameterverification.
142
143  @Test
144  @Override
145  public void testWithMockedMapReduceMultiRegion() throws Exception {
146    testWithMockedMapReduce(
147        UTIL, "testWithMockedMapReduceMultiRegion", 10, 1, 10, true);
148        // It does not matter whether true or false is given to setLocalityEnabledTo,
149        // because it is not read in testWithMockedMapReduce().
150  }
151
152  @Test
153  @Override
154  public void testWithMapReduceMultiRegion() throws Exception {
155    testWithMapReduce(UTIL, "testWithMapReduceMultiRegion", 10, 1, 10, false);
156  }
157
158  @Test
159  @Override
160  // run the MR job while HBase is offline
161  public void testWithMapReduceAndOfflineHBaseMultiRegion() throws Exception {
162    testWithMapReduce(UTIL, "testWithMapReduceAndOfflineHBaseMultiRegion", 10, 1, 10, true);
163  }
164
165  @Override
166  public void testRestoreSnapshotDoesNotCreateBackRefLinksInit(TableName tableName,
167      String snapshotName, Path tmpTableDir) throws Exception {
168    JobConf job = new JobConf(UTIL.getConfiguration());
169    TableMapReduceUtil.initTableSnapshotMapJob(snapshotName,
170      COLUMNS, TestTableSnapshotMapper.class, ImmutableBytesWritable.class,
171      NullWritable.class, job, false, tmpTableDir);
172  }
173
174  @Override
175  protected void testWithMockedMapReduce(HBaseTestingUtility util, String snapshotName,
176      int numRegions, int numSplitsPerRegion, int expectedNumSplits, boolean setLocalityEnabledTo)
177      throws Exception {
178    setupCluster();
179    final TableName tableName = TableName.valueOf(name.getMethodName());
180    try {
181      createTableAndSnapshot(
182        util, tableName, snapshotName, getStartRow(), getEndRow(), numRegions);
183
184      JobConf job = new JobConf(util.getConfiguration());
185      // setLocalityEnabledTo is ignored no matter what is specified, so as to test the case that
186      // SNAPSHOT_INPUTFORMAT_LOCALITY_ENABLED_KEY is not explicitly specified
187      // and the default value is taken.
188      Path tmpTableDir = util.getDataTestDirOnTestFS(snapshotName);
189
190      if (numSplitsPerRegion > 1) {
191        TableMapReduceUtil.initTableSnapshotMapJob(snapshotName,
192                COLUMNS, TestTableSnapshotMapper.class, ImmutableBytesWritable.class,
193                NullWritable.class, job, false, tmpTableDir, new RegionSplitter.UniformSplit(),
194                numSplitsPerRegion);
195      } else {
196        TableMapReduceUtil.initTableSnapshotMapJob(snapshotName,
197                COLUMNS, TestTableSnapshotMapper.class, ImmutableBytesWritable.class,
198                NullWritable.class, job, false, tmpTableDir);
199      }
200
201      // mapred doesn't support start and end keys? o.O
202      verifyWithMockedMapReduce(job, numRegions, expectedNumSplits, getStartRow(), getEndRow());
203
204    } finally {
205      util.getAdmin().deleteSnapshot(snapshotName);
206      util.deleteTable(tableName);
207      tearDownCluster();
208    }
209  }
210
211  private void verifyWithMockedMapReduce(JobConf job, int numRegions, int expectedNumSplits,
212      byte[] startRow, byte[] stopRow) throws IOException, InterruptedException {
213    TableSnapshotInputFormat tsif = new TableSnapshotInputFormat();
214    InputSplit[] splits = tsif.getSplits(job, 0);
215
216    Assert.assertEquals(expectedNumSplits, splits.length);
217
218    HBaseTestingUtility.SeenRowTracker rowTracker =
219      new HBaseTestingUtility.SeenRowTracker(startRow, stopRow);
220
221    // SNAPSHOT_INPUTFORMAT_LOCALITY_ENABLED_KEY is not explicitly specified,
222    // so the default value is taken.
223    boolean localityEnabled = SNAPSHOT_INPUTFORMAT_LOCALITY_ENABLED_DEFAULT;
224
225    for (int i = 0; i < splits.length; i++) {
226      // validate input split
227      InputSplit split = splits[i];
228      Assert.assertTrue(split instanceof TableSnapshotInputFormat.TableSnapshotRegionSplit);
229      if (localityEnabled) {
230        // When localityEnabled is true, meant to verify split.getLocations()
231        // by the following statement:
232        //   Assert.assertTrue(split.getLocations() != null && split.getLocations().length != 0);
233        // However, getLocations() of some splits could return an empty array (length is 0),
234        // so drop the verification on length.
235        // TODO: investigate how to verify split.getLocations() when localityEnabled is true
236        Assert.assertTrue(split.getLocations() != null);
237      } else {
238        Assert.assertTrue(split.getLocations() != null && split.getLocations().length == 0);
239      }
240
241      // validate record reader
242      OutputCollector collector = mock(OutputCollector.class);
243      Reporter reporter = mock(Reporter.class);
244      RecordReader<ImmutableBytesWritable, Result> rr = tsif.getRecordReader(split, job, reporter);
245
246      // validate we can read all the data back
247      ImmutableBytesWritable key = rr.createKey();
248      Result value = rr.createValue();
249      while (rr.next(key, value)) {
250        verifyRowFromMap(key, value);
251        rowTracker.addRow(key.copyBytes());
252      }
253
254      rr.close();
255    }
256
257    // validate all rows are seen
258    rowTracker.validate();
259  }
260
261  @Override
262  protected void testWithMapReduceImpl(HBaseTestingUtility util, TableName tableName,
263      String snapshotName, Path tableDir, int numRegions, int numSplitsPerRegion, int expectedNumSplits,
264      boolean shutdownCluster) throws Exception {
265    doTestWithMapReduce(util, tableName, snapshotName, getStartRow(), getEndRow(), tableDir,
266      numRegions, numSplitsPerRegion, expectedNumSplits, shutdownCluster);
267  }
268
269  // this is also called by the IntegrationTestTableSnapshotInputFormat
270  public static void doTestWithMapReduce(HBaseTestingUtility util, TableName tableName,
271      String snapshotName, byte[] startRow, byte[] endRow, Path tableDir, int numRegions,
272      int numSplitsPerRegion,int expectedNumSplits, boolean shutdownCluster) throws Exception {
273
274    //create the table and snapshot
275    createTableAndSnapshot(util, tableName, snapshotName, startRow, endRow, numRegions);
276
277    if (shutdownCluster) {
278      util.shutdownMiniHBaseCluster();
279    }
280
281    try {
282      // create the job
283      JobConf jobConf = new JobConf(util.getConfiguration());
284
285      jobConf.setJarByClass(util.getClass());
286      org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil.addDependencyJarsForClasses(jobConf,
287        TestTableSnapshotInputFormat.class);
288
289      if(numSplitsPerRegion > 1) {
290        TableMapReduceUtil.initTableSnapshotMapJob(snapshotName, COLUMNS,
291                TestTableSnapshotMapper.class, ImmutableBytesWritable.class,
292                NullWritable.class, jobConf, true, tableDir, new RegionSplitter.UniformSplit(),
293                numSplitsPerRegion);
294      } else {
295        TableMapReduceUtil.initTableSnapshotMapJob(snapshotName, COLUMNS,
296                TestTableSnapshotMapper.class, ImmutableBytesWritable.class,
297                NullWritable.class, jobConf, true, tableDir);
298      }
299
300      jobConf.setReducerClass(TestTableSnapshotInputFormat.TestTableSnapshotReducer.class);
301      jobConf.setNumReduceTasks(1);
302      jobConf.setOutputFormat(NullOutputFormat.class);
303
304      RunningJob job = JobClient.runJob(jobConf);
305      Assert.assertTrue(job.isSuccessful());
306    } finally {
307      if (!shutdownCluster) {
308        util.getAdmin().deleteSnapshot(snapshotName);
309        util.deleteTable(tableName);
310      }
311    }
312  }
313}