1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.mapreduce;
19
20 import static org.junit.Assert.assertEquals;
21 import static org.junit.Assert.assertFalse;
22 import static org.junit.Assert.assertTrue;
23 import static org.junit.Assert.fail;
24 import static org.mockito.Matchers.any;
25 import static org.mockito.Mockito.doAnswer;
26 import static org.mockito.Mockito.mock;
27 import static org.mockito.Mockito.when;
28
29 import java.io.ByteArrayOutputStream;
30 import java.io.File;
31 import java.io.IOException;
32 import java.io.PrintStream;
33 import java.net.URL;
34 import java.util.ArrayList;
35 import java.util.Arrays;
36 import java.util.List;
37
38 import org.apache.commons.logging.Log;
39 import org.apache.commons.logging.LogFactory;
40 import org.apache.hadoop.conf.Configuration;
41 import org.apache.hadoop.fs.FileSystem;
42 import org.apache.hadoop.fs.Path;
43 import org.apache.hadoop.hbase.HConstants;
44 import org.apache.hadoop.hbase.Cell;
45 import org.apache.hadoop.hbase.CellUtil;
46 import org.apache.hadoop.hbase.CategoryBasedTimeout;
47 import org.apache.hadoop.hbase.HBaseTestingUtility;
48 import org.apache.hadoop.hbase.HColumnDescriptor;
49 import org.apache.hadoop.hbase.HRegionInfo;
50 import org.apache.hadoop.hbase.HTableDescriptor;
51 import org.apache.hadoop.hbase.KeyValue;
52 import org.apache.hadoop.hbase.testclassification.MediumTests;
53 import org.apache.hadoop.hbase.TableName;
54 import org.apache.hadoop.hbase.client.Delete;
55 import org.apache.hadoop.hbase.client.Durability;
56 import org.apache.hadoop.hbase.client.Get;
57 import org.apache.hadoop.hbase.client.HTable;
58 import org.apache.hadoop.hbase.client.Put;
59 import org.apache.hadoop.hbase.client.Result;
60 import org.apache.hadoop.hbase.client.ResultScanner;
61 import org.apache.hadoop.hbase.client.Scan;
62 import org.apache.hadoop.hbase.client.Table;
63 import org.apache.hadoop.hbase.filter.Filter;
64 import org.apache.hadoop.hbase.filter.FilterBase;
65 import org.apache.hadoop.hbase.filter.PrefixFilter;
66 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
67 import org.apache.hadoop.hbase.mapreduce.Import.KeyValueImporter;
68 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
69 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
70 import org.apache.hadoop.hbase.wal.WAL;
71 import org.apache.hadoop.hbase.wal.WALKey;
72 import org.apache.hadoop.hbase.util.Bytes;
73 import org.apache.hadoop.hbase.util.LauncherSecurityManager;
74 import org.apache.hadoop.mapreduce.Job;
75 import org.apache.hadoop.mapreduce.Mapper.Context;
76 import org.apache.hadoop.util.GenericOptionsParser;
77 import org.junit.After;
78 import org.junit.AfterClass;
79 import org.junit.Assert;
80 import org.junit.Before;
81 import org.junit.BeforeClass;
82 import org.junit.Rule;
83 import org.junit.Test;
84 import org.junit.experimental.categories.Category;
85 import org.junit.rules.TestRule;
86 import org.mockito.invocation.InvocationOnMock;
87 import org.mockito.stubbing.Answer;
88
89
90
91
92 @Category(MediumTests.class)
93 public class TestImportExport {
94 private static final Log LOG = LogFactory.getLog(TestImportExport.class);
95 private static final HBaseTestingUtility UTIL = new HBaseTestingUtility();
96 private static final byte[] ROW1 = Bytes.toBytesBinary("\\x32row1");
97 private static final byte[] ROW2 = Bytes.toBytesBinary("\\x32row2");
98 private static final byte[] ROW3 = Bytes.toBytesBinary("\\x32row3");
99 private static final String FAMILYA_STRING = "a";
100 private static final String FAMILYB_STRING = "b";
101 private static final byte[] FAMILYA = Bytes.toBytes(FAMILYA_STRING);
102 private static final byte[] FAMILYB = Bytes.toBytes(FAMILYB_STRING);
103 private static final byte[] QUAL = Bytes.toBytes("q");
104 private static final String OUTPUT_DIR = "outputdir";
105 private static String FQ_OUTPUT_DIR;
106 private static final String EXPORT_BATCH_SIZE = "100";
107
108 private static long now = System.currentTimeMillis();
109
110 @Rule
111 public final TestRule timeout = CategoryBasedTimeout.builder().withTimeout(this.getClass()).
112 withLookingForStuckThread(true).build();
113
114 @BeforeClass
115 public static void beforeClass() throws Exception {
116
117 UTIL.getConfiguration().setBoolean(HBaseTestingUtility.USE_LOCAL_FILESYSTEM, true);
118 UTIL.getConfiguration().setInt(HConstants.REGION_SERVER_HIGH_PRIORITY_HANDLER_COUNT, 10);
119 UTIL.setJobWithoutMRCluster();
120 UTIL.startMiniCluster();
121 FQ_OUTPUT_DIR =
122 new Path(OUTPUT_DIR).makeQualified(FileSystem.get(UTIL.getConfiguration())).toString();
123 }
124
125 @AfterClass
126 public static void afterClass() throws Exception {
127 UTIL.shutdownMiniCluster();
128 }
129
130 @Before
131 @After
132 public void cleanup() throws Exception {
133 FileSystem fs = FileSystem.get(UTIL.getConfiguration());
134 fs.delete(new Path(OUTPUT_DIR), true);
135 }
136
137
138
139
140
141
142
143
144
145 boolean runExport(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
146
147 GenericOptionsParser opts = new GenericOptionsParser(new Configuration(UTIL.getConfiguration()), args);
148 Configuration conf = opts.getConfiguration();
149 args = opts.getRemainingArgs();
150 Job job = Export.createSubmittableJob(conf, args);
151 job.waitForCompletion(false);
152 return job.isSuccessful();
153 }
154
155
156
157
158
159
160
161
162
163 boolean runImport(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
164
165 GenericOptionsParser opts = new GenericOptionsParser(new Configuration(UTIL.getConfiguration()), args);
166 Configuration conf = opts.getConfiguration();
167 args = opts.getRemainingArgs();
168 Job job = Import.createSubmittableJob(conf, args);
169 job.waitForCompletion(false);
170 return job.isSuccessful();
171 }
172
173
174
175
176
177 @Test
178 public void testSimpleCase() throws Exception {
179 String EXPORT_TABLE = "exportSimpleCase";
180 try (Table t = UTIL.createTable(TableName.valueOf(EXPORT_TABLE), FAMILYA, 3);) {
181 Put p = new Put(ROW1);
182 p.add(FAMILYA, QUAL, now, QUAL);
183 p.add(FAMILYA, QUAL, now+1, QUAL);
184 p.add(FAMILYA, QUAL, now+2, QUAL);
185 t.put(p);
186 p = new Put(ROW2);
187 p.add(FAMILYA, QUAL, now, QUAL);
188 p.add(FAMILYA, QUAL, now+1, QUAL);
189 p.add(FAMILYA, QUAL, now+2, QUAL);
190 t.put(p);
191 p = new Put(ROW3);
192 p.add(FAMILYA, QUAL, now, QUAL);
193 p.add(FAMILYA, QUAL, now + 1, QUAL);
194 p.add(FAMILYA, QUAL, now + 2, QUAL);
195 t.put(p);
196 }
197
198 String[] args = new String[] {
199
200 "-D" + TableInputFormat.SCAN_ROW_START + "=\\x32row1",
201 "-D" + TableInputFormat.SCAN_ROW_STOP + "=\\x32row3",
202 EXPORT_TABLE,
203 FQ_OUTPUT_DIR,
204 "1000",
205 };
206 assertTrue(runExport(args));
207
208 String IMPORT_TABLE = "importTableSimpleCase";
209 try (Table t = UTIL.createTable(TableName.valueOf(IMPORT_TABLE), FAMILYB, 3);) {
210 args = new String[] {
211 "-D" + Import.CF_RENAME_PROP + "="+FAMILYA_STRING+":"+FAMILYB_STRING,
212 IMPORT_TABLE,
213 FQ_OUTPUT_DIR
214 };
215 assertTrue(runImport(args));
216
217 Get g = new Get(ROW1);
218 g.setMaxVersions();
219 Result r = t.get(g);
220 assertEquals(3, r.size());
221 g = new Get(ROW2);
222 g.setMaxVersions();
223 r = t.get(g);
224 assertEquals(3, r.size());
225 g = new Get(ROW3);
226 r = t.get(g);
227 assertEquals(0, r.size());
228 }
229 }
230
231
232
233
234
235
236 @Test
237 public void testMetaExport() throws Exception {
238 String EXPORT_TABLE = TableName.META_TABLE_NAME.getNameAsString();
239 String[] args = new String[] { EXPORT_TABLE, FQ_OUTPUT_DIR, "1", "0", "0" };
240 assertTrue(runExport(args));
241 }
242
243
244
245
246
247 @Test
248 public void testImport94Table() throws Exception {
249 final String name = "exportedTableIn94Format";
250 URL url = TestImportExport.class.getResource(name);
251 File f = new File(url.toURI());
252 if (!f.exists()) {
253 LOG.warn("FAILED TO FIND " + f + "; skipping out on test");
254 return;
255 }
256 assertTrue(f.exists());
257 LOG.info("FILE=" + f);
258 Path importPath = new Path(f.toURI());
259 FileSystem fs = FileSystem.get(UTIL.getConfiguration());
260 fs.copyFromLocalFile(importPath, new Path(FQ_OUTPUT_DIR + Path.SEPARATOR + name));
261 String IMPORT_TABLE = name;
262 try (Table t = UTIL.createTable(TableName.valueOf(IMPORT_TABLE), Bytes.toBytes("f1"), 3);) {
263 String[] args = new String[] {
264 "-Dhbase.import.version=0.94" ,
265 IMPORT_TABLE, FQ_OUTPUT_DIR
266 };
267 assertTrue(runImport(args));
268
269
270
271
272
273
274
275
276 assertEquals(5, UTIL.countRows(t));
277 }
278 }
279
280
281
282
283 @Test
284 public void testExportScannerBatching() throws Exception {
285 String BATCH_TABLE = "exportWithBatch";
286 HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(BATCH_TABLE));
287 desc.addFamily(new HColumnDescriptor(FAMILYA)
288 .setMaxVersions(1)
289 );
290 UTIL.getHBaseAdmin().createTable(desc);
291 try (Table t = new HTable(UTIL.getConfiguration(), desc.getTableName());) {
292 Put p = new Put(ROW1);
293 p.add(FAMILYA, QUAL, now, QUAL);
294 p.add(FAMILYA, QUAL, now+1, QUAL);
295 p.add(FAMILYA, QUAL, now+2, QUAL);
296 p.add(FAMILYA, QUAL, now+3, QUAL);
297 p.add(FAMILYA, QUAL, now+4, QUAL);
298 t.put(p);
299
300 String[] args = new String[] {
301 "-D" + Export.EXPORT_BATCHING + "=" + EXPORT_BATCH_SIZE,
302 BATCH_TABLE,
303 FQ_OUTPUT_DIR
304 };
305 assertTrue(runExport(args));
306
307 FileSystem fs = FileSystem.get(UTIL.getConfiguration());
308 fs.delete(new Path(FQ_OUTPUT_DIR), true);
309 }
310 }
311
312 @Test
313 public void testWithDeletes() throws Exception {
314 String IMPORT_TABLE = "importWithDeletes";
315 String EXPORT_TABLE = "exportWithDeletes";
316 HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(EXPORT_TABLE));
317 desc.addFamily(new HColumnDescriptor(FAMILYA)
318 .setMaxVersions(5)
319 .setKeepDeletedCells(true)
320 );
321 UTIL.getHBaseAdmin().createTable(desc);
322 try (Table t = new HTable(UTIL.getConfiguration(), desc.getTableName());) {
323 Put p = new Put(ROW1);
324 p.add(FAMILYA, QUAL, now, QUAL);
325 p.add(FAMILYA, QUAL, now+1, QUAL);
326 p.add(FAMILYA, QUAL, now+2, QUAL);
327 p.add(FAMILYA, QUAL, now+3, QUAL);
328 p.add(FAMILYA, QUAL, now+4, QUAL);
329 t.put(p);
330
331 Delete d = new Delete(ROW1, now+3);
332 t.delete(d);
333 d = new Delete(ROW1);
334 d.deleteColumns(FAMILYA, QUAL, now+2);
335 t.delete(d);
336
337 String[] args = new String[] {
338 "-D" + Export.RAW_SCAN + "=true",
339 EXPORT_TABLE,
340 FQ_OUTPUT_DIR,
341 "1000",
342 };
343 assertTrue(runExport(args));
344
345 desc = new HTableDescriptor(TableName.valueOf(IMPORT_TABLE));
346 desc.addFamily(new HColumnDescriptor(FAMILYA)
347 .setMaxVersions(5)
348 .setKeepDeletedCells(true)
349 );
350 }
351 UTIL.getHBaseAdmin().createTable(desc);
352 try (Table t = new HTable(UTIL.getConfiguration(), desc.getTableName());) {
353 String [] args = new String[] {
354 IMPORT_TABLE,
355 FQ_OUTPUT_DIR
356 };
357 assertTrue(runImport(args));
358
359 Scan s = new Scan();
360 s.setMaxVersions();
361 s.setRaw(true);
362 ResultScanner scanner = t.getScanner(s);
363 Result r = scanner.next();
364 Cell[] res = r.rawCells();
365 assertTrue(CellUtil.isDeleteFamily(res[0]));
366 assertEquals(now+4, res[1].getTimestamp());
367 assertEquals(now+3, res[2].getTimestamp());
368 assertTrue(CellUtil.isDelete(res[3]));
369 assertEquals(now+2, res[4].getTimestamp());
370 assertEquals(now+1, res[5].getTimestamp());
371 assertEquals(now, res[6].getTimestamp());
372 }
373 }
374
375 @Test
376 public void testWithMultipleDeleteFamilyMarkersOfSameRowSameFamily() throws Exception {
377 String EXPORT_TABLE = "exportWithMultipleDeleteFamilyMarkersOfSameRowSameFamily";
378 HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(EXPORT_TABLE));
379 desc.addFamily(new HColumnDescriptor(FAMILYA)
380 .setMaxVersions(5)
381 .setKeepDeletedCells(true)
382 );
383 UTIL.getHBaseAdmin().createTable(desc);
384 HTable exportT = new HTable(UTIL.getConfiguration(), EXPORT_TABLE);
385
386 Put p = new Put(ROW1);
387 p.add(FAMILYA, QUAL, now, QUAL);
388 exportT.put(p);
389
390
391 Delete d = new Delete(ROW1, now+3);
392 exportT.delete(d);
393
394
395 p = new Put(ROW1);
396 p.add(FAMILYA, QUAL, now + 5, "s".getBytes());
397 exportT.put(p);
398
399
400 d = new Delete(ROW1, now+7);
401 exportT.delete(d);
402
403
404 String[] args = new String[] {
405 "-D" + Export.RAW_SCAN + "=true",
406 EXPORT_TABLE,
407 FQ_OUTPUT_DIR,
408 "1000",
409 };
410 assertTrue(runExport(args));
411
412 String IMPORT_TABLE = "importWithMultipleDeleteFamilyMarkersOfSameRowSameFamily";
413 desc = new HTableDescriptor(TableName.valueOf(IMPORT_TABLE));
414 desc.addFamily(new HColumnDescriptor(FAMILYA)
415 .setMaxVersions(5)
416 .setKeepDeletedCells(true)
417 );
418 UTIL.getHBaseAdmin().createTable(desc);
419
420 HTable importT = new HTable(UTIL.getConfiguration(), IMPORT_TABLE);
421 args = new String[] {
422 IMPORT_TABLE,
423 FQ_OUTPUT_DIR
424 };
425 assertTrue(runImport(args));
426
427 Scan s = new Scan();
428 s.setMaxVersions();
429 s.setRaw(true);
430
431 ResultScanner importedTScanner = importT.getScanner(s);
432 Result importedTResult = importedTScanner.next();
433
434 ResultScanner exportedTScanner = exportT.getScanner(s);
435 Result exportedTResult = exportedTScanner.next();
436 try {
437 Result.compareResults(exportedTResult, importedTResult);
438 } catch (Exception e) {
439 fail("Original and imported tables data comparision failed with error:"+e.getMessage());
440 } finally {
441 exportT.close();
442 importT.close();
443 }
444 }
445
446
447
448
449
450 @Test
451 public void testWithFilter() throws Exception {
452
453 String EXPORT_TABLE = "exportSimpleCase_ImportWithFilter";
454 HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(EXPORT_TABLE));
455 desc.addFamily(new HColumnDescriptor(FAMILYA).setMaxVersions(5));
456 UTIL.getHBaseAdmin().createTable(desc);
457 Table exportTable = new HTable(UTIL.getConfiguration(), desc.getTableName());
458
459 Put p1 = new Put(ROW1);
460 p1.add(FAMILYA, QUAL, now, QUAL);
461 p1.add(FAMILYA, QUAL, now + 1, QUAL);
462 p1.add(FAMILYA, QUAL, now + 2, QUAL);
463 p1.add(FAMILYA, QUAL, now + 3, QUAL);
464 p1.add(FAMILYA, QUAL, now + 4, QUAL);
465
466
467 Put p2 = new Put(ROW2);
468 p2.add(FAMILYA, QUAL, now, QUAL);
469
470 exportTable.put(Arrays.asList(p1, p2));
471
472
473 String[] args = new String[] { EXPORT_TABLE, FQ_OUTPUT_DIR, "1000" };
474 assertTrue(runExport(args));
475
476
477 String IMPORT_TABLE = "importWithFilter";
478 desc = new HTableDescriptor(TableName.valueOf(IMPORT_TABLE));
479 desc.addFamily(new HColumnDescriptor(FAMILYA).setMaxVersions(5));
480 UTIL.getHBaseAdmin().createTable(desc);
481
482 Table importTable = new HTable(UTIL.getConfiguration(), desc.getTableName());
483 args = new String[] { "-D" + Import.FILTER_CLASS_CONF_KEY + "=" + PrefixFilter.class.getName(),
484 "-D" + Import.FILTER_ARGS_CONF_KEY + "=" + Bytes.toString(ROW1), IMPORT_TABLE,
485 FQ_OUTPUT_DIR,
486 "1000" };
487 assertTrue(runImport(args));
488
489
490 PrefixFilter filter = new PrefixFilter(ROW1);
491 int count = getCount(exportTable, filter);
492
493 Assert.assertEquals("Unexpected row count between export and import tables", count,
494 getCount(importTable, null));
495
496
497
498
499 args = new String[] { "-D" + Import.FILTER_CLASS_CONF_KEY + "=" + Filter.class.getName(),
500 "-D" + Import.FILTER_ARGS_CONF_KEY + "=" + Bytes.toString(ROW1) + "", EXPORT_TABLE,
501 FQ_OUTPUT_DIR, "1000" };
502 assertFalse(runImport(args));
503
504
505 exportTable.close();
506 importTable.close();
507 }
508
509
510
511
512
513
514
515
516
517 private int getCount(Table table, Filter filter) throws IOException {
518 Scan scan = new Scan();
519 scan.setFilter(filter);
520 ResultScanner results = table.getScanner(scan);
521 int count = 0;
522 for (Result res : results) {
523 count += res.size();
524 }
525 results.close();
526 return count;
527 }
528
529
530
531
532 @Test
533 public void testImportMain() throws Exception {
534 PrintStream oldPrintStream = System.err;
535 SecurityManager SECURITY_MANAGER = System.getSecurityManager();
536 LauncherSecurityManager newSecurityManager= new LauncherSecurityManager();
537 System.setSecurityManager(newSecurityManager);
538 ByteArrayOutputStream data = new ByteArrayOutputStream();
539 String[] args = {};
540 System.setErr(new PrintStream(data));
541 try {
542 System.setErr(new PrintStream(data));
543 Import.main(args);
544 fail("should be SecurityException");
545 } catch (SecurityException e) {
546 assertEquals(-1, newSecurityManager.getExitCode());
547 assertTrue(data.toString().contains("Wrong number of arguments:"));
548 assertTrue(data.toString().contains("-Dimport.bulk.output=/path/for/output"));
549 assertTrue(data.toString().contains("-Dimport.filter.class=<name of filter class>"));
550 assertTrue(data.toString().contains("-Dimport.bulk.output=/path/for/output"));
551 assertTrue(data.toString().contains("-Dmapreduce.reduce.speculative=false"));
552 } finally {
553 System.setErr(oldPrintStream);
554 System.setSecurityManager(SECURITY_MANAGER);
555 }
556 }
557
558
559
560
561 @Test
562 public void testExportMain() throws Exception {
563 PrintStream oldPrintStream = System.err;
564 SecurityManager SECURITY_MANAGER = System.getSecurityManager();
565 LauncherSecurityManager newSecurityManager= new LauncherSecurityManager();
566 System.setSecurityManager(newSecurityManager);
567 ByteArrayOutputStream data = new ByteArrayOutputStream();
568 String[] args = {};
569 System.setErr(new PrintStream(data));
570 try {
571 System.setErr(new PrintStream(data));
572 Export.main(args);
573 fail("should be SecurityException");
574 } catch (SecurityException e) {
575 assertEquals(-1, newSecurityManager.getExitCode());
576 assertTrue(data.toString().contains("Wrong number of arguments:"));
577 assertTrue(data.toString().contains(
578 "Usage: Export [-D <property=value>]* <tablename> <outputdir> [<versions> " +
579 "[<starttime> [<endtime>]] [^[regex pattern] or [Prefix] to filter]]"));
580 assertTrue(data.toString().contains("-D hbase.mapreduce.scan.column.family=<familyName>"));
581 assertTrue(data.toString().contains("-D hbase.mapreduce.include.deleted.rows=true"));
582 assertTrue(data.toString().contains("-Dhbase.client.scanner.caching=100"));
583 assertTrue(data.toString().contains("-Dmapreduce.map.speculative=false"));
584 assertTrue(data.toString().contains("-Dmapreduce.reduce.speculative=false"));
585 assertTrue(data.toString().contains("-Dhbase.export.scanner.batch=10"));
586 } finally {
587 System.setErr(oldPrintStream);
588 System.setSecurityManager(SECURITY_MANAGER);
589 }
590 }
591
592
593
594
595 @SuppressWarnings({ "unchecked", "rawtypes" })
596 @Test
597 public void testKeyValueImporter() throws Exception {
598 KeyValueImporter importer = new KeyValueImporter();
599 Configuration configuration = new Configuration();
600 Context ctx = mock(Context.class);
601 when(ctx.getConfiguration()).thenReturn(configuration);
602
603 doAnswer(new Answer<Void>() {
604
605 @Override
606 public Void answer(InvocationOnMock invocation) throws Throwable {
607 ImmutableBytesWritable writer = (ImmutableBytesWritable) invocation.getArguments()[0];
608 KeyValue key = (KeyValue) invocation.getArguments()[1];
609 assertEquals("Key", Bytes.toString(writer.get()));
610 assertEquals("row", Bytes.toString(key.getRow()));
611 return null;
612 }
613 }).when(ctx).write(any(ImmutableBytesWritable.class), any(KeyValue.class));
614
615 importer.setup(ctx);
616 Result value = mock(Result.class);
617 KeyValue[] keys = {
618 new KeyValue(Bytes.toBytes("row"), Bytes.toBytes("family"), Bytes.toBytes("qualifier"),
619 Bytes.toBytes("value")),
620 new KeyValue(Bytes.toBytes("row"), Bytes.toBytes("family"), Bytes.toBytes("qualifier"),
621 Bytes.toBytes("value1")) };
622 when(value.rawCells()).thenReturn(keys);
623 importer.map(new ImmutableBytesWritable(Bytes.toBytes("Key")), value, ctx);
624
625 }
626
627
628
629
630
631 @Test
632 public void testAddFilterAndArguments() throws IOException {
633 Configuration configuration = new Configuration();
634
635 List<String> args = new ArrayList<String>();
636 args.add("param1");
637 args.add("param2");
638
639 Import.addFilterAndArguments(configuration, FilterBase.class, args);
640 assertEquals("org.apache.hadoop.hbase.filter.FilterBase",
641 configuration.get(Import.FILTER_CLASS_CONF_KEY));
642 assertEquals("param1,param2", configuration.get(Import.FILTER_ARGS_CONF_KEY));
643 }
644
645 @Test
646 public void testDurability() throws IOException, InterruptedException, ClassNotFoundException {
647
648 String exportTableName = "exporttestDurability";
649 try (Table exportTable = UTIL.createTable(TableName.valueOf(exportTableName), FAMILYA, 3);) {
650
651 Put put = new Put(ROW1);
652 put.add(FAMILYA, QUAL, now, QUAL);
653 put.add(FAMILYA, QUAL, now + 1, QUAL);
654 put.add(FAMILYA, QUAL, now + 2, QUAL);
655 exportTable.put(put);
656
657 put = new Put(ROW2);
658 put.add(FAMILYA, QUAL, now, QUAL);
659 put.add(FAMILYA, QUAL, now + 1, QUAL);
660 put.add(FAMILYA, QUAL, now + 2, QUAL);
661 exportTable.put(put);
662 }
663
664
665 String[] args = new String[] { exportTableName, FQ_OUTPUT_DIR, "1000"};
666 assertTrue(runExport(args));
667
668
669 String importTableName = "importTestDurability1";
670 WAL wal = null;
671 HRegionInfo region = null;
672 TableWALActionListener walListener = null;
673 try (Table importTable =
674 UTIL.createTable(TableName.valueOf(importTableName), FAMILYA, 3);) {
675
676
677 walListener = new TableWALActionListener(importTableName);
678 region = UTIL.getHBaseCluster().getRegionServerThreads().get(0).getRegionServer()
679 .getOnlineRegions(importTable.getName()).get(0).getRegionInfo();
680 wal = UTIL.getMiniHBaseCluster().getRegionServer(0).getWAL(region);
681 wal.registerWALActionsListener(walListener);
682
683
684 args =
685 new String[] { "-D" + Import.WAL_DURABILITY + "=" + Durability.SKIP_WAL.name(),
686 importTableName, FQ_OUTPUT_DIR };
687 assertTrue(runImport(args));
688
689 assertTrue(!walListener.isWALVisited());
690
691 assertTrue(getCount(importTable, null) == 2);
692
693
694 }
695 importTableName = "importTestDurability2";
696 try (Table importTable =
697 UTIL.createTable(TableName.valueOf(importTableName), FAMILYA, 3);) {
698 region = UTIL.getHBaseCluster().getRegionServerThreads().get(0).getRegionServer()
699 .getOnlineRegions(importTable.getName()).get(0).getRegionInfo();
700 wal = UTIL.getMiniHBaseCluster().getRegionServer(0).getWAL(region);
701 walListener = new TableWALActionListener(importTableName);
702 wal.registerWALActionsListener(walListener);
703 args = new String[] { importTableName, FQ_OUTPUT_DIR };
704 assertTrue(runImport(args));
705
706 assertTrue(walListener.isWALVisited());
707
708 assertTrue(getCount(importTable, null) == 2);
709 }
710 }
711
712
713
714
715
716 private static class TableWALActionListener extends WALActionsListener.Base {
717
718 private String tableName;
719 private boolean isVisited = false;
720
721 public TableWALActionListener(String tableName) {
722 this.tableName = tableName;
723 }
724
725 @Override
726 public void visitLogEntryBeforeWrite(HTableDescriptor htd, WALKey logKey, WALEdit logEdit) {
727 if (tableName.equalsIgnoreCase(htd.getNameAsString())) {
728 isVisited = true;
729 }
730 }
731
732 public boolean isWALVisited() {
733 return isVisited;
734 }
735 }
736 }