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
25 import org.apache.commons.logging.Log;
26 import org.apache.commons.logging.LogFactory;
27 import org.apache.hadoop.conf.Configuration;
28 import org.apache.hadoop.fs.FileSystem;
29 import org.apache.hadoop.fs.Path;
30 import org.apache.hadoop.hbase.HConstants;
31 import org.apache.hadoop.hbase.ProcedureInfo;
32 import org.apache.hadoop.hbase.util.Threads;
33 import org.apache.hadoop.hbase.exceptions.IllegalArgumentIOException;
34 import org.apache.hadoop.hbase.exceptions.TimeoutIOException;
35 import org.apache.hadoop.hbase.procedure2.store.ProcedureStore;
36 import org.apache.hadoop.hbase.procedure2.store.wal.WALProcedureStore;
37 import org.apache.hadoop.hbase.protobuf.generated.ErrorHandlingProtos.ForeignExceptionMessage;
38
39 import static org.junit.Assert.assertEquals;
40 import static org.junit.Assert.assertFalse;
41 import static org.junit.Assert.assertTrue;
42
43 public class ProcedureTestingUtility {
44 private static final Log LOG = LogFactory.getLog(ProcedureTestingUtility.class);
45
46 private ProcedureTestingUtility() {
47 }
48
49 public static ProcedureStore createStore(final Configuration conf, final FileSystem fs,
50 final Path baseDir) throws IOException {
51 return createWalStore(conf, fs, baseDir);
52 }
53
54 public static WALProcedureStore createWalStore(final Configuration conf, final FileSystem fs,
55 final Path logDir) throws IOException {
56 return new WALProcedureStore(conf, fs, logDir, new WALProcedureStore.LeaseRecovery() {
57 @Override
58 public void recoverFileLease(FileSystem fs, Path path) throws IOException {
59
60 }
61 });
62 }
63
64 public static <TEnv> void restart(ProcedureExecutor<TEnv> procExecutor)
65 throws Exception {
66 restart(procExecutor, null);
67 }
68
69 public static <TEnv> void restart(ProcedureExecutor<TEnv> procExecutor,
70 Runnable beforeStartAction) throws Exception {
71 ProcedureStore procStore = procExecutor.getStore();
72 int storeThreads = procExecutor.getNumThreads();
73 int execThreads = procExecutor.getNumThreads();
74
75 procExecutor.stop();
76 procExecutor.join();
77 procStore.stop(false);
78
79 if (beforeStartAction != null) {
80 beforeStartAction.run();
81 }
82
83 procStore.start(storeThreads);
84 procExecutor.start(execThreads);
85 }
86
87 public static <TEnv> void setKillBeforeStoreUpdate(ProcedureExecutor<TEnv> procExecutor,
88 boolean value) {
89 if (procExecutor.testing == null) {
90 procExecutor.testing = new ProcedureExecutor.Testing();
91 }
92 procExecutor.testing.killBeforeStoreUpdate = value;
93 LOG.warn("Set Kill before store update to: " + procExecutor.testing.killBeforeStoreUpdate);
94 }
95
96 public static <TEnv> void setToggleKillBeforeStoreUpdate(ProcedureExecutor<TEnv> procExecutor,
97 boolean value) {
98 if (procExecutor.testing == null) {
99 procExecutor.testing = new ProcedureExecutor.Testing();
100 }
101 procExecutor.testing.toggleKillBeforeStoreUpdate = value;
102 }
103
104 public static <TEnv> void toggleKillBeforeStoreUpdate(ProcedureExecutor<TEnv> procExecutor) {
105 if (procExecutor.testing == null) {
106 procExecutor.testing = new ProcedureExecutor.Testing();
107 }
108 procExecutor.testing.killBeforeStoreUpdate = !procExecutor.testing.killBeforeStoreUpdate;
109 LOG.warn("Set Kill before store update to: " + procExecutor.testing.killBeforeStoreUpdate);
110 }
111
112 public static <TEnv> void setKillAndToggleBeforeStoreUpdate(ProcedureExecutor<TEnv> procExecutor,
113 boolean value) {
114 ProcedureTestingUtility.setKillBeforeStoreUpdate(procExecutor, value);
115 ProcedureTestingUtility.setToggleKillBeforeStoreUpdate(procExecutor, value);
116 }
117
118 public static <TEnv> long submitAndWait(ProcedureExecutor<TEnv> procExecutor, Procedure proc) {
119 return submitAndWait(procExecutor, proc, HConstants.NO_NONCE, HConstants.NO_NONCE);
120 }
121
122 public static <TEnv> long submitAndWait(ProcedureExecutor<TEnv> procExecutor, Procedure proc,
123 final long nonceGroup,
124 final long nonce) {
125 long procId = procExecutor.submitProcedure(proc, nonceGroup, nonce);
126 waitProcedure(procExecutor, procId);
127 return procId;
128 }
129
130 public static <TEnv> void waitProcedure(ProcedureExecutor<TEnv> procExecutor, long procId) {
131 while (!procExecutor.isFinished(procId) && procExecutor.isRunning()) {
132 Threads.sleepWithoutInterrupt(250);
133 }
134 }
135
136 public static <TEnv> void waitNoProcedureRunning(ProcedureExecutor<TEnv> procExecutor) {
137 int stableRuns = 0;
138 while (stableRuns < 10) {
139 if (procExecutor.getActiveExecutorCount() > 0 || procExecutor.getRunnableSet().size() > 0) {
140 stableRuns = 0;
141 Threads.sleepWithoutInterrupt(100);
142 } else {
143 stableRuns++;
144 Threads.sleepWithoutInterrupt(25);
145 }
146 }
147 }
148
149 public static <TEnv> void assertProcNotYetCompleted(ProcedureExecutor<TEnv> procExecutor,
150 long procId) {
151 assertFalse("expected a running proc", procExecutor.isFinished(procId));
152 assertEquals(null, procExecutor.getResult(procId));
153 }
154
155 public static <TEnv> void assertProcNotFailed(ProcedureExecutor<TEnv> procExecutor,
156 long procId) {
157 ProcedureInfo result = procExecutor.getResult(procId);
158 assertTrue("expected procedure result", result != null);
159 assertProcNotFailed(result);
160 }
161
162 public static void assertProcNotFailed(final ProcedureInfo result) {
163 ForeignExceptionMessage exception = result.getForeignExceptionMessage();
164 String msg = exception != null ? result.getExceptionFullMessage() : "no exception found";
165 assertFalse(msg, result.isFailed());
166 }
167
168 public static void assertIsAbortException(final ProcedureInfo result) {
169 assertEquals(true, result.isFailed());
170 LOG.info(result.getExceptionFullMessage());
171 Throwable cause = getExceptionCause(result);
172 assertTrue("expected abort exception, got " + cause,
173 cause instanceof ProcedureAbortedException);
174 }
175
176 public static void assertIsTimeoutException(final ProcedureInfo result) {
177 assertEquals(true, result.isFailed());
178 LOG.info(result.getExceptionFullMessage());
179 Throwable cause = getExceptionCause(result);
180 assertTrue("expected TimeoutIOException, got " + cause, cause instanceof TimeoutIOException);
181 }
182
183 public static void assertIsIllegalArgumentException(final ProcedureInfo result) {
184 assertEquals(true, result.isFailed());
185 LOG.info(result.getExceptionFullMessage());
186 Throwable cause = ProcedureTestingUtility.getExceptionCause(result);
187 assertTrue("expected IllegalArgumentIOException, got " + cause,
188 cause instanceof IllegalArgumentIOException);
189 }
190
191 public static Throwable getExceptionCause(final ProcedureInfo procInfo) {
192 assert procInfo.getForeignExceptionMessage() != null;
193 return RemoteProcedureException.fromProto(procInfo.getForeignExceptionMessage()).getCause();
194 }
195
196 public static class TestProcedure extends Procedure<Void> {
197 public TestProcedure() {}
198
199 public TestProcedure(long procId) {
200 this(procId, 0);
201 }
202
203 public TestProcedure(long procId, long parentId) {
204 setProcId(procId);
205 if (parentId > 0) {
206 setParentProcId(parentId);
207 }
208 }
209
210 public void addStackId(final int index) {
211 addStackIndex(index);
212 }
213
214 @Override
215 protected Procedure[] execute(Void env) { return null; }
216
217 @Override
218 protected void rollback(Void env) { }
219
220 @Override
221 protected boolean abort(Void env) { return false; }
222
223 @Override
224 protected void serializeStateData(final OutputStream stream) throws IOException { }
225
226 @Override
227 protected void deserializeStateData(final InputStream stream) throws IOException { }
228 }
229 }