1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.procedure2.store.wal;
20
21 import java.io.FileNotFoundException;
22 import java.io.IOException;
23 import java.io.InputStream;
24 import java.io.OutputStream;
25 import java.util.Arrays;
26 import java.util.Comparator;
27 import java.util.Iterator;
28 import java.util.HashSet;
29 import java.util.Set;
30
31 import org.apache.commons.logging.Log;
32 import org.apache.commons.logging.LogFactory;
33 import org.apache.hadoop.fs.FileSystem;
34 import org.apache.hadoop.fs.FileStatus;
35 import org.apache.hadoop.fs.Path;
36 import org.apache.hadoop.hbase.HBaseCommonTestingUtility;
37 import org.apache.hadoop.hbase.procedure2.Procedure;
38 import org.apache.hadoop.hbase.procedure2.ProcedureTestingUtility;
39 import org.apache.hadoop.hbase.procedure2.ProcedureTestingUtility.LoadCounter;
40 import org.apache.hadoop.hbase.procedure2.ProcedureTestingUtility.TestProcedure;
41 import org.apache.hadoop.hbase.procedure2.SequentialProcedure;
42 import org.apache.hadoop.hbase.procedure2.store.ProcedureStore;
43 import org.apache.hadoop.hbase.procedure2.store.ProcedureStore.ProcedureIterator;
44 import org.apache.hadoop.hbase.testclassification.SmallTests;
45 import org.apache.hadoop.hbase.util.Bytes;
46 import org.apache.hadoop.io.IOUtils;
47
48 import org.junit.After;
49 import org.junit.Before;
50 import org.junit.Assert;
51 import org.junit.Test;
52 import org.junit.experimental.categories.Category;
53
54 import static org.junit.Assert.assertEquals;
55 import static org.junit.Assert.assertFalse;
56 import static org.junit.Assert.assertTrue;
57 import static org.junit.Assert.fail;
58
59 @Category(SmallTests.class)
60 public class TestWALProcedureStore {
61 private static final Log LOG = LogFactory.getLog(TestWALProcedureStore.class);
62
63 private static final int PROCEDURE_STORE_SLOTS = 1;
64 private static final Procedure NULL_PROC = null;
65
66 private WALProcedureStore procStore;
67
68 private HBaseCommonTestingUtility htu;
69 private FileSystem fs;
70 private Path testDir;
71 private Path logDir;
72
73 @Before
74 public void setUp() throws IOException {
75 htu = new HBaseCommonTestingUtility();
76 testDir = htu.getDataTestDir();
77 fs = testDir.getFileSystem(htu.getConfiguration());
78 assertTrue(testDir.depth() > 1);
79
80 logDir = new Path(testDir, "proc-logs");
81 procStore = ProcedureTestingUtility.createWalStore(htu.getConfiguration(), fs, logDir);
82 procStore.start(PROCEDURE_STORE_SLOTS);
83 procStore.recoverLease();
84 procStore.load(new LoadCounter());
85 }
86
87 @After
88 public void tearDown() throws IOException {
89 procStore.stop(false);
90 fs.delete(logDir, true);
91 }
92
93 private void storeRestart(ProcedureStore.ProcedureLoader loader) throws Exception {
94 procStore.stop(false);
95 procStore.start(PROCEDURE_STORE_SLOTS);
96 procStore.recoverLease();
97 procStore.load(loader);
98 }
99
100 @Test
101 public void testEmptyRoll() throws Exception {
102 for (int i = 0; i < 10; ++i) {
103 procStore.periodicRollForTesting();
104 }
105 FileStatus[] status = fs.listStatus(logDir);
106 assertEquals(1, status.length);
107 }
108
109 @Test
110 public void testEmptyLogLoad() throws Exception {
111 LoadCounter loader = new LoadCounter();
112 storeRestart(loader);
113 assertEquals(0, loader.getMaxProcId());
114 assertEquals(0, loader.getLoadedCount());
115 assertEquals(0, loader.getCorruptedCount());
116 }
117
118 @Test
119 public void testLoad() throws Exception {
120 Set<Long> procIds = new HashSet<>();
121
122
123 Procedure proc1 = new TestSequentialProcedure();
124 procIds.add(proc1.getProcId());
125 procStore.insert(proc1, null);
126
127 Procedure proc2 = new TestSequentialProcedure();
128 Procedure[] child2 = new Procedure[2];
129 child2[0] = new TestSequentialProcedure();
130 child2[1] = new TestSequentialProcedure();
131
132 procIds.add(proc2.getProcId());
133 procIds.add(child2[0].getProcId());
134 procIds.add(child2[1].getProcId());
135 procStore.insert(proc2, child2);
136
137
138 verifyProcIdsOnRestart(procIds);
139
140
141 procStore.update(proc1);
142 procStore.update(child2[1]);
143 procStore.delete(child2[1].getProcId());
144 procIds.remove(child2[1].getProcId());
145
146
147 verifyProcIdsOnRestart(procIds);
148
149
150 procStore.stop(false);
151 FileStatus[] logs = fs.listStatus(logDir);
152 assertEquals(3, logs.length);
153 for (int i = 0; i < logs.length; ++i) {
154 corruptLog(logs[i], 4);
155 }
156 verifyProcIdsOnRestart(procIds);
157 }
158
159 @Test
160 public void testNoTrailerDoubleRestart() throws Exception {
161
162 Procedure proc0 = new TestSequentialProcedure();
163 procStore.insert(proc0, null);
164 Procedure proc1 = new TestSequentialProcedure();
165 procStore.insert(proc1, null);
166 Procedure proc2 = new TestSequentialProcedure();
167 procStore.insert(proc2, null);
168 procStore.rollWriterForTesting();
169
170
171 procStore.delete(proc1.getProcId());
172 procStore.rollWriterForTesting();
173
174
175 procStore.update(proc2);
176 procStore.rollWriterForTesting();
177
178
179 procStore.delete(proc2.getProcId());
180
181
182 procStore.stop(false);
183 FileStatus[] logs = fs.listStatus(logDir);
184 assertEquals(4, logs.length);
185 for (int i = 0; i < logs.length; ++i) {
186 corruptLog(logs[i], 4);
187 }
188
189
190 LoadCounter loader = new LoadCounter();
191 storeRestart(loader);
192 assertEquals(1, loader.getLoadedCount());
193 assertEquals(0, loader.getCorruptedCount());
194
195
196 assertEquals(5, fs.listStatus(logDir).length);
197 loader = new LoadCounter();
198 storeRestart(loader);
199 assertEquals(1, loader.getLoadedCount());
200 assertEquals(0, loader.getCorruptedCount());
201
202
203 procStore.delete(proc0.getProcId());
204 procStore.periodicRollForTesting();
205 assertEquals(1, fs.listStatus(logDir).length);
206 storeRestart(loader);
207 }
208
209 @Test
210 public void testCorruptedTrailer() throws Exception {
211
212 for (int i = 0; i < 100; ++i) {
213 procStore.insert(new TestSequentialProcedure(), null);
214 }
215
216
217 procStore.stop(false);
218
219
220 FileStatus[] logs = fs.listStatus(logDir);
221 assertEquals(1, logs.length);
222 corruptLog(logs[0], 4);
223
224 LoadCounter loader = new LoadCounter();
225 storeRestart(loader);
226 assertEquals(100, loader.getLoadedCount());
227 assertEquals(0, loader.getCorruptedCount());
228 }
229
230 @Test
231 public void testCorruptedEntries() throws Exception {
232
233 for (int i = 0; i < 100; ++i) {
234 procStore.insert(new TestSequentialProcedure(), null);
235 }
236
237
238 procStore.stop(false);
239
240
241
242 FileStatus[] logs = fs.listStatus(logDir);
243 assertEquals(1, logs.length);
244 corruptLog(logs[0], 1823);
245
246 LoadCounter loader = new LoadCounter();
247 storeRestart(loader);
248 assertTrue(procStore.getCorruptedLogs() != null);
249 assertEquals(1, procStore.getCorruptedLogs().size());
250 assertEquals(85, loader.getLoadedCount());
251 assertEquals(0, loader.getCorruptedCount());
252 }
253
254 @Test
255 public void testCorruptedProcedures() throws Exception {
256
257 TestProcedure[] rootProcs = new TestProcedure[10];
258 for (int i = 1; i <= rootProcs.length; i++) {
259 rootProcs[i-1] = new TestProcedure(i, 0);
260 procStore.insert(rootProcs[i-1], null);
261 rootProcs[i-1].addStackId(0);
262 procStore.update(rootProcs[i-1]);
263 }
264
265 procStore.rollWriterForTesting();
266 for (int i = 1; i <= rootProcs.length; i++) {
267 TestProcedure b = new TestProcedure(rootProcs.length + i, i);
268 rootProcs[i-1].addStackId(1);
269 procStore.insert(rootProcs[i-1], new Procedure[] { b });
270 }
271
272 procStore.rollWriterForTesting();
273 for (int i = 1; i <= rootProcs.length; i++) {
274 procStore.update(new TestProcedure(rootProcs.length + i, i));
275 }
276
277
278 procStore.stop(false);
279
280
281
282 FileStatus[] logs = fs.listStatus(logDir);
283 assertEquals(Arrays.toString(logs), 2, logs.length);
284 Arrays.sort(logs, new Comparator<FileStatus>() {
285 @Override
286 public int compare(FileStatus o1, FileStatus o2) {
287 return o1.getPath().getName().compareTo(o2.getPath().getName());
288 }
289 });
290
291 LoadCounter loader = new LoadCounter();
292 storeRestart(loader);
293 assertEquals(rootProcs.length * 2, loader.getLoadedCount());
294 assertEquals(0, loader.getCorruptedCount());
295
296
297 fs.delete(logs[0].getPath(), false);
298 loader.reset();
299 storeRestart(loader);
300 assertEquals(0, loader.getLoadedCount());
301 assertEquals(rootProcs.length, loader.getCorruptedCount());
302 for (Procedure proc: loader.getCorrupted()) {
303 assertTrue(proc.toString(), proc.getParentProcId() <= rootProcs.length);
304 assertTrue(proc.toString(),
305 proc.getProcId() > rootProcs.length &&
306 proc.getProcId() <= (rootProcs.length * 2));
307 }
308 }
309
310 @Test(timeout=60000)
311 public void testWalReplayOrder_AB_A() throws Exception {
312
313
314
315 TestProcedure a = new TestProcedure(1, 0);
316 TestProcedure b = new TestProcedure(2, 1);
317
318 procStore.insert(a, null);
319 a.addStackId(0);
320 procStore.update(a);
321
322 procStore.insert(a, new Procedure[] { b });
323 b.addStackId(1);
324 procStore.update(b);
325
326 procStore.rollWriterForTesting();
327
328 a.addStackId(2);
329 procStore.update(a);
330
331 storeRestart(new ProcedureStore.ProcedureLoader() {
332 @Override
333 public void setMaxProcId(long maxProcId) {
334 assertEquals(2, maxProcId);
335 }
336
337 @Override
338 public void load(ProcedureIterator procIter) throws IOException {
339 assertTrue(procIter.hasNext());
340 assertEquals(1, procIter.nextAsProcedureInfo().getProcId());
341 assertTrue(procIter.hasNext());
342 assertEquals(2, procIter.nextAsProcedureInfo().getProcId());
343 assertFalse(procIter.hasNext());
344 }
345
346 @Override
347 public void handleCorrupted(ProcedureIterator procIter) throws IOException {
348 assertFalse(procIter.hasNext());
349 }
350 });
351 }
352
353 @Test(timeout=60000)
354 public void testWalReplayOrder_ABC_BAD() throws Exception {
355
356
357
358 TestProcedure a = new TestProcedure(1, 0);
359 TestProcedure b = new TestProcedure(2, 1);
360 TestProcedure c = new TestProcedure(3, 2);
361 TestProcedure d = new TestProcedure(4, 0);
362
363 procStore.insert(a, null);
364 a.addStackId(0);
365 procStore.update(a);
366
367 procStore.insert(a, new Procedure[] { b });
368 b.addStackId(1);
369 procStore.update(b);
370
371 procStore.insert(b, new Procedure[] { c });
372 b.addStackId(2);
373 procStore.update(b);
374
375 procStore.rollWriterForTesting();
376
377 b.addStackId(3);
378 procStore.update(b);
379
380 a.addStackId(4);
381 procStore.update(a);
382
383 procStore.insert(d, null);
384 d.addStackId(0);
385 procStore.update(d);
386
387 storeRestart(new ProcedureStore.ProcedureLoader() {
388 @Override
389 public void setMaxProcId(long maxProcId) {
390 assertEquals(4, maxProcId);
391 }
392
393 @Override
394 public void load(ProcedureIterator procIter) throws IOException {
395 assertTrue(procIter.hasNext());
396 assertEquals(4, procIter.nextAsProcedureInfo().getProcId());
397
398
399
400 assertTrue(procIter.hasNext());
401 assertEquals(1, procIter.nextAsProcedureInfo().getProcId());
402 assertTrue(procIter.hasNext());
403 assertEquals(2, procIter.nextAsProcedureInfo().getProcId());
404 assertTrue(procIter.hasNext());
405 assertEquals(3, procIter.nextAsProcedureInfo().getProcId());
406 assertFalse(procIter.hasNext());
407 }
408
409 @Override
410 public void handleCorrupted(ProcedureIterator procIter) throws IOException {
411 assertFalse(procIter.hasNext());
412 }
413 });
414 }
415
416 @Test
417 public void testRollAndRemove() throws IOException {
418
419 Procedure proc1 = new TestSequentialProcedure();
420 procStore.insert(proc1, null);
421
422 Procedure proc2 = new TestSequentialProcedure();
423 procStore.insert(proc2, null);
424
425
426 procStore.rollWriterForTesting();
427 assertEquals(2, procStore.getActiveLogs().size());
428
429
430
431 procStore.update(proc1);
432 procStore.update(proc2);
433 assertEquals(1, procStore.getActiveLogs().size());
434
435
436 procStore.rollWriterForTesting();
437 assertEquals(2, procStore.getActiveLogs().size());
438
439
440
441 procStore.delete(proc1.getProcId());
442 procStore.delete(proc2.getProcId());
443 assertEquals(1, procStore.getActiveLogs().size());
444 }
445
446 @Test
447 public void testFileNotFoundDuringLeaseRecovery() throws IOException {
448 TestProcedure[] procs = new TestProcedure[3];
449 for (int i = 0; i < procs.length; ++i) {
450 procs[i] = new TestProcedure(i + 1, 0);
451 procStore.insert(procs[i], null);
452 }
453 procStore.rollWriterForTesting();
454 for (int i = 0; i < procs.length; ++i) {
455 procStore.update(procs[i]);
456 procStore.rollWriterForTesting();
457 }
458 procStore.stop(false);
459
460 FileStatus[] status = fs.listStatus(logDir);
461 assertEquals(procs.length + 2, status.length);
462
463
464 procStore = new WALProcedureStore(htu.getConfiguration(), fs, logDir,
465 new WALProcedureStore.LeaseRecovery() {
466 private int count = 0;
467
468 @Override
469 public void recoverFileLease(FileSystem fs, Path path) throws IOException {
470 if (++count <= 2) {
471 fs.delete(path, false);
472 LOG.debug("Simulate FileNotFound at count=" + count + " for " + path);
473 throw new FileNotFoundException("test file not found " + path);
474 }
475 LOG.debug("Simulate recoverFileLease() at count=" + count + " for " + path);
476 }
477 });
478
479 final LoadCounter loader = new LoadCounter();
480 procStore.start(PROCEDURE_STORE_SLOTS);
481 procStore.recoverLease();
482 procStore.load(loader);
483 assertEquals(procs.length, loader.getMaxProcId());
484 assertEquals(procs.length - 1, loader.getLoadedCount());
485 assertEquals(0, loader.getCorruptedCount());
486 }
487
488 private void corruptLog(final FileStatus logFile, final long dropBytes)
489 throws IOException {
490 assertTrue(logFile.getLen() > dropBytes);
491 LOG.debug("corrupt log " + logFile.getPath() +
492 " size=" + logFile.getLen() + " drop=" + dropBytes);
493 Path tmpPath = new Path(testDir, "corrupted.log");
494 InputStream in = fs.open(logFile.getPath());
495 OutputStream out = fs.create(tmpPath);
496 IOUtils.copyBytes(in, out, logFile.getLen() - dropBytes, true);
497 if (!fs.rename(tmpPath, logFile.getPath())) {
498 throw new IOException("Unable to rename");
499 }
500 }
501
502 private void verifyProcIdsOnRestart(final Set<Long> procIds) throws Exception {
503 LOG.debug("expected: " + procIds);
504 LoadCounter loader = new LoadCounter();
505 storeRestart(loader);
506 assertEquals(procIds.size(), loader.getLoadedCount());
507 assertEquals(0, loader.getCorruptedCount());
508 }
509
510 private void assertEmptyLogDir() {
511 try {
512 FileStatus[] status = fs.listStatus(logDir);
513 assertTrue("expected empty state-log dir", status == null || status.length == 0);
514 } catch (FileNotFoundException e) {
515 fail("expected the state-log dir to be present: " + logDir);
516 } catch (IOException e) {
517 fail("got en exception on state-log dir list: " + e.getMessage());
518 }
519 }
520
521 public static class TestSequentialProcedure extends SequentialProcedure<Void> {
522 private static long seqid = 0;
523
524 public TestSequentialProcedure() {
525 setProcId(++seqid);
526 }
527
528 @Override
529 protected Procedure[] execute(Void env) { return null; }
530
531 @Override
532 protected void rollback(Void env) { }
533
534 @Override
535 protected boolean abort(Void env) { return false; }
536
537 @Override
538 protected void serializeStateData(final OutputStream stream) throws IOException {
539 long procId = getProcId();
540 if (procId % 2 == 0) {
541 stream.write(Bytes.toBytes(procId));
542 }
543 }
544
545 @Override
546 protected void deserializeStateData(InputStream stream) throws IOException {
547 long procId = getProcId();
548 if (procId % 2 == 0) {
549 byte[] bProcId = new byte[8];
550 assertEquals(8, stream.read(bProcId));
551 assertEquals(procId, Bytes.toLong(bProcId));
552 } else {
553 assertEquals(0, stream.available());
554 }
555 }
556 }
557 }