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;
20
21 import java.io.IOException;
22 import java.io.InputStream;
23 import java.io.OutputStream;
24 import java.util.concurrent.atomic.AtomicBoolean;
25
26 import org.apache.commons.logging.Log;
27 import org.apache.commons.logging.LogFactory;
28 import org.apache.hadoop.fs.FileSystem;
29 import org.apache.hadoop.fs.FileStatus;
30 import org.apache.hadoop.fs.Path;
31 import org.apache.hadoop.hbase.HBaseCommonTestingUtility;
32 import org.apache.hadoop.hbase.ProcedureInfo;
33 import org.apache.hadoop.hbase.procedure2.store.ProcedureStore;
34 import org.apache.hadoop.hbase.testclassification.SmallTests;
35 import org.apache.hadoop.hbase.util.Bytes;
36 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
37 import org.apache.hadoop.hbase.util.Threads;
38 import org.junit.After;
39 import org.junit.Before;
40 import org.junit.Test;
41 import org.junit.experimental.categories.Category;
42
43 import static org.junit.Assert.assertEquals;
44 import static org.junit.Assert.assertFalse;
45 import static org.junit.Assert.assertTrue;
46
47 @Category(SmallTests.class)
48 public class TestProcedureRecovery {
49 private static final Log LOG = LogFactory.getLog(TestProcedureRecovery.class);
50
51 private static final int PROCEDURE_EXECUTOR_SLOTS = 1;
52
53 private static ProcedureExecutor<Void> procExecutor;
54 private static ProcedureStore procStore;
55 private static int procSleepInterval;
56
57 private HBaseCommonTestingUtility htu;
58 private FileSystem fs;
59 private Path testDir;
60 private Path logDir;
61
62 @Before
63 public void setUp() throws IOException {
64 htu = new HBaseCommonTestingUtility();
65 testDir = htu.getDataTestDir();
66 fs = testDir.getFileSystem(htu.getConfiguration());
67 assertTrue(testDir.depth() > 1);
68
69 logDir = new Path(testDir, "proc-logs");
70 procStore = ProcedureTestingUtility.createStore(htu.getConfiguration(), fs, logDir);
71 procExecutor = new ProcedureExecutor(htu.getConfiguration(), null, procStore);
72 procExecutor.testing = new ProcedureExecutor.Testing();
73 procStore.start(PROCEDURE_EXECUTOR_SLOTS);
74 procExecutor.start(PROCEDURE_EXECUTOR_SLOTS);
75 procSleepInterval = 0;
76
77 ProcedureTestingUtility.setToggleKillBeforeStoreUpdate(procExecutor, false);
78 ProcedureTestingUtility.setKillBeforeStoreUpdate(procExecutor, false);
79 }
80
81 @After
82 public void tearDown() throws IOException {
83 procExecutor.stop();
84 procStore.stop(false);
85 fs.delete(logDir, true);
86 }
87
88 private void restart() throws Exception {
89 dumpLogDirState();
90 ProcedureTestingUtility.restart(procExecutor);
91 dumpLogDirState();
92 }
93
94 public static class TestSingleStepProcedure extends SequentialProcedure<Void> {
95 private int step = 0;
96
97 public TestSingleStepProcedure() { }
98
99 @Override
100 protected Procedure[] execute(Void env) {
101 LOG.debug("execute procedure " + this + " step=" + step);
102 step++;
103 setResult(Bytes.toBytes(step));
104 return null;
105 }
106
107 @Override
108 protected void rollback(Void env) { }
109
110 @Override
111 protected boolean abort(Void env) { return true; }
112 }
113
114 public static class BaseTestStepProcedure extends SequentialProcedure<Void> {
115 private AtomicBoolean abort = new AtomicBoolean(false);
116 private int step = 0;
117
118 @Override
119 protected Procedure[] execute(Void env) {
120 LOG.debug("execute procedure " + this + " step=" + step);
121 ProcedureTestingUtility.toggleKillBeforeStoreUpdate(procExecutor);
122 step++;
123 Threads.sleepWithoutInterrupt(procSleepInterval);
124 if (isAborted()) {
125 setFailure(new RemoteProcedureException(getClass().getName(),
126 new ProcedureAbortedException(
127 "got an abort at " + getClass().getName() + " step=" + step)));
128 return null;
129 }
130 return null;
131 }
132
133 @Override
134 protected void rollback(Void env) {
135 LOG.debug("rollback procedure " + this + " step=" + step);
136 ProcedureTestingUtility.toggleKillBeforeStoreUpdate(procExecutor);
137 step++;
138 }
139
140 @Override
141 protected boolean abort(Void env) {
142 abort.set(true);
143 return true;
144 }
145
146 private boolean isAborted() {
147 boolean aborted = abort.get();
148 BaseTestStepProcedure proc = this;
149 while (proc.hasParent() && !aborted) {
150 proc = (BaseTestStepProcedure)procExecutor.getProcedure(proc.getParentProcId());
151 aborted = proc.isAborted();
152 }
153 return aborted;
154 }
155 }
156
157 public static class TestMultiStepProcedure extends BaseTestStepProcedure {
158 public TestMultiStepProcedure() { }
159
160 @Override
161 public Procedure[] execute(Void env) {
162 super.execute(env);
163 return isFailed() ? null : new Procedure[] { new Step1Procedure() };
164 }
165
166 public static class Step1Procedure extends BaseTestStepProcedure {
167 public Step1Procedure() { }
168
169 @Override
170 protected Procedure[] execute(Void env) {
171 super.execute(env);
172 return isFailed() ? null : new Procedure[] { new Step2Procedure() };
173 }
174 }
175
176 public static class Step2Procedure extends BaseTestStepProcedure {
177 public Step2Procedure() { }
178 }
179 }
180
181 @Test
182 public void testNoopLoad() throws Exception {
183 restart();
184 }
185
186 @Test(timeout=30000)
187 public void testSingleStepProcRecovery() throws Exception {
188 Procedure proc = new TestSingleStepProcedure();
189 procExecutor.testing.killBeforeStoreUpdate = true;
190 long procId = ProcedureTestingUtility.submitAndWait(procExecutor, proc);
191 assertFalse(procExecutor.isRunning());
192 procExecutor.testing.killBeforeStoreUpdate = false;
193
194
195 long restartTs = EnvironmentEdgeManager.currentTime();
196 restart();
197 waitProcedure(procId);
198 ProcedureInfo result = procExecutor.getResult(procId);
199 assertTrue(result.getLastUpdate() > restartTs);
200 ProcedureTestingUtility.assertProcNotFailed(result);
201 assertEquals(1, Bytes.toInt(result.getResult()));
202 long resultTs = result.getLastUpdate();
203
204
205 restart();
206 result = procExecutor.getResult(procId);
207 ProcedureTestingUtility.assertProcNotFailed(result);
208 assertEquals(resultTs, result.getLastUpdate());
209 assertEquals(1, Bytes.toInt(result.getResult()));
210 }
211
212 @Test(timeout=30000)
213 public void testMultiStepProcRecovery() throws Exception {
214
215 Procedure proc = new TestMultiStepProcedure();
216 long procId = ProcedureTestingUtility.submitAndWait(procExecutor, proc);
217 assertFalse(procExecutor.isRunning());
218
219
220 restart();
221 waitProcedure(procId);
222 ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
223 assertFalse(procExecutor.isRunning());
224
225
226 restart();
227 waitProcedure(procId);
228 ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
229 assertFalse(procExecutor.isRunning());
230
231
232 restart();
233 waitProcedure(procId);
234 assertTrue(procExecutor.isRunning());
235
236
237 ProcedureInfo result = procExecutor.getResult(procId);
238 ProcedureTestingUtility.assertProcNotFailed(result);
239 }
240
241 @Test(timeout=30000)
242 public void testMultiStepRollbackRecovery() throws Exception {
243
244 Procedure proc = new TestMultiStepProcedure();
245 long procId = ProcedureTestingUtility.submitAndWait(procExecutor, proc);
246 assertFalse(procExecutor.isRunning());
247
248
249 restart();
250 waitProcedure(procId);
251 ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
252 assertFalse(procExecutor.isRunning());
253
254
255 restart();
256 waitProcedure(procId);
257 ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
258 assertFalse(procExecutor.isRunning());
259
260
261 procSleepInterval = 2500;
262 restart();
263 assertTrue(procExecutor.abort(procId));
264 waitProcedure(procId);
265 assertFalse(procExecutor.isRunning());
266
267
268 restart();
269 waitProcedure(procId);
270 ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
271 assertFalse(procExecutor.isRunning());
272
273
274 restart();
275 waitProcedure(procId);
276 ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
277 assertFalse(procExecutor.isRunning());
278
279
280 restart();
281 waitProcedure(procId);
282
283
284 ProcedureInfo result = procExecutor.getResult(procId);
285 ProcedureTestingUtility.assertIsAbortException(result);
286 }
287
288 @Test(timeout=30000)
289 public void testCompletedProcWithSameNonce() throws Exception {
290 final long nonceGroup = 123;
291 final long nonce = 2222;
292 Procedure proc = new TestSingleStepProcedure();
293
294 long procId = ProcedureTestingUtility.submitAndWait(procExecutor, proc, nonceGroup, nonce);
295
296
297 restart();
298 Procedure proc2 = new TestSingleStepProcedure();
299
300 long procId2 = ProcedureTestingUtility.submitAndWait(procExecutor, proc2, nonceGroup, nonce);
301 assertTrue(procId == procId2);
302
303 ProcedureInfo result = procExecutor.getResult(procId2);
304 ProcedureTestingUtility.assertProcNotFailed(result);
305 }
306
307 @Test(timeout=30000)
308 public void testRunningProcWithSameNonce() throws Exception {
309 final long nonceGroup = 456;
310 final long nonce = 33333;
311 Procedure proc = new TestMultiStepProcedure();
312 long procId = ProcedureTestingUtility.submitAndWait(procExecutor, proc, nonceGroup, nonce);
313
314
315 restart();
316 Procedure proc2 = new TestMultiStepProcedure();
317
318 long procId2 = ProcedureTestingUtility.submitAndWait(procExecutor, proc2, nonceGroup, nonce);
319
320 assertTrue(procId == procId2);
321 }
322
323 public static class TestStateMachineProcedure
324 extends StateMachineProcedure<Void, TestStateMachineProcedure.State> {
325 enum State { STATE_1, STATE_2, STATE_3, DONE }
326
327 public TestStateMachineProcedure() {}
328
329 private AtomicBoolean aborted = new AtomicBoolean(false);
330 private int iResult = 0;
331
332 @Override
333 protected StateMachineProcedure.Flow executeFromState(Void env, State state) {
334 switch (state) {
335 case STATE_1:
336 LOG.info("execute step 1 " + this);
337 setNextState(State.STATE_2);
338 iResult += 3;
339 break;
340 case STATE_2:
341 LOG.info("execute step 2 " + this);
342 setNextState(State.STATE_3);
343 iResult += 5;
344 break;
345 case STATE_3:
346 LOG.info("execute step 3 " + this);
347 Threads.sleepWithoutInterrupt(procSleepInterval);
348 if (aborted.get()) {
349 LOG.info("aborted step 3 " + this);
350 setAbortFailure("test", "aborted");
351 break;
352 }
353 setNextState(State.DONE);
354 iResult += 7;
355 setResult(Bytes.toBytes(iResult));
356 return Flow.NO_MORE_STATE;
357 default:
358 throw new UnsupportedOperationException();
359 }
360 return Flow.HAS_MORE_STATE;
361 }
362
363 @Override
364 protected void rollbackState(Void env, final State state) {
365 switch (state) {
366 case STATE_1:
367 LOG.info("rollback step 1 " + this);
368 break;
369 case STATE_2:
370 LOG.info("rollback step 2 " + this);
371 break;
372 case STATE_3:
373 LOG.info("rollback step 3 " + this);
374 break;
375 default:
376 throw new UnsupportedOperationException();
377 }
378 }
379
380 @Override
381 protected State getState(final int stateId) {
382 return State.values()[stateId];
383 }
384
385 @Override
386 protected int getStateId(final State state) {
387 return state.ordinal();
388 }
389
390 @Override
391 protected State getInitialState() {
392 return State.STATE_1;
393 }
394
395 @Override
396 protected boolean abort(Void env) {
397 aborted.set(true);
398 return true;
399 }
400
401 @Override
402 protected void serializeStateData(final OutputStream stream) throws IOException {
403 super.serializeStateData(stream);
404 stream.write(Bytes.toBytes(iResult));
405 }
406
407 @Override
408 protected void deserializeStateData(final InputStream stream) throws IOException {
409 super.deserializeStateData(stream);
410 byte[] data = new byte[4];
411 stream.read(data);
412 iResult = Bytes.toInt(data);
413 }
414 }
415
416 @Test(timeout=30000)
417 public void testStateMachineRecovery() throws Exception {
418 ProcedureTestingUtility.setToggleKillBeforeStoreUpdate(procExecutor, true);
419 ProcedureTestingUtility.setKillBeforeStoreUpdate(procExecutor, true);
420
421
422 Procedure proc = new TestStateMachineProcedure();
423 long procId = ProcedureTestingUtility.submitAndWait(procExecutor, proc);
424 assertFalse(procExecutor.isRunning());
425
426
427 restart();
428 waitProcedure(procId);
429 ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
430 assertFalse(procExecutor.isRunning());
431
432
433 restart();
434 waitProcedure(procId);
435 ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
436 assertFalse(procExecutor.isRunning());
437
438
439 restart();
440 waitProcedure(procId);
441 assertTrue(procExecutor.isRunning());
442
443
444 ProcedureInfo result = procExecutor.getResult(procId);
445 ProcedureTestingUtility.assertProcNotFailed(result);
446 assertEquals(15, Bytes.toInt(result.getResult()));
447 }
448
449 @Test(timeout=30000)
450 public void testStateMachineRollbackRecovery() throws Exception {
451 ProcedureTestingUtility.setToggleKillBeforeStoreUpdate(procExecutor, true);
452 ProcedureTestingUtility.setKillBeforeStoreUpdate(procExecutor, true);
453
454
455 Procedure proc = new TestStateMachineProcedure();
456 long procId = ProcedureTestingUtility.submitAndWait(procExecutor, proc);
457 ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
458 assertFalse(procExecutor.isRunning());
459
460
461 restart();
462 waitProcedure(procId);
463 ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
464 assertFalse(procExecutor.isRunning());
465
466
467 restart();
468 waitProcedure(procId);
469 ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
470 assertFalse(procExecutor.isRunning());
471
472
473 procSleepInterval = 2500;
474 restart();
475 assertTrue(procExecutor.abort(procId));
476 waitProcedure(procId);
477 ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
478 assertFalse(procExecutor.isRunning());
479
480
481 restart();
482 waitProcedure(procId);
483 assertFalse(procExecutor.isRunning());
484 ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
485
486
487 restart();
488 waitProcedure(procId);
489 assertFalse(procExecutor.isRunning());
490 ProcedureTestingUtility.assertProcNotYetCompleted(procExecutor, procId);
491
492
493 restart();
494 waitProcedure(procId);
495 assertTrue(procExecutor.isRunning());
496
497
498 ProcedureInfo result = procExecutor.getResult(procId);
499 ProcedureTestingUtility.assertIsAbortException(result);
500 }
501
502 private void waitProcedure(final long procId) {
503 ProcedureTestingUtility.waitProcedure(procExecutor, procId);
504 dumpLogDirState();
505 }
506
507 private void dumpLogDirState() {
508 try {
509 FileStatus[] files = fs.listStatus(logDir);
510 if (files != null && files.length > 0) {
511 for (FileStatus file: files) {
512 assertTrue(file.toString(), file.isFile());
513 LOG.debug("log file " + file.getPath() + " size=" + file.getLen());
514 }
515 } else {
516 LOG.debug("no files under: " + logDir);
517 }
518 } catch (IOException e) {
519 LOG.warn("Unable to dump " + logDir, e);
520 }
521 }
522 }