1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.wal;
20
21 import static org.junit.Assert.assertEquals;
22 import static org.junit.Assert.assertFalse;
23 import static org.junit.Assert.assertTrue;
24 import static org.junit.Assert.fail;
25
26 import com.google.common.base.Joiner;
27 import com.google.common.collect.ImmutableList;
28 import com.google.common.collect.ImmutableMap;
29 import com.google.protobuf.ByteString;
30
31 import java.io.FileNotFoundException;
32 import java.io.IOException;
33 import java.lang.reflect.Method;
34 import java.security.PrivilegedExceptionAction;
35 import java.util.ArrayList;
36 import java.util.Arrays;
37 import java.util.Collections;
38 import java.util.HashMap;
39 import java.util.HashSet;
40 import java.util.List;
41 import java.util.Map;
42 import java.util.NavigableSet;
43 import java.util.Set;
44 import java.util.concurrent.atomic.AtomicBoolean;
45 import java.util.concurrent.atomic.AtomicInteger;
46 import java.util.concurrent.atomic.AtomicLong;
47
48 import org.apache.commons.logging.Log;
49 import org.apache.commons.logging.LogFactory;
50 import org.apache.hadoop.conf.Configuration;
51 import org.apache.hadoop.fs.FSDataInputStream;
52 import org.apache.hadoop.fs.FSDataOutputStream;
53 import org.apache.hadoop.fs.FileStatus;
54 import org.apache.hadoop.fs.FileSystem;
55 import org.apache.hadoop.fs.FileUtil;
56 import org.apache.hadoop.fs.Path;
57 import org.apache.hadoop.fs.PathFilter;
58 import org.apache.hadoop.hbase.Cell;
59 import org.apache.hadoop.hbase.HBaseConfiguration;
60 import org.apache.hadoop.hbase.HBaseTestingUtility;
61 import org.apache.hadoop.hbase.HConstants;
62 import org.apache.hadoop.hbase.HRegionInfo;
63 import org.apache.hadoop.hbase.KeyValue;
64 import org.apache.hadoop.hbase.ServerName;
65 import org.apache.hadoop.hbase.TableName;
66 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
67 import org.apache.hadoop.hbase.protobuf.generated.WALProtos;
68 import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode;
69 import org.apache.hadoop.hbase.regionserver.HRegion;
70 import org.apache.hadoop.hbase.regionserver.wal.FaultySequenceFileLogReader;
71 import org.apache.hadoop.hbase.regionserver.wal.InstrumentedLogWriter;
72 import org.apache.hadoop.hbase.regionserver.wal.ProtobufLogReader;
73 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
74 import org.apache.hadoop.hbase.security.User;
75 import org.apache.hadoop.hbase.testclassification.LargeTests;
76 import org.apache.hadoop.hbase.testclassification.RegionServerTests;
77 import org.apache.hadoop.hbase.util.Bytes;
78 import org.apache.hadoop.hbase.util.CancelableProgressable;
79 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
80 import org.apache.hadoop.hbase.util.FSUtils;
81 import org.apache.hadoop.hbase.util.Threads;
82 import org.apache.hadoop.hbase.wal.WAL.Entry;
83 import org.apache.hadoop.hbase.wal.WAL.Reader;
84 import org.apache.hadoop.hbase.wal.WALProvider.Writer;
85 import org.apache.hadoop.hbase.wal.WALSplitter.CorruptedLogFileException;
86 import org.apache.hadoop.hdfs.DFSTestUtil;
87 import org.apache.hadoop.hdfs.server.namenode.LeaseExpiredException;
88 import org.apache.hadoop.ipc.RemoteException;
89 import org.junit.After;
90 import org.junit.AfterClass;
91 import org.junit.Before;
92 import org.junit.BeforeClass;
93 import org.junit.Rule;
94 import org.junit.Test;
95 import org.junit.experimental.categories.Category;
96 import org.junit.rules.TestName;
97 import org.mockito.Mockito;
98 import org.mockito.invocation.InvocationOnMock;
99 import org.mockito.stubbing.Answer;
100
101
102
103
104 @Category({RegionServerTests.class, LargeTests.class})
105 public class TestWALSplit {
106 {
107
108
109
110
111
112 }
113 private final static Log LOG = LogFactory.getLog(TestWALSplit.class);
114
115 private static Configuration conf;
116 private FileSystem fs;
117
118 protected final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
119
120 private Path HBASEDIR;
121 private Path WALDIR;
122 private Path OLDLOGDIR;
123 private Path CORRUPTDIR;
124 private Path TABLEDIR;
125
126 private static final int NUM_WRITERS = 10;
127 private static final int ENTRIES = 10;
128
129 private static final String FILENAME_BEING_SPLIT = "testfile";
130 private static final TableName TABLE_NAME =
131 TableName.valueOf("t1");
132 private static final byte[] FAMILY = "f1".getBytes();
133 private static final byte[] QUALIFIER = "q1".getBytes();
134 private static final byte[] VALUE = "v1".getBytes();
135 private static final String WAL_FILE_PREFIX = "wal.dat.";
136 private static List<String> REGIONS = new ArrayList<String>();
137 private static final String HBASE_SKIP_ERRORS = "hbase.hlog.split.skip.errors";
138 private static String ROBBER;
139 private static String ZOMBIE;
140 private static String [] GROUP = new String [] {"supergroup"};
141 private RecoveryMode mode;
142
143 static enum Corruptions {
144 INSERT_GARBAGE_ON_FIRST_LINE,
145 INSERT_GARBAGE_IN_THE_MIDDLE,
146 APPEND_GARBAGE,
147 TRUNCATE,
148 TRUNCATE_TRAILER
149 }
150
151 @BeforeClass
152 public static void setUpBeforeClass() throws Exception {
153 conf = TEST_UTIL.getConfiguration();
154 conf.setClass("hbase.regionserver.hlog.writer.impl",
155 InstrumentedLogWriter.class, Writer.class);
156
157 System.setProperty("hbase.tests.use.shortcircuit.reads", "false");
158
159 Map<String, String []> u2g_map = new HashMap<String, String []>(2);
160 ROBBER = User.getCurrent().getName() + "-robber";
161 ZOMBIE = User.getCurrent().getName() + "-zombie";
162 u2g_map.put(ROBBER, GROUP);
163 u2g_map.put(ZOMBIE, GROUP);
164 DFSTestUtil.updateConfWithFakeGroupMapping(conf, u2g_map);
165 conf.setInt("dfs.heartbeat.interval", 1);
166 TEST_UTIL.startMiniDFSCluster(2);
167 }
168
169 @AfterClass
170 public static void tearDownAfterClass() throws Exception {
171 TEST_UTIL.shutdownMiniDFSCluster();
172 }
173
174 @Rule
175 public TestName name = new TestName();
176 private WALFactory wals = null;
177
178 @Before
179 public void setUp() throws Exception {
180 LOG.info("Cleaning up cluster for new test.");
181 fs = TEST_UTIL.getDFSCluster().getFileSystem();
182 HBASEDIR = TEST_UTIL.createRootDir();
183 OLDLOGDIR = new Path(HBASEDIR, HConstants.HREGION_OLDLOGDIR_NAME);
184 CORRUPTDIR = new Path(HBASEDIR, HConstants.CORRUPT_DIR_NAME);
185 TABLEDIR = FSUtils.getTableDir(HBASEDIR, TABLE_NAME);
186 REGIONS.clear();
187 Collections.addAll(REGIONS, "bbb", "ccc");
188 InstrumentedLogWriter.activateFailure = false;
189 this.mode = (conf.getBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, false) ?
190 RecoveryMode.LOG_REPLAY : RecoveryMode.LOG_SPLITTING);
191 wals = new WALFactory(conf, null, name.getMethodName());
192 WALDIR = new Path(HBASEDIR, DefaultWALProvider.getWALDirectoryName(name.getMethodName()));
193
194 }
195
196 @After
197 public void tearDown() throws Exception {
198 try {
199 wals.close();
200 } catch(IOException exception) {
201
202 LOG.info("Ignoring an error while closing down our WALFactory. Fine for some tests, but if" +
203 " you see a failure look here.");
204 LOG.debug("exception details", exception);
205 } finally {
206 wals = null;
207 fs.delete(HBASEDIR, true);
208 }
209 }
210
211
212
213
214
215
216
217 @Test (timeout=300000)
218 public void testLogCannotBeWrittenOnceParsed() throws IOException, InterruptedException {
219 final AtomicLong counter = new AtomicLong(0);
220 AtomicBoolean stop = new AtomicBoolean(false);
221
222 final String region = REGIONS.get(0);
223 final int numWriters = 3;
224 Thread zombie = new ZombieLastLogWriterRegionServer(counter, stop, region, numWriters);
225 try {
226 long startCount = counter.get();
227 zombie.start();
228
229 while (startCount == counter.get()) Threads.sleep(1);
230
231 Threads.sleep(1000);
232 final Configuration conf2 = HBaseConfiguration.create(this.conf);
233 final User robber = User.createUserForTesting(conf2, ROBBER, GROUP);
234 int count = robber.runAs(new PrivilegedExceptionAction<Integer>() {
235 @Override
236 public Integer run() throws Exception {
237 StringBuilder ls = new StringBuilder("Contents of WALDIR (").append(WALDIR)
238 .append("):\n");
239 for (FileStatus status : fs.listStatus(WALDIR)) {
240 ls.append("\t").append(status.toString()).append("\n");
241 }
242 LOG.debug(ls);
243 LOG.info("Splitting WALs out from under zombie. Expecting " + numWriters + " files.");
244 WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf2, wals);
245 LOG.info("Finished splitting out from under zombie.");
246 Path[] logfiles = getLogForRegion(HBASEDIR, TABLE_NAME, region);
247 assertEquals("wrong number of split files for region", numWriters, logfiles.length);
248 int count = 0;
249 for (Path logfile: logfiles) {
250 count += countWAL(logfile);
251 }
252 return count;
253 }
254 });
255 LOG.info("zombie=" + counter.get() + ", robber=" + count);
256 assertTrue("The log file could have at most 1 extra log entry, but can't have less. " +
257 "Zombie could write " + counter.get() + " and logfile had only " + count,
258 counter.get() == count || counter.get() + 1 == count);
259 } finally {
260 stop.set(true);
261 zombie.interrupt();
262 Threads.threadDumpingIsAlive(zombie);
263 }
264 }
265
266
267
268
269
270
271
272 class ZombieLastLogWriterRegionServer extends Thread {
273 final AtomicLong editsCount;
274 final AtomicBoolean stop;
275 final int numOfWriters;
276
277
278
279 final String region;
280 final User user;
281
282 public ZombieLastLogWriterRegionServer(AtomicLong counter, AtomicBoolean stop,
283 final String region, final int writers)
284 throws IOException, InterruptedException {
285 super("ZombieLastLogWriterRegionServer");
286 setDaemon(true);
287 this.stop = stop;
288 this.editsCount = counter;
289 this.region = region;
290 this.user = User.createUserForTesting(conf, ZOMBIE, GROUP);
291 numOfWriters = writers;
292 }
293
294 @Override
295 public void run() {
296 try {
297 doWriting();
298 } catch (IOException e) {
299 LOG.warn(getName() + " Writer exiting " + e);
300 } catch (InterruptedException e) {
301 LOG.warn(getName() + " Writer exiting " + e);
302 }
303 }
304
305 private void doWriting() throws IOException, InterruptedException {
306 this.user.runAs(new PrivilegedExceptionAction<Object>() {
307 @Override
308 public Object run() throws Exception {
309
310
311 int walToKeepOpen = numOfWriters - 1;
312
313
314 Writer writer = null;
315 try {
316 writer = generateWALs(numOfWriters, ENTRIES, walToKeepOpen);
317 } catch (IOException e1) {
318 throw new RuntimeException("Failed", e1);
319 }
320
321 editsCount.addAndGet(numOfWriters * ENTRIES);
322 loop(writer);
323
324
325 try {
326 writer.close();
327 fail("Writing closing after parsing should give an error.");
328 } catch (IOException exception) {
329 LOG.debug("ignoring error when closing final writer.", exception);
330 }
331 return null;
332 }
333 });
334 }
335
336 private void loop(final Writer writer) {
337 byte [] regionBytes = Bytes.toBytes(this.region);
338 while (!stop.get()) {
339 try {
340 long seq = appendEntry(writer, TABLE_NAME, regionBytes,
341 ("r" + editsCount.get()).getBytes(), regionBytes, QUALIFIER, VALUE, 0);
342 long count = editsCount.incrementAndGet();
343 LOG.info(getName() + " sync count=" + count + ", seq=" + seq);
344 try {
345 Thread.sleep(1);
346 } catch (InterruptedException e) {
347
348 }
349 } catch (IOException ex) {
350 LOG.error(getName() + " ex " + ex.toString());
351 if (ex instanceof RemoteException) {
352 LOG.error("Juliet: got RemoteException " + ex.getMessage() +
353 " while writing " + (editsCount.get() + 1));
354 } else {
355 LOG.error(getName() + " failed to write....at " + editsCount.get());
356 fail("Failed to write " + editsCount.get());
357 }
358 break;
359 } catch (Throwable t) {
360 LOG.error(getName() + " HOW? " + t);
361 LOG.debug("exception details", t);
362 break;
363 }
364 }
365 LOG.info(getName() + " Writer exiting");
366 }
367 }
368
369
370
371
372
373 @Test (timeout=300000)
374 public void testRecoveredEditsPathForMeta() throws IOException {
375 byte [] encoded = HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes();
376 Path tdir = FSUtils.getTableDir(HBASEDIR, TableName.META_TABLE_NAME);
377 Path regiondir = new Path(tdir,
378 HRegionInfo.FIRST_META_REGIONINFO.getEncodedName());
379 fs.mkdirs(regiondir);
380 long now = System.currentTimeMillis();
381 Entry entry =
382 new Entry(new WALKey(encoded,
383 TableName.META_TABLE_NAME, 1, now, HConstants.DEFAULT_CLUSTER_ID),
384 new WALEdit());
385 Path p = WALSplitter.getRegionSplitEditsPath(fs, entry, HBASEDIR,
386 FILENAME_BEING_SPLIT);
387 String parentOfParent = p.getParent().getParent().getName();
388 assertEquals(parentOfParent, HRegionInfo.FIRST_META_REGIONINFO.getEncodedName());
389 }
390
391
392
393
394
395 @Test (timeout=300000)
396 public void testOldRecoveredEditsFileSidelined() throws IOException {
397 byte [] encoded = HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes();
398 Path tdir = FSUtils.getTableDir(HBASEDIR, TableName.META_TABLE_NAME);
399 Path regiondir = new Path(tdir,
400 HRegionInfo.FIRST_META_REGIONINFO.getEncodedName());
401 fs.mkdirs(regiondir);
402 long now = System.currentTimeMillis();
403 Entry entry =
404 new Entry(new WALKey(encoded,
405 TableName.META_TABLE_NAME, 1, now, HConstants.DEFAULT_CLUSTER_ID),
406 new WALEdit());
407 Path parent = WALSplitter.getRegionDirRecoveredEditsDir(regiondir);
408 assertEquals(parent.getName(), HConstants.RECOVERED_EDITS_DIR);
409 fs.createNewFile(parent);
410
411 Path p = WALSplitter.getRegionSplitEditsPath(fs, entry, HBASEDIR,
412 FILENAME_BEING_SPLIT);
413 String parentOfParent = p.getParent().getParent().getName();
414 assertEquals(parentOfParent, HRegionInfo.FIRST_META_REGIONINFO.getEncodedName());
415 WALFactory.createRecoveredEditsWriter(fs, p, conf).close();
416 }
417
418 private void useDifferentDFSClient() throws IOException {
419
420
421 fs.initialize(fs.getUri(), conf);
422 }
423
424 @Test (timeout=300000)
425 public void testSplitPreservesEdits() throws IOException{
426 final String REGION = "region__1";
427 REGIONS.clear();
428 REGIONS.add(REGION);
429
430 generateWALs(1, 10, -1, 0);
431 useDifferentDFSClient();
432 WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals);
433 Path originalLog = (fs.listStatus(OLDLOGDIR))[0].getPath();
434 Path[] splitLog = getLogForRegion(HBASEDIR, TABLE_NAME, REGION);
435 assertEquals(1, splitLog.length);
436
437 assertTrue("edits differ after split", logsAreEqual(originalLog, splitLog[0]));
438 }
439
440 @Test (timeout=300000)
441 public void testSplitRemovesRegionEventsEdits() throws IOException{
442 final String REGION = "region__1";
443 REGIONS.clear();
444 REGIONS.add(REGION);
445
446 generateWALs(1, 10, -1, 100);
447 useDifferentDFSClient();
448 WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals);
449 Path originalLog = (fs.listStatus(OLDLOGDIR))[0].getPath();
450 Path[] splitLog = getLogForRegion(HBASEDIR, TABLE_NAME, REGION);
451 assertEquals(1, splitLog.length);
452
453 assertFalse("edits differ after split", logsAreEqual(originalLog, splitLog[0]));
454
455 assertEquals(10, countWAL(splitLog[0]));
456 }
457
458
459 @Test (timeout=300000)
460 public void testSplitLeavesCompactionEventsEdits() throws IOException{
461 HRegionInfo hri = new HRegionInfo(TABLE_NAME);
462 REGIONS.clear();
463 REGIONS.add(hri.getEncodedName());
464 Path regionDir = new Path(FSUtils.getTableDir(HBASEDIR, TABLE_NAME), hri.getEncodedName());
465 LOG.info("Creating region directory: " + regionDir);
466 assertTrue(fs.mkdirs(regionDir));
467
468 Writer writer = generateWALs(1, 10, 0, 10);
469 String[] compactInputs = new String[]{"file1", "file2", "file3"};
470 String compactOutput = "file4";
471 appendCompactionEvent(writer, hri, compactInputs, compactOutput);
472 writer.close();
473
474 useDifferentDFSClient();
475 WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals);
476
477 Path originalLog = (fs.listStatus(OLDLOGDIR))[0].getPath();
478
479 assertEquals(21, countWAL(originalLog));
480
481 Path[] splitLog = getLogForRegion(HBASEDIR, TABLE_NAME, hri.getEncodedName());
482 assertEquals(1, splitLog.length);
483
484 assertFalse("edits differ after split", logsAreEqual(originalLog, splitLog[0]));
485
486 assertEquals(11, countWAL(splitLog[0]));
487 }
488
489
490
491
492
493 private int splitAndCount(final int expectedFiles, final int expectedEntries)
494 throws IOException {
495 useDifferentDFSClient();
496 WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals);
497 int result = 0;
498 for (String region : REGIONS) {
499 Path[] logfiles = getLogForRegion(HBASEDIR, TABLE_NAME, region);
500 assertEquals(expectedFiles, logfiles.length);
501 int count = 0;
502 for (Path logfile: logfiles) {
503 count += countWAL(logfile);
504 }
505 if (-1 != expectedEntries) {
506 assertEquals(expectedEntries, count);
507 }
508 result += count;
509 }
510 return result;
511 }
512
513 @Test (timeout=300000)
514 public void testEmptyLogFiles() throws IOException {
515 testEmptyLogFiles(true);
516 }
517
518 @Test (timeout=300000)
519 public void testEmptyOpenLogFiles() throws IOException {
520 testEmptyLogFiles(false);
521 }
522
523 private void testEmptyLogFiles(final boolean close) throws IOException {
524
525
526 fs.mkdirs(WALDIR);
527 injectEmptyFile(".empty", close);
528 generateWALs(Integer.MAX_VALUE);
529 injectEmptyFile("empty", close);
530 splitAndCount(NUM_WRITERS, NUM_WRITERS * ENTRIES);
531 }
532
533 @Test (timeout=300000)
534 public void testOpenZeroLengthReportedFileButWithDataGetsSplit() throws IOException {
535
536 generateWALs(5);
537 splitAndCount(NUM_WRITERS, NUM_WRITERS * ENTRIES);
538 }
539
540 @Test (timeout=300000)
541 public void testTralingGarbageCorruptionFileSkipErrorsPasses() throws IOException {
542 conf.setBoolean(HBASE_SKIP_ERRORS, true);
543 generateWALs(Integer.MAX_VALUE);
544 corruptWAL(new Path(WALDIR, WAL_FILE_PREFIX + "5"),
545 Corruptions.APPEND_GARBAGE, true);
546 splitAndCount(NUM_WRITERS, NUM_WRITERS * ENTRIES);
547 }
548
549 @Test (timeout=300000)
550 public void testFirstLineCorruptionLogFileSkipErrorsPasses() throws IOException {
551 conf.setBoolean(HBASE_SKIP_ERRORS, true);
552 generateWALs(Integer.MAX_VALUE);
553 corruptWAL(new Path(WALDIR, WAL_FILE_PREFIX + "5"),
554 Corruptions.INSERT_GARBAGE_ON_FIRST_LINE, true);
555 splitAndCount(NUM_WRITERS - 1, (NUM_WRITERS - 1) * ENTRIES);
556 }
557
558 @Test (timeout=300000)
559 public void testMiddleGarbageCorruptionSkipErrorsReadsHalfOfFile() throws IOException {
560 conf.setBoolean(HBASE_SKIP_ERRORS, true);
561 generateWALs(Integer.MAX_VALUE);
562 corruptWAL(new Path(WALDIR, WAL_FILE_PREFIX + "5"),
563 Corruptions.INSERT_GARBAGE_IN_THE_MIDDLE, false);
564
565
566
567 int goodEntries = (NUM_WRITERS - 1) * ENTRIES;
568 int firstHalfEntries = (int) Math.ceil(ENTRIES / 2) - 1;
569 int allRegionsCount = splitAndCount(NUM_WRITERS, -1);
570 assertTrue("The file up to the corrupted area hasn't been parsed",
571 REGIONS.size() * (goodEntries + firstHalfEntries) <= allRegionsCount);
572 }
573
574 @Test (timeout=300000)
575 public void testCorruptedFileGetsArchivedIfSkipErrors() throws IOException {
576 conf.setBoolean(HBASE_SKIP_ERRORS, true);
577 for (FaultySequenceFileLogReader.FailureType failureType :
578 FaultySequenceFileLogReader.FailureType.values()) {
579 final Set<String> walDirContents = splitCorruptWALs(failureType);
580 final Set<String> archivedLogs = new HashSet<String>();
581 final StringBuilder archived = new StringBuilder("Archived logs in CORRUPTDIR:");
582 for (FileStatus log : fs.listStatus(CORRUPTDIR)) {
583 archived.append("\n\t").append(log.toString());
584 archivedLogs.add(log.getPath().getName());
585 }
586 LOG.debug(archived.toString());
587 assertEquals(failureType.name() + ": expected to find all of our wals corrupt.",
588 walDirContents, archivedLogs);
589 }
590 }
591
592
593
594
595
596 private Set<String> splitCorruptWALs(final FaultySequenceFileLogReader.FailureType failureType)
597 throws IOException {
598 Class<?> backupClass = conf.getClass("hbase.regionserver.hlog.reader.impl",
599 Reader.class);
600 InstrumentedLogWriter.activateFailure = false;
601
602 try {
603 conf.setClass("hbase.regionserver.hlog.reader.impl",
604 FaultySequenceFileLogReader.class, Reader.class);
605 conf.set("faultysequencefilelogreader.failuretype", failureType.name());
606
607 try {
608 wals.shutdown();
609 } catch (IOException exception) {
610
611 LOG.debug("Ignoring problem closing WALFactory.", exception);
612 }
613 wals.close();
614 try {
615 for (FileStatus log : fs.listStatus(CORRUPTDIR)) {
616 fs.delete(log.getPath(), true);
617 }
618 } catch (FileNotFoundException exception) {
619 LOG.debug("no previous CORRUPTDIR to clean.");
620 }
621
622 wals = new WALFactory(conf, null, name.getMethodName());
623 generateWALs(-1);
624
625 final Set<String> walDirContents = new HashSet<String>();
626 for (FileStatus status : fs.listStatus(WALDIR)) {
627 walDirContents.add(status.getPath().getName());
628 }
629 useDifferentDFSClient();
630 WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals);
631 return walDirContents;
632 } finally {
633 conf.setClass("hbase.regionserver.hlog.reader.impl", backupClass,
634 Reader.class);
635 }
636 }
637
638 @Test (timeout=300000, expected = IOException.class)
639 public void testTrailingGarbageCorruptionLogFileSkipErrorsFalseThrows()
640 throws IOException {
641 conf.setBoolean(HBASE_SKIP_ERRORS, false);
642 splitCorruptWALs(FaultySequenceFileLogReader.FailureType.BEGINNING);
643 }
644
645 @Test (timeout=300000)
646 public void testCorruptedLogFilesSkipErrorsFalseDoesNotTouchLogs()
647 throws IOException {
648 conf.setBoolean(HBASE_SKIP_ERRORS, false);
649 try {
650 splitCorruptWALs(FaultySequenceFileLogReader.FailureType.BEGINNING);
651 } catch (IOException e) {
652 LOG.debug("split with 'skip errors' set to 'false' correctly threw");
653 }
654 assertEquals("if skip.errors is false all files should remain in place",
655 NUM_WRITERS, fs.listStatus(WALDIR).length);
656 }
657
658 private void ignoreCorruption(final Corruptions corruption, final int entryCount,
659 final int expectedCount) throws IOException {
660 conf.setBoolean(HBASE_SKIP_ERRORS, false);
661
662 final String REGION = "region__1";
663 REGIONS.clear();
664 REGIONS.add(REGION);
665
666 Path c1 = new Path(WALDIR, WAL_FILE_PREFIX + "0");
667 generateWALs(1, entryCount, -1, 0);
668 corruptWAL(c1, corruption, true);
669
670 useDifferentDFSClient();
671 WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals);
672
673 Path[] splitLog = getLogForRegion(HBASEDIR, TABLE_NAME, REGION);
674 assertEquals(1, splitLog.length);
675
676 int actualCount = 0;
677 Reader in = wals.createReader(fs, splitLog[0]);
678 @SuppressWarnings("unused")
679 Entry entry;
680 while ((entry = in.next()) != null) ++actualCount;
681 assertEquals(expectedCount, actualCount);
682 in.close();
683
684
685 FileStatus[] archivedLogs = fs.listStatus(CORRUPTDIR);
686 assertEquals(archivedLogs.length, 0);
687
688 }
689
690 @Test (timeout=300000)
691 public void testEOFisIgnored() throws IOException {
692 int entryCount = 10;
693 ignoreCorruption(Corruptions.TRUNCATE, entryCount, entryCount-1);
694 }
695
696 @Test (timeout=300000)
697 public void testCorruptWALTrailer() throws IOException {
698 int entryCount = 10;
699 ignoreCorruption(Corruptions.TRUNCATE_TRAILER, entryCount, entryCount);
700 }
701
702 @Test (timeout=300000)
703 public void testLogsGetArchivedAfterSplit() throws IOException {
704 conf.setBoolean(HBASE_SKIP_ERRORS, false);
705 generateWALs(-1);
706 useDifferentDFSClient();
707 WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals);
708 FileStatus[] archivedLogs = fs.listStatus(OLDLOGDIR);
709 assertEquals("wrong number of files in the archive log", NUM_WRITERS, archivedLogs.length);
710 }
711
712 @Test (timeout=300000)
713 public void testSplit() throws IOException {
714 generateWALs(-1);
715 splitAndCount(NUM_WRITERS, NUM_WRITERS * ENTRIES);
716 }
717
718 @Test (timeout=300000)
719 public void testLogDirectoryShouldBeDeletedAfterSuccessfulSplit()
720 throws IOException {
721 generateWALs(-1);
722 useDifferentDFSClient();
723 WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals);
724 FileStatus [] statuses = null;
725 try {
726 statuses = fs.listStatus(WALDIR);
727 if (statuses != null) {
728 fail("Files left in log dir: " +
729 Joiner.on(",").join(FileUtil.stat2Paths(statuses)));
730 }
731 } catch (FileNotFoundException e) {
732
733 }
734 }
735
736 @Test(timeout=300000, expected = IOException.class)
737 public void testSplitWillFailIfWritingToRegionFails() throws Exception {
738
739 Writer writer = generateWALs(4);
740 useDifferentDFSClient();
741
742 String region = "break";
743 Path regiondir = new Path(TABLEDIR, region);
744 fs.mkdirs(regiondir);
745
746 InstrumentedLogWriter.activateFailure = false;
747 appendEntry(writer, TABLE_NAME, Bytes.toBytes(region),
748 ("r" + 999).getBytes(), FAMILY, QUALIFIER, VALUE, 0);
749 writer.close();
750
751 try {
752 InstrumentedLogWriter.activateFailure = true;
753 WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals);
754 } catch (IOException e) {
755 assertTrue(e.getMessage().
756 contains("This exception is instrumented and should only be thrown for testing"));
757 throw e;
758 } finally {
759 InstrumentedLogWriter.activateFailure = false;
760 }
761 }
762
763 @Test (timeout=300000)
764 public void testSplitDeletedRegion() throws IOException {
765 REGIONS.clear();
766 String region = "region_that_splits";
767 REGIONS.add(region);
768
769 generateWALs(1);
770 useDifferentDFSClient();
771
772 Path regiondir = new Path(TABLEDIR, region);
773 fs.delete(regiondir, true);
774 WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals);
775 assertFalse(fs.exists(regiondir));
776 }
777
778 @Test (timeout=300000)
779 public void testIOEOnOutputThread() throws Exception {
780 conf.setBoolean(HBASE_SKIP_ERRORS, false);
781
782 generateWALs(-1);
783 useDifferentDFSClient();
784 FileStatus[] logfiles = fs.listStatus(WALDIR);
785 assertTrue("There should be some log file",
786 logfiles != null && logfiles.length > 0);
787
788
789
790 int largestLogFile = 0;
791 long largestSize = 0;
792 for (int i = 0; i < logfiles.length; i++) {
793 if (logfiles[i].getLen() > largestSize) {
794 largestLogFile = i;
795 largestSize = logfiles[i].getLen();
796 }
797 }
798 assertTrue("There should be some log greater than size 0.", 0 < largestSize);
799
800 WALSplitter logSplitter = new WALSplitter(wals,
801 conf, HBASEDIR, fs, null, null, this.mode) {
802 @Override
803 protected Writer createWriter(Path logfile) throws IOException {
804 Writer mockWriter = Mockito.mock(Writer.class);
805 Mockito.doThrow(new IOException("Injected")).when(
806 mockWriter).append(Mockito.<Entry>any());
807 return mockWriter;
808 }
809 };
810
811
812 final AtomicBoolean stop = new AtomicBoolean(false);
813 final Thread someOldThread = new Thread("Some-old-thread") {
814 @Override
815 public void run() {
816 while(!stop.get()) Threads.sleep(10);
817 }
818 };
819 someOldThread.setDaemon(true);
820 someOldThread.start();
821 final Thread t = new Thread("Background-thread-dumper") {
822 public void run() {
823 try {
824 Threads.threadDumpingIsAlive(someOldThread);
825 } catch (InterruptedException e) {
826 e.printStackTrace();
827 }
828 }
829 };
830 t.setDaemon(true);
831 t.start();
832 try {
833 logSplitter.splitLogFile(logfiles[largestLogFile], null);
834 fail("Didn't throw!");
835 } catch (IOException ioe) {
836 assertTrue(ioe.toString().contains("Injected"));
837 } finally {
838
839 stop.set(true);
840 }
841 }
842
843
844
845
846 private void retryOverHdfsProblem(final FileSystem spiedFs) throws Exception {
847 generateWALs(-1);
848 useDifferentDFSClient();
849
850 try {
851 WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, spiedFs, conf, wals);
852 assertEquals(NUM_WRITERS, fs.listStatus(OLDLOGDIR).length);
853 assertFalse(fs.exists(WALDIR));
854 } catch (IOException e) {
855 fail("There shouldn't be any exception but: " + e.toString());
856 }
857 }
858
859
860 @Test (timeout=300000)
861 public void testMovedWALDuringRecovery() throws Exception {
862
863
864 FileSystem spiedFs = Mockito.spy(fs);
865
866
867 Mockito.doThrow(new LeaseExpiredException("Injected: File does not exist")).
868 when(spiedFs).append(Mockito.<Path>any());
869 retryOverHdfsProblem(spiedFs);
870 }
871
872 @Test (timeout=300000)
873 public void testRetryOpenDuringRecovery() throws Exception {
874 FileSystem spiedFs = Mockito.spy(fs);
875
876
877
878
879
880
881
882
883
884 Mockito.doAnswer(new Answer<FSDataInputStream>() {
885 private final String[] errors = new String[] {
886 "Cannot obtain block length", "Could not obtain the last block",
887 "Blocklist for " + OLDLOGDIR + " has changed"};
888 private int count = 0;
889
890 public FSDataInputStream answer(InvocationOnMock invocation) throws Throwable {
891 if (count < 3) {
892 throw new IOException(errors[count++]);
893 }
894 return (FSDataInputStream)invocation.callRealMethod();
895 }
896 }).when(spiedFs).open(Mockito.<Path>any(), Mockito.anyInt());
897 retryOverHdfsProblem(spiedFs);
898 }
899
900 @Test (timeout=300000)
901 public void testTerminationAskedByReporter() throws IOException, CorruptedLogFileException {
902 generateWALs(1, 10, -1);
903 FileStatus logfile = fs.listStatus(WALDIR)[0];
904 useDifferentDFSClient();
905
906 final AtomicInteger count = new AtomicInteger();
907
908 CancelableProgressable localReporter
909 = new CancelableProgressable() {
910 @Override
911 public boolean progress() {
912 count.getAndIncrement();
913 return false;
914 }
915 };
916
917 FileSystem spiedFs = Mockito.spy(fs);
918 Mockito.doAnswer(new Answer<FSDataInputStream>() {
919 public FSDataInputStream answer(InvocationOnMock invocation) throws Throwable {
920 Thread.sleep(1500);
921 return (FSDataInputStream)invocation.callRealMethod();
922 }
923 }).when(spiedFs).open(Mockito.<Path>any(), Mockito.anyInt());
924
925 try {
926 conf.setInt("hbase.splitlog.report.period", 1000);
927 boolean ret = WALSplitter.splitLogFile(
928 HBASEDIR, logfile, spiedFs, conf, localReporter, null, null, this.mode, wals);
929 assertFalse("Log splitting should failed", ret);
930 assertTrue(count.get() > 0);
931 } catch (IOException e) {
932 fail("There shouldn't be any exception but: " + e.toString());
933 } finally {
934
935 conf.setInt("hbase.splitlog.report.period", 59000);
936 }
937 }
938
939
940
941
942
943 @Test (timeout=300000)
944 public void testThreading() throws Exception {
945 doTestThreading(20000, 128*1024*1024, 0);
946 }
947
948
949
950
951
952 @Test (timeout=300000)
953 public void testThreadingSlowWriterSmallBuffer() throws Exception {
954 doTestThreading(200, 1024, 50);
955 }
956
957
958
959
960
961
962
963
964
965
966
967
968
969 private void doTestThreading(final int numFakeEdits,
970 final int bufferSize,
971 final int writerSlowness) throws Exception {
972
973 Configuration localConf = new Configuration(conf);
974 localConf.setInt("hbase.regionserver.hlog.splitlog.buffersize", bufferSize);
975
976
977 Path logPath = new Path(WALDIR, WAL_FILE_PREFIX + ".fake");
978 FSDataOutputStream out = fs.create(logPath);
979 out.close();
980
981
982 final List<String> regions = ImmutableList.of("r0", "r1", "r2", "r3", "r4");
983 makeRegionDirs(regions);
984
985
986 WALSplitter logSplitter = new WALSplitter(wals,
987 localConf, HBASEDIR, fs, null, null, this.mode) {
988
989
990 @Override
991 protected Writer createWriter(Path logfile) throws IOException {
992 Writer mockWriter = Mockito.mock(Writer.class);
993 Mockito.doAnswer(new Answer<Void>() {
994 int expectedIndex = 0;
995
996 @Override
997 public Void answer(InvocationOnMock invocation) {
998 if (writerSlowness > 0) {
999 try {
1000 Thread.sleep(writerSlowness);
1001 } catch (InterruptedException ie) {
1002 Thread.currentThread().interrupt();
1003 }
1004 }
1005 Entry entry = (Entry) invocation.getArguments()[0];
1006 WALEdit edit = entry.getEdit();
1007 List<Cell> cells = edit.getCells();
1008 assertEquals(1, cells.size());
1009 Cell cell = cells.get(0);
1010
1011
1012 assertEquals(expectedIndex, Bytes.toInt(cell.getRowArray(), cell.getRowOffset(),
1013 cell.getRowLength()));
1014 expectedIndex++;
1015 return null;
1016 }
1017 }).when(mockWriter).append(Mockito.<Entry>any());
1018 return mockWriter;
1019 }
1020
1021
1022 @Override
1023 protected Reader getReader(Path curLogFile, CancelableProgressable reporter)
1024 throws IOException {
1025 Reader mockReader = Mockito.mock(Reader.class);
1026 Mockito.doAnswer(new Answer<Entry>() {
1027 int index = 0;
1028
1029 @Override
1030 public Entry answer(InvocationOnMock invocation) throws Throwable {
1031 if (index >= numFakeEdits) return null;
1032
1033
1034 int regionIdx = index % regions.size();
1035 byte region[] = new byte[] {(byte)'r', (byte) (0x30 + regionIdx)};
1036
1037 Entry ret = createTestEntry(TABLE_NAME, region,
1038 Bytes.toBytes((int)(index / regions.size())),
1039 FAMILY, QUALIFIER, VALUE, index);
1040 index++;
1041 return ret;
1042 }
1043 }).when(mockReader).next();
1044 return mockReader;
1045 }
1046 };
1047
1048 logSplitter.splitLogFile(fs.getFileStatus(logPath), null);
1049
1050
1051 Map<byte[], Long> outputCounts = logSplitter.outputSink.getOutputCounts();
1052 for (Map.Entry<byte[], Long> entry : outputCounts.entrySet()) {
1053 LOG.info("Got " + entry.getValue() + " output edits for region " +
1054 Bytes.toString(entry.getKey()));
1055 assertEquals((long)entry.getValue(), numFakeEdits / regions.size());
1056 }
1057 assertEquals("Should have as many outputs as regions", regions.size(), outputCounts.size());
1058 }
1059
1060
1061 @Test (timeout=300000)
1062 public void testSplitLogFileDeletedRegionDir() throws IOException {
1063 LOG.info("testSplitLogFileDeletedRegionDir");
1064 final String REGION = "region__1";
1065 REGIONS.clear();
1066 REGIONS.add(REGION);
1067
1068 generateWALs(1, 10, -1);
1069 useDifferentDFSClient();
1070
1071 Path regiondir = new Path(TABLEDIR, REGION);
1072 LOG.info("Region directory is" + regiondir);
1073 fs.delete(regiondir, true);
1074 WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals);
1075 assertFalse(fs.exists(regiondir));
1076 }
1077
1078 @Test (timeout=300000)
1079 public void testSplitLogFileEmpty() throws IOException {
1080 LOG.info("testSplitLogFileEmpty");
1081
1082
1083 fs.mkdirs(WALDIR);
1084 injectEmptyFile(".empty", true);
1085 useDifferentDFSClient();
1086
1087 WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals);
1088 Path tdir = FSUtils.getTableDir(HBASEDIR, TABLE_NAME);
1089 assertFalse(fs.exists(tdir));
1090
1091 assertEquals(0, countWAL(fs.listStatus(OLDLOGDIR)[0].getPath()));
1092 }
1093
1094 @Test (timeout=300000)
1095 public void testSplitLogFileMultipleRegions() throws IOException {
1096 LOG.info("testSplitLogFileMultipleRegions");
1097 generateWALs(1, 10, -1);
1098 splitAndCount(1, 10);
1099 }
1100
1101 @Test (timeout=300000)
1102 public void testSplitLogFileFirstLineCorruptionLog()
1103 throws IOException {
1104 conf.setBoolean(HBASE_SKIP_ERRORS, true);
1105 generateWALs(1, 10, -1);
1106 FileStatus logfile = fs.listStatus(WALDIR)[0];
1107
1108 corruptWAL(logfile.getPath(),
1109 Corruptions.INSERT_GARBAGE_ON_FIRST_LINE, true);
1110
1111 useDifferentDFSClient();
1112 WALSplitter.split(HBASEDIR, WALDIR, OLDLOGDIR, fs, conf, wals);
1113
1114 final Path corruptDir = new Path(FSUtils.getRootDir(conf), HConstants.CORRUPT_DIR_NAME);
1115 assertEquals(1, fs.listStatus(corruptDir).length);
1116 }
1117
1118
1119
1120
1121
1122 @Test (timeout=300000)
1123 public void testConcurrentSplitLogAndReplayRecoverEdit() throws IOException {
1124 LOG.info("testConcurrentSplitLogAndReplayRecoverEdit");
1125
1126 String regionName = "r0";
1127 final Path regiondir = new Path(TABLEDIR, regionName);
1128 REGIONS.clear();
1129 REGIONS.add(regionName);
1130 generateWALs(-1);
1131
1132 wals.getWAL(Bytes.toBytes(regionName), null);
1133 FileStatus[] logfiles = fs.listStatus(WALDIR);
1134 assertTrue("There should be some log file",
1135 logfiles != null && logfiles.length > 0);
1136
1137 WALSplitter logSplitter = new WALSplitter(wals,
1138 conf, HBASEDIR, fs, null, null, this.mode) {
1139 @Override
1140 protected Writer createWriter(Path logfile)
1141 throws IOException {
1142 Writer writer = wals.createRecoveredEditsWriter(this.fs, logfile);
1143
1144
1145
1146 NavigableSet<Path> files = WALSplitter.getSplitEditFilesSorted(fs, regiondir);
1147 if (files != null && !files.isEmpty()) {
1148 for (Path file : files) {
1149 if (!this.fs.delete(file, false)) {
1150 LOG.error("Failed delete of " + file);
1151 } else {
1152 LOG.debug("Deleted recovered.edits file=" + file);
1153 }
1154 }
1155 }
1156 return writer;
1157 }
1158 };
1159 try{
1160 logSplitter.splitLogFile(logfiles[0], null);
1161 } catch (IOException e) {
1162 LOG.info(e);
1163 fail("Throws IOException when spliting "
1164 + "log, it is most likely because writing file does not "
1165 + "exist which is caused by concurrent replayRecoveredEditsIfAny()");
1166 }
1167 if (fs.exists(CORRUPTDIR)) {
1168 if (fs.listStatus(CORRUPTDIR).length > 0) {
1169 fail("There are some corrupt logs, "
1170 + "it is most likely caused by concurrent replayRecoveredEditsIfAny()");
1171 }
1172 }
1173 }
1174
1175 private Writer generateWALs(int leaveOpen) throws IOException {
1176 return generateWALs(NUM_WRITERS, ENTRIES, leaveOpen, 0);
1177 }
1178
1179 private Writer generateWALs(int writers, int entries, int leaveOpen) throws IOException {
1180 return generateWALs(writers, entries, leaveOpen, 7);
1181 }
1182
1183 private void makeRegionDirs(List<String> regions) throws IOException {
1184 for (String region : regions) {
1185 LOG.debug("Creating dir for region " + region);
1186 fs.mkdirs(new Path(TABLEDIR, region));
1187 }
1188 }
1189
1190
1191
1192
1193
1194 private Writer generateWALs(int writers, int entries, int leaveOpen, int regionEvents) throws IOException {
1195 makeRegionDirs(REGIONS);
1196 fs.mkdirs(WALDIR);
1197 Writer [] ws = new Writer[writers];
1198 int seq = 0;
1199 int numRegionEventsAdded = 0;
1200 for (int i = 0; i < writers; i++) {
1201 ws[i] = wals.createWALWriter(fs, new Path(WALDIR, WAL_FILE_PREFIX + i));
1202 for (int j = 0; j < entries; j++) {
1203 int prefix = 0;
1204 for (String region : REGIONS) {
1205 String row_key = region + prefix++ + i + j;
1206 appendEntry(ws[i], TABLE_NAME, region.getBytes(), row_key.getBytes(), FAMILY, QUALIFIER,
1207 VALUE, seq++);
1208
1209 if (numRegionEventsAdded < regionEvents) {
1210 numRegionEventsAdded ++;
1211 appendRegionEvent(ws[i], region);
1212 }
1213 }
1214 }
1215 if (i != leaveOpen) {
1216 ws[i].close();
1217 LOG.info("Closing writer " + i);
1218 }
1219 }
1220 if (leaveOpen < 0 || leaveOpen >= writers) {
1221 return null;
1222 }
1223 return ws[leaveOpen];
1224 }
1225
1226
1227
1228 private Path[] getLogForRegion(Path rootdir, TableName table, String region)
1229 throws IOException {
1230 Path tdir = FSUtils.getTableDir(rootdir, table);
1231 @SuppressWarnings("deprecation")
1232 Path editsdir = WALSplitter.getRegionDirRecoveredEditsDir(HRegion.getRegionDir(tdir,
1233 Bytes.toString(region.getBytes())));
1234 FileStatus[] files = fs.listStatus(editsdir, new PathFilter() {
1235 @Override
1236 public boolean accept(Path p) {
1237 if (WALSplitter.isSequenceIdFile(p)) {
1238 return false;
1239 }
1240 return true;
1241 }
1242 });
1243 Path[] paths = new Path[files.length];
1244 for (int i = 0; i < files.length; i++) {
1245 paths[i] = files[i].getPath();
1246 }
1247 return paths;
1248 }
1249
1250 private void corruptWAL(Path path, Corruptions corruption, boolean close) throws IOException {
1251 FSDataOutputStream out;
1252 int fileSize = (int) fs.listStatus(path)[0].getLen();
1253
1254 FSDataInputStream in = fs.open(path);
1255 byte[] corrupted_bytes = new byte[fileSize];
1256 in.readFully(0, corrupted_bytes, 0, fileSize);
1257 in.close();
1258
1259 switch (corruption) {
1260 case APPEND_GARBAGE:
1261 fs.delete(path, false);
1262 out = fs.create(path);
1263 out.write(corrupted_bytes);
1264 out.write("-----".getBytes());
1265 closeOrFlush(close, out);
1266 break;
1267
1268 case INSERT_GARBAGE_ON_FIRST_LINE:
1269 fs.delete(path, false);
1270 out = fs.create(path);
1271 out.write(0);
1272 out.write(corrupted_bytes);
1273 closeOrFlush(close, out);
1274 break;
1275
1276 case INSERT_GARBAGE_IN_THE_MIDDLE:
1277 fs.delete(path, false);
1278 out = fs.create(path);
1279 int middle = (int) Math.floor(corrupted_bytes.length / 2);
1280 out.write(corrupted_bytes, 0, middle);
1281 out.write(0);
1282 out.write(corrupted_bytes, middle, corrupted_bytes.length - middle);
1283 closeOrFlush(close, out);
1284 break;
1285
1286 case TRUNCATE:
1287 fs.delete(path, false);
1288 out = fs.create(path);
1289 out.write(corrupted_bytes, 0, fileSize
1290 - (32 + ProtobufLogReader.PB_WAL_COMPLETE_MAGIC.length + Bytes.SIZEOF_INT));
1291 closeOrFlush(close, out);
1292 break;
1293
1294 case TRUNCATE_TRAILER:
1295 fs.delete(path, false);
1296 out = fs.create(path);
1297 out.write(corrupted_bytes, 0, fileSize - Bytes.SIZEOF_INT);
1298 closeOrFlush(close, out);
1299 break;
1300 }
1301 }
1302
1303 private void closeOrFlush(boolean close, FSDataOutputStream out)
1304 throws IOException {
1305 if (close) {
1306 out.close();
1307 } else {
1308 Method syncMethod = null;
1309 try {
1310 syncMethod = out.getClass().getMethod("hflush", new Class<?> []{});
1311 } catch (NoSuchMethodException e) {
1312 try {
1313 syncMethod = out.getClass().getMethod("sync", new Class<?> []{});
1314 } catch (NoSuchMethodException ex) {
1315 throw new IOException("This version of Hadoop supports " +
1316 "neither Syncable.sync() nor Syncable.hflush().");
1317 }
1318 }
1319 try {
1320 syncMethod.invoke(out, new Object[]{});
1321 } catch (Exception e) {
1322 throw new IOException(e);
1323 }
1324
1325 }
1326 }
1327
1328 private int countWAL(Path log) throws IOException {
1329 int count = 0;
1330 Reader in = wals.createReader(fs, log);
1331 while (in.next() != null) {
1332 count++;
1333 }
1334 in.close();
1335 return count;
1336 }
1337
1338 private static void appendCompactionEvent(Writer w, HRegionInfo hri, String[] inputs,
1339 String output) throws IOException {
1340 WALProtos.CompactionDescriptor.Builder desc = WALProtos.CompactionDescriptor.newBuilder();
1341 desc.setTableName(ByteString.copyFrom(hri.getTable().toBytes()))
1342 .setEncodedRegionName(ByteString.copyFrom(hri.getEncodedNameAsBytes()))
1343 .setRegionName(ByteString.copyFrom(hri.getRegionName()))
1344 .setFamilyName(ByteString.copyFrom(FAMILY))
1345 .setStoreHomeDir(hri.getEncodedName() + "/" + Bytes.toString(FAMILY))
1346 .addAllCompactionInput(Arrays.asList(inputs))
1347 .addCompactionOutput(output);
1348
1349 WALEdit edit = WALEdit.createCompaction(hri, desc.build());
1350 WALKey key = new WALKey(hri.getEncodedNameAsBytes(), TABLE_NAME, 1,
1351 EnvironmentEdgeManager.currentTime(), HConstants.DEFAULT_CLUSTER_ID);
1352 w.append(new Entry(key, edit));
1353 w.sync();
1354 }
1355
1356 private static void appendRegionEvent(Writer w, String region) throws IOException {
1357 WALProtos.RegionEventDescriptor regionOpenDesc = ProtobufUtil.toRegionEventDescriptor(
1358 WALProtos.RegionEventDescriptor.EventType.REGION_OPEN,
1359 TABLE_NAME.toBytes(),
1360 region.getBytes(),
1361 String.valueOf(region.hashCode()).getBytes(),
1362 1,
1363 ServerName.parseServerName("ServerName:9099"), ImmutableMap.<byte[], List<Path>>of());
1364 final long time = EnvironmentEdgeManager.currentTime();
1365 KeyValue kv = new KeyValue(region.getBytes(), WALEdit.METAFAMILY, WALEdit.REGION_EVENT,
1366 time, regionOpenDesc.toByteArray());
1367 final WALKey walKey = new WALKey(region.getBytes(), TABLE_NAME, 1, time,
1368 HConstants.DEFAULT_CLUSTER_ID);
1369 w.append(
1370 new Entry(walKey, new WALEdit().add(kv)));
1371 w.sync();
1372 }
1373
1374 public static long appendEntry(Writer writer, TableName table, byte[] region,
1375 byte[] row, byte[] family, byte[] qualifier,
1376 byte[] value, long seq)
1377 throws IOException {
1378 LOG.info(Thread.currentThread().getName() + " append");
1379 writer.append(createTestEntry(table, region, row, family, qualifier, value, seq));
1380 LOG.info(Thread.currentThread().getName() + " sync");
1381 writer.sync();
1382 return seq;
1383 }
1384
1385 private static Entry createTestEntry(
1386 TableName table, byte[] region,
1387 byte[] row, byte[] family, byte[] qualifier,
1388 byte[] value, long seq) {
1389 long time = System.nanoTime();
1390
1391 seq++;
1392 final KeyValue cell = new KeyValue(row, family, qualifier, time, KeyValue.Type.Put, value);
1393 WALEdit edit = new WALEdit();
1394 edit.add(cell);
1395 return new Entry(new WALKey(region, table, seq, time,
1396 HConstants.DEFAULT_CLUSTER_ID), edit);
1397 }
1398
1399 private void injectEmptyFile(String suffix, boolean closeFile)
1400 throws IOException {
1401 Writer writer = wals.createWALWriter(fs, new Path(WALDIR, WAL_FILE_PREFIX + suffix),
1402 conf);
1403 if (closeFile) writer.close();
1404 }
1405
1406 private boolean logsAreEqual(Path p1, Path p2) throws IOException {
1407 Reader in1, in2;
1408 in1 = wals.createReader(fs, p1);
1409 in2 = wals.createReader(fs, p2);
1410 Entry entry1;
1411 Entry entry2;
1412 while ((entry1 = in1.next()) != null) {
1413 entry2 = in2.next();
1414 if ((entry1.getKey().compareTo(entry2.getKey()) != 0) ||
1415 (!entry1.getEdit().toString().equals(entry2.getEdit().toString()))) {
1416 return false;
1417 }
1418 }
1419 in1.close();
1420 in2.close();
1421 return true;
1422 }
1423 }