1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.mapreduce;
20
21 import static org.mockito.Mockito.mock;
22 import static org.mockito.Mockito.when;
23
24 import java.io.IOException;
25 import java.util.Arrays;
26 import java.util.List;
27
28 import org.apache.hadoop.conf.Configuration;
29 import org.apache.hadoop.fs.FileSystem;
30 import org.apache.hadoop.fs.Path;
31 import org.apache.hadoop.hbase.CategoryBasedTimeout;
32 import org.apache.hadoop.hbase.HBaseTestingUtility;
33 import org.apache.hadoop.hbase.HConstants;
34 import org.apache.hadoop.hbase.HDFSBlocksDistribution;
35 import org.apache.hadoop.hbase.HRegionInfo;
36 import org.apache.hadoop.hbase.client.HBaseAdmin;
37 import org.apache.hadoop.hbase.client.Table;
38 import org.apache.hadoop.hbase.snapshot.SnapshotTestingUtils;
39 import org.apache.hadoop.hbase.testclassification.LargeTests;
40 import org.apache.hadoop.hbase.TableName;
41 import org.apache.hadoop.hbase.client.Result;
42 import org.apache.hadoop.hbase.client.Scan;
43 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
44 import org.apache.hadoop.hbase.mapreduce.TableSnapshotInputFormat.TableSnapshotRegionSplit;
45 import org.apache.hadoop.hbase.util.Bytes;
46 import org.apache.hadoop.hbase.util.FSUtils;
47 import org.apache.hadoop.hbase.util.RegionSplitter;
48 import org.apache.hadoop.io.NullWritable;
49 import org.apache.hadoop.mapreduce.InputSplit;
50 import org.apache.hadoop.mapreduce.Job;
51 import org.apache.hadoop.mapreduce.RecordReader;
52 import org.apache.hadoop.mapreduce.Reducer;
53 import org.apache.hadoop.mapreduce.TaskAttemptContext;
54 import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
55 import org.junit.After;
56 import org.junit.Assert;
57 import org.junit.Rule;
58 import org.junit.Test;
59 import org.junit.experimental.categories.Category;
60 import org.junit.rules.TestRule;
61
62 import com.google.common.collect.Lists;
63
64 @Category(LargeTests.class)
65 public class TestTableSnapshotInputFormat extends TableSnapshotInputFormatTestBase {
66 @Rule public final TestRule timeout = CategoryBasedTimeout.builder().
67 withTimeout(this.getClass()).withLookingForStuckThread(true).build();
68
69 private static final byte[] bbb = Bytes.toBytes("bbb");
70 private static final byte[] yyy = Bytes.toBytes("yyy");
71 private static final byte[] bbc = Bytes.toBytes("bbc");
72 private static final byte[] yya = Bytes.toBytes("yya");
73
74 @Override
75 protected byte[] getStartRow() {
76 return bbb;
77 }
78
79 @Override
80 protected byte[] getEndRow() {
81 return yyy;
82 }
83
84 @After
85 public void tearDown() throws Exception {
86 }
87
88 @Test
89 public void testGetBestLocations() throws IOException {
90 TableSnapshotInputFormatImpl tsif = new TableSnapshotInputFormatImpl();
91 Configuration conf = UTIL.getConfiguration();
92
93 HDFSBlocksDistribution blockDistribution = new HDFSBlocksDistribution();
94 Assert.assertEquals(Lists.newArrayList(), tsif.getBestLocations(conf, blockDistribution));
95
96 blockDistribution.addHostsAndBlockWeight(new String[] {"h1"}, 1);
97 Assert.assertEquals(Lists.newArrayList("h1"), tsif.getBestLocations(conf, blockDistribution));
98
99 blockDistribution.addHostsAndBlockWeight(new String[] {"h1"}, 1);
100 Assert.assertEquals(Lists.newArrayList("h1"), tsif.getBestLocations(conf, blockDistribution));
101
102 blockDistribution.addHostsAndBlockWeight(new String[] {"h2"}, 1);
103 Assert.assertEquals(Lists.newArrayList("h1"), tsif.getBestLocations(conf, blockDistribution));
104
105 blockDistribution = new HDFSBlocksDistribution();
106 blockDistribution.addHostsAndBlockWeight(new String[] {"h1"}, 10);
107 blockDistribution.addHostsAndBlockWeight(new String[] {"h2"}, 7);
108 blockDistribution.addHostsAndBlockWeight(new String[] {"h3"}, 5);
109 blockDistribution.addHostsAndBlockWeight(new String[] {"h4"}, 1);
110 Assert.assertEquals(Lists.newArrayList("h1"), tsif.getBestLocations(conf, blockDistribution));
111
112 blockDistribution.addHostsAndBlockWeight(new String[] {"h2"}, 2);
113 Assert.assertEquals(Lists.newArrayList("h1", "h2"),
114 tsif.getBestLocations(conf, blockDistribution));
115
116 blockDistribution.addHostsAndBlockWeight(new String[] {"h2"}, 3);
117 Assert.assertEquals(Lists.newArrayList("h2", "h1"),
118 tsif.getBestLocations(conf, blockDistribution));
119
120 blockDistribution.addHostsAndBlockWeight(new String[] {"h3"}, 6);
121 blockDistribution.addHostsAndBlockWeight(new String[] {"h4"}, 9);
122
123 Assert.assertEquals(Lists.newArrayList("h2", "h3", "h4", "h1"),
124 tsif.getBestLocations(conf, blockDistribution));
125 }
126
127 public static enum TestTableSnapshotCounters {
128 VALIDATION_ERROR
129 }
130
131 public static class TestTableSnapshotMapper
132 extends TableMapper<ImmutableBytesWritable, NullWritable> {
133 @Override
134 protected void map(ImmutableBytesWritable key, Result value,
135 Context context) throws IOException, InterruptedException {
136
137 verifyRowFromMap(key, value);
138 context.write(key, NullWritable.get());
139 }
140 }
141
142 public static class TestTableSnapshotReducer
143 extends Reducer<ImmutableBytesWritable, NullWritable, NullWritable, NullWritable> {
144 HBaseTestingUtility.SeenRowTracker rowTracker =
145 new HBaseTestingUtility.SeenRowTracker(bbb, yyy);
146 @Override
147 protected void reduce(ImmutableBytesWritable key, Iterable<NullWritable> values,
148 Context context) throws IOException, InterruptedException {
149 rowTracker.addRow(key.get());
150 }
151
152 @Override
153 protected void cleanup(Context context) throws IOException,
154 InterruptedException {
155 rowTracker.validate();
156 }
157 }
158
159 @Test
160 public void testInitTableSnapshotMapperJobConfig() throws Exception {
161 setupCluster();
162 TableName tableName = TableName.valueOf("testInitTableSnapshotMapperJobConfig");
163 String snapshotName = "foo";
164
165 try {
166 createTableAndSnapshot(UTIL, tableName, snapshotName, getStartRow(), getEndRow(), 1);
167 Job job = new Job(UTIL.getConfiguration());
168 Path tmpTableDir = UTIL.getRandomDir();
169
170 TableMapReduceUtil.initTableSnapshotMapperJob(snapshotName,
171 new Scan(), TestTableSnapshotMapper.class, ImmutableBytesWritable.class,
172 NullWritable.class, job, false, tmpTableDir);
173
174
175
176 Assert.assertEquals(
177 "Snapshot job should be configured for default LruBlockCache.",
178 HConstants.HFILE_BLOCK_CACHE_SIZE_DEFAULT,
179 job.getConfiguration().getFloat(HConstants.HFILE_BLOCK_CACHE_SIZE_KEY, -1), 0.01);
180 Assert.assertEquals(
181 "Snapshot job should not use BucketCache.",
182 0, job.getConfiguration().getFloat("hbase.bucketcache.size", -1), 0.01);
183 } finally {
184 UTIL.getHBaseAdmin().deleteSnapshot(snapshotName);
185 UTIL.deleteTable(tableName);
186 tearDownCluster();
187 }
188 }
189
190 @Override
191 public void testRestoreSnapshotDoesNotCreateBackRefLinksInit(TableName tableName,
192 String snapshotName, Path tmpTableDir) throws Exception {
193 Job job = new Job(UTIL.getConfiguration());
194 TableMapReduceUtil.initTableSnapshotMapperJob(snapshotName,
195 new Scan(), TestTableSnapshotMapper.class, ImmutableBytesWritable.class,
196 NullWritable.class, job, false, tmpTableDir);
197 }
198
199 @Override
200 public void testWithMockedMapReduce(HBaseTestingUtility util, String snapshotName,
201 int numRegions, int numSplitsPerRegion, int expectedNumSplits) throws Exception {
202 setupCluster();
203 TableName tableName = TableName.valueOf("testWithMockedMapReduce");
204 try {
205 createTableAndSnapshot(
206 util, tableName, snapshotName, getStartRow(), getEndRow(), numRegions);
207
208 Job job = new Job(util.getConfiguration());
209 Path tmpTableDir = util.getRandomDir();
210 Scan scan = new Scan(getStartRow(), getEndRow());
211
212 if (numSplitsPerRegion > 1) {
213 TableMapReduceUtil.initTableSnapshotMapperJob(snapshotName,
214 scan, TestTableSnapshotMapper.class, ImmutableBytesWritable.class,
215 NullWritable.class, job, false, tmpTableDir, new RegionSplitter.UniformSplit(),
216 numSplitsPerRegion);
217 } else {
218 TableMapReduceUtil.initTableSnapshotMapperJob(snapshotName,
219 scan, TestTableSnapshotMapper.class, ImmutableBytesWritable.class,
220 NullWritable.class, job, false, tmpTableDir);
221 }
222
223 verifyWithMockedMapReduce(job, numRegions, expectedNumSplits, getStartRow(), getEndRow());
224
225 } finally {
226 util.getHBaseAdmin().deleteSnapshot(snapshotName);
227 util.deleteTable(tableName);
228 tearDownCluster();
229 }
230 }
231
232 public static void blockUntilSplitFinished(HBaseTestingUtility util, TableName tableName,
233 int expectedRegionSize) throws Exception {
234 for (int i = 0; i < 100; i++) {
235 List<HRegionInfo> hRegionInfoList = util.getHBaseAdmin().getTableRegions(tableName);
236 if (hRegionInfoList.size() >= expectedRegionSize) {
237 break;
238 }
239 Thread.sleep(1000);
240 }
241 }
242
243 @Test
244 public void testWithMockedMapReduceWithSplitsPerRegion() throws Exception {
245 setupCluster();
246 String snapshotName = "testWithMockedMapReduceWithSplitsPerRegion";
247 final TableName tableName = TableName.valueOf(snapshotName);
248 try {
249 createTableAndSnapshot(UTIL, tableName, snapshotName, getStartRow(), getEndRow(), 10);
250
251 Configuration conf = UTIL.getConfiguration();
252 Job job = new Job(conf);
253 Path tmpTableDir = UTIL.getRandomDir();
254
255 Scan scan = new Scan(bbc, yya);
256
257 TableMapReduceUtil.initTableSnapshotMapperJob(snapshotName, scan,
258 TestTableSnapshotMapper.class, ImmutableBytesWritable.class, NullWritable.class, job, false,
259 tmpTableDir, new RegionSplitter.UniformSplit(), 5);
260
261 verifyWithMockedMapReduce(job, 10, 40, bbc, yya);
262 } finally {
263 UTIL.getHBaseAdmin().deleteSnapshot(snapshotName);
264 UTIL.deleteTable(tableName);
265 tearDownCluster();
266 }
267 }
268
269 @Test
270 public void testWithMockedMapReduceWithNoStartRowStopRow() throws Exception {
271 setupCluster();
272 String snapshotName = "testWithMockedMapReduceWithNoStartRowStopRow";
273 final TableName tableName = TableName.valueOf(snapshotName);
274 try {
275 createTableAndSnapshot(UTIL, tableName, snapshotName, getStartRow(), getEndRow(), 10);
276
277 Configuration conf = UTIL.getConfiguration();
278 Job job = new Job(conf);
279 Path tmpTableDir = UTIL.getRandomDir();
280
281 Scan scan2 = new Scan();
282
283 TableMapReduceUtil.initTableSnapshotMapperJob(snapshotName, scan2,
284 TestTableSnapshotMapper.class, ImmutableBytesWritable.class, NullWritable.class, job, false,
285 tmpTableDir, new RegionSplitter.UniformSplit(), 5);
286
287 verifyWithMockedMapReduce(job, 10, 50, HConstants.EMPTY_START_ROW,
288 HConstants.EMPTY_START_ROW);
289
290 } finally {
291 UTIL.getHBaseAdmin().deleteSnapshot(snapshotName);
292 UTIL.deleteTable(tableName);
293 tearDownCluster();
294 }
295 }
296
297 @Test
298 public void testNoDuplicateResultsWhenSplitting() throws Exception {
299 setupCluster();
300 TableName tableName = TableName.valueOf("testNoDuplicateResultsWhenSplitting");
301 String snapshotName = "testSnapshotBug";
302 try {
303 if (UTIL.getHBaseAdmin().tableExists(tableName)) {
304 UTIL.deleteTable(tableName);
305 }
306
307 UTIL.createTable(tableName, FAMILIES);
308 HBaseAdmin admin = UTIL.getHBaseAdmin();
309
310
311 Table table = UTIL.getConnection().getTable(tableName);
312 UTIL.loadTable(table, FAMILIES);
313
314
315 admin.split(tableName, Bytes.toBytes("eee"));
316 blockUntilSplitFinished(UTIL, tableName, 2);
317
318 Path rootDir = FSUtils.getRootDir(UTIL.getConfiguration());
319 FileSystem fs = rootDir.getFileSystem(UTIL.getConfiguration());
320
321 SnapshotTestingUtils.createSnapshotAndValidate(admin, tableName, Arrays.asList(FAMILIES),
322 null, snapshotName, rootDir, fs, true);
323
324
325 byte[] value = Bytes.toBytes("after_snapshot_value");
326 UTIL.loadTable(table, FAMILIES, value);
327
328
329 admin.flush(tableName);
330 table.close();
331
332 Job job = new Job(UTIL.getConfiguration());
333 Path tmpTableDir = UTIL.getRandomDir();
334
335 Scan scan = new Scan().setStartRow(getStartRow()).setStopRow(getEndRow());
336
337 TableMapReduceUtil.initTableSnapshotMapperJob(snapshotName, scan,
338 TestTableSnapshotMapper.class, ImmutableBytesWritable.class, NullWritable.class, job, false,
339 tmpTableDir);
340
341 verifyWithMockedMapReduce(job, 2, 2, getStartRow(), getEndRow());
342 } finally {
343 UTIL.getHBaseAdmin().deleteSnapshot(snapshotName);
344 UTIL.deleteTable(tableName);
345 tearDownCluster();
346 }
347 }
348
349 private void verifyWithMockedMapReduce(Job job, int numRegions, int expectedNumSplits,
350 byte[] startRow, byte[] stopRow)
351 throws IOException, InterruptedException {
352 TableSnapshotInputFormat tsif = new TableSnapshotInputFormat();
353 List<InputSplit> splits = tsif.getSplits(job);
354
355 Assert.assertEquals(expectedNumSplits, splits.size());
356
357 HBaseTestingUtility.SeenRowTracker rowTracker = new HBaseTestingUtility.SeenRowTracker(startRow,
358 stopRow.length > 0 ? stopRow : Bytes.toBytes("\uffff"));
359
360 for (int i = 0; i < splits.size(); i++) {
361
362 InputSplit split = splits.get(i);
363 Assert.assertTrue(split instanceof TableSnapshotRegionSplit);
364 TableSnapshotRegionSplit snapshotRegionSplit = (TableSnapshotRegionSplit) split;
365 Scan scan =
366 TableMapReduceUtil.convertStringToScan(snapshotRegionSplit.getDelegate().getScan());
367 if (startRow.length > 0) {
368 Assert.assertTrue(
369 Bytes.toStringBinary(startRow) + " should <= " + Bytes.toStringBinary(scan.getStartRow()),
370 Bytes.compareTo(startRow, scan.getStartRow()) <= 0);
371 }
372 if (stopRow.length > 0) {
373 Assert.assertTrue(
374 Bytes.toStringBinary(stopRow) + " should >= " + Bytes.toStringBinary(scan.getStopRow()),
375 Bytes.compareTo(stopRow, scan.getStopRow()) >= 0);
376 }
377 Assert.assertTrue("startRow should < stopRow",
378 Bytes.compareTo(scan.getStartRow(), scan.getStopRow()) < 0);
379
380
381 TaskAttemptContext taskAttemptContext = mock(TaskAttemptContext.class);
382 when(taskAttemptContext.getConfiguration()).thenReturn(job.getConfiguration());
383 RecordReader<ImmutableBytesWritable, Result> rr =
384 tsif.createRecordReader(split, taskAttemptContext);
385 rr.initialize(split, taskAttemptContext);
386
387
388 while (rr.nextKeyValue()) {
389 byte[] row = rr.getCurrentKey().get();
390 verifyRowFromMap(rr.getCurrentKey(), rr.getCurrentValue());
391 rowTracker.addRow(row);
392 }
393
394 rr.close();
395 }
396
397
398 rowTracker.validate();
399 }
400
401 @Override
402 protected void testWithMapReduceImpl(HBaseTestingUtility util, TableName tableName,
403 String snapshotName, Path tableDir, int numRegions, int numSplitsPerRegion, int expectedNumSplits,
404 boolean shutdownCluster) throws Exception {
405 doTestWithMapReduce(util, tableName, snapshotName, getStartRow(), getEndRow(), tableDir,
406 numRegions, numSplitsPerRegion, expectedNumSplits, shutdownCluster);
407 }
408
409
410 public static void doTestWithMapReduce(HBaseTestingUtility util, TableName tableName,
411 String snapshotName, byte[] startRow, byte[] endRow, Path tableDir, int numRegions,
412 int numSplitsPerRegion, int expectedNumSplits, boolean shutdownCluster) throws Exception {
413
414
415 createTableAndSnapshot(util, tableName, snapshotName, startRow, endRow, numRegions);
416
417 if (shutdownCluster) {
418 util.shutdownMiniHBaseCluster();
419 }
420
421 try {
422
423 Job job = new Job(util.getConfiguration());
424 Scan scan = new Scan(startRow, endRow);
425
426 job.setJarByClass(util.getClass());
427 TableMapReduceUtil.addDependencyJarsForClasses(job.getConfiguration(),
428 TestTableSnapshotInputFormat.class);
429
430 if (numSplitsPerRegion > 1) {
431 TableMapReduceUtil.initTableSnapshotMapperJob(snapshotName,
432 scan, TestTableSnapshotMapper.class, ImmutableBytesWritable.class,
433 NullWritable.class, job, true, tableDir, new RegionSplitter.UniformSplit(),
434 numSplitsPerRegion);
435 } else {
436 TableMapReduceUtil.initTableSnapshotMapperJob(snapshotName,
437 scan, TestTableSnapshotMapper.class, ImmutableBytesWritable.class,
438 NullWritable.class, job, true, tableDir);
439 }
440
441 job.setReducerClass(TestTableSnapshotInputFormat.TestTableSnapshotReducer.class);
442 job.setNumReduceTasks(1);
443 job.setOutputFormatClass(NullOutputFormat.class);
444
445 Assert.assertTrue(job.waitForCompletion(true));
446 } finally {
447 if (!shutdownCluster) {
448 util.getHBaseAdmin().deleteSnapshot(snapshotName);
449 util.deleteTable(tableName);
450 }
451 }
452 }
453 }