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