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.ArrayList;
25 import java.util.Set;
26
27 import org.apache.commons.logging.Log;
28 import org.apache.commons.logging.LogFactory;
29 import org.apache.hadoop.conf.Configuration;
30 import org.apache.hadoop.fs.FileSystem;
31 import org.apache.hadoop.fs.Path;
32 import org.apache.hadoop.hbase.HConstants;
33 import org.apache.hadoop.hbase.ProcedureInfo;
34 import org.apache.hadoop.hbase.io.util.StreamUtils;
35 import org.apache.hadoop.hbase.util.Threads;
36 import org.apache.hadoop.hbase.exceptions.IllegalArgumentIOException;
37 import org.apache.hadoop.hbase.exceptions.TimeoutIOException;
38 import org.apache.hadoop.hbase.procedure2.store.ProcedureStore;
39 import org.apache.hadoop.hbase.procedure2.store.ProcedureStore.ProcedureIterator;
40 import org.apache.hadoop.hbase.procedure2.store.NoopProcedureStore;
41 import org.apache.hadoop.hbase.procedure2.store.wal.WALProcedureStore;
42 import org.apache.hadoop.hbase.protobuf.generated.ErrorHandlingProtos.ForeignExceptionMessage;
43 import org.apache.hadoop.hbase.protobuf.generated.ProcedureProtos;
44 import org.apache.hadoop.hbase.util.NonceKey;
45 import static org.junit.Assert.assertEquals;
46 import static org.junit.Assert.assertFalse;
47 import static org.junit.Assert.assertTrue;
48
49 public class ProcedureTestingUtility {
50 private static final Log LOG = LogFactory.getLog(ProcedureTestingUtility.class);
51
52 private ProcedureTestingUtility() {
53 }
54
55 public static ProcedureStore createStore(final Configuration conf, final FileSystem fs,
56 final Path baseDir) throws IOException {
57 return createWalStore(conf, fs, baseDir);
58 }
59
60 public static WALProcedureStore createWalStore(final Configuration conf, final FileSystem fs,
61 final Path logDir) throws IOException {
62 return new WALProcedureStore(conf, fs, logDir, new WALProcedureStore.LeaseRecovery() {
63 @Override
64 public void recoverFileLease(FileSystem fs, Path path) throws IOException {
65
66 }
67 });
68 }
69
70 public static <TEnv> void restart(ProcedureExecutor<TEnv> procExecutor)
71 throws Exception {
72 restart(procExecutor, null, true);
73 }
74
75 public static <TEnv> void restart(ProcedureExecutor<TEnv> procExecutor,
76 Runnable beforeStartAction, boolean failOnCorrupted) throws Exception {
77 ProcedureStore procStore = procExecutor.getStore();
78 int storeThreads = procExecutor.getNumThreads();
79 int execThreads = procExecutor.getNumThreads();
80
81 procExecutor.stop();
82 procExecutor.join();
83 procStore.stop(false);
84
85 if (beforeStartAction != null) {
86 beforeStartAction.run();
87 }
88
89 procStore.start(storeThreads);
90 procExecutor.start(execThreads, failOnCorrupted);
91 }
92
93 public static <TEnv> void setKillBeforeStoreUpdate(ProcedureExecutor<TEnv> procExecutor,
94 boolean value) {
95 if (procExecutor.testing == null) {
96 procExecutor.testing = new ProcedureExecutor.Testing();
97 }
98 procExecutor.testing.killBeforeStoreUpdate = value;
99 LOG.warn("Set Kill before store update to: " + procExecutor.testing.killBeforeStoreUpdate);
100 }
101
102 public static <TEnv> void setToggleKillBeforeStoreUpdate(ProcedureExecutor<TEnv> procExecutor,
103 boolean value) {
104 if (procExecutor.testing == null) {
105 procExecutor.testing = new ProcedureExecutor.Testing();
106 }
107 procExecutor.testing.toggleKillBeforeStoreUpdate = value;
108 }
109
110 public static <TEnv> void toggleKillBeforeStoreUpdate(ProcedureExecutor<TEnv> procExecutor) {
111 if (procExecutor.testing == null) {
112 procExecutor.testing = new ProcedureExecutor.Testing();
113 }
114 procExecutor.testing.killBeforeStoreUpdate = !procExecutor.testing.killBeforeStoreUpdate;
115 LOG.warn("Set Kill before store update to: " + procExecutor.testing.killBeforeStoreUpdate);
116 }
117
118 public static <TEnv> void setKillAndToggleBeforeStoreUpdate(ProcedureExecutor<TEnv> procExecutor,
119 boolean value) {
120 ProcedureTestingUtility.setKillBeforeStoreUpdate(procExecutor, value);
121 ProcedureTestingUtility.setToggleKillBeforeStoreUpdate(procExecutor, value);
122 }
123
124 public static <TEnv> long submitAndWait(Configuration conf, TEnv env, Procedure<TEnv> proc)
125 throws IOException {
126 NoopProcedureStore procStore = new NoopProcedureStore();
127 ProcedureExecutor<TEnv> procExecutor = new ProcedureExecutor<TEnv>(conf, env, procStore);
128 procStore.start(1);
129 procExecutor.start(1, false);
130 try {
131 return submitAndWait(procExecutor, proc, HConstants.NO_NONCE, HConstants.NO_NONCE);
132 } finally {
133 procStore.stop(false);
134 procExecutor.stop();
135 }
136 }
137
138 public static <TEnv> long submitAndWait(ProcedureExecutor<TEnv> procExecutor, Procedure proc) {
139 return submitAndWait(procExecutor, proc, HConstants.NO_NONCE, HConstants.NO_NONCE);
140 }
141
142 public static <TEnv> long submitAndWait(ProcedureExecutor<TEnv> procExecutor, Procedure proc,
143 final long nonceGroup, final long nonce) {
144 long procId = submitProcedure(procExecutor, proc, nonceGroup, nonce);
145 waitProcedure(procExecutor, procId);
146 return procId;
147 }
148
149 public static <TEnv> long submitProcedure(ProcedureExecutor<TEnv> procExecutor, Procedure proc,
150 final long nonceGroup, final long nonce) {
151 final NonceKey nonceKey = procExecutor.createNonceKey(nonceGroup, nonce);
152 long procId = procExecutor.registerNonce(nonceKey);
153 assertFalse(procId >= 0);
154 return procExecutor.submitProcedure(proc, nonceKey);
155 }
156
157 public static <TEnv> void waitProcedure(ProcedureExecutor<TEnv> procExecutor, long procId) {
158 while (!procExecutor.isFinished(procId) && procExecutor.isRunning()) {
159 Threads.sleepWithoutInterrupt(250);
160 }
161 }
162
163 public static <TEnv> void waitNoProcedureRunning(ProcedureExecutor<TEnv> procExecutor) {
164 int stableRuns = 0;
165 while (stableRuns < 10) {
166 if (procExecutor.getActiveExecutorCount() > 0 || procExecutor.getRunnableSet().size() > 0) {
167 stableRuns = 0;
168 Threads.sleepWithoutInterrupt(100);
169 } else {
170 stableRuns++;
171 Threads.sleepWithoutInterrupt(25);
172 }
173 }
174 }
175
176 public static <TEnv> void assertProcNotYetCompleted(ProcedureExecutor<TEnv> procExecutor,
177 long procId) {
178 assertFalse("expected a running proc", procExecutor.isFinished(procId));
179 assertEquals(null, procExecutor.getResult(procId));
180 }
181
182 public static <TEnv> void assertProcNotFailed(ProcedureExecutor<TEnv> procExecutor,
183 long procId) {
184 ProcedureInfo result = procExecutor.getResult(procId);
185 assertTrue("expected procedure result", result != null);
186 assertProcNotFailed(result);
187 }
188
189 public static void assertProcNotFailed(final ProcedureInfo result) {
190 ForeignExceptionMessage exception = result.getForeignExceptionMessage();
191 String msg = exception != null ? result.getExceptionFullMessage() : "no exception found";
192 assertFalse(msg, result.isFailed());
193 }
194
195 public static Throwable assertProcFailed(final ProcedureInfo result) {
196 assertEquals(true, result.isFailed());
197 LOG.info("procId=" + result.getProcId() + " exception: " + result.getException().getMessage());
198 return getExceptionCause(result);
199 }
200
201 public static void assertIsAbortException(final ProcedureInfo result) {
202 assertEquals(true, result.isFailed());
203 LOG.info(result.getExceptionFullMessage());
204 Throwable cause = getExceptionCause(result);
205 assertTrue("expected abort exception, got " + cause,
206 cause instanceof ProcedureAbortedException);
207 }
208
209 public static void assertIsTimeoutException(final ProcedureInfo result) {
210 assertEquals(true, result.isFailed());
211 LOG.info(result.getExceptionFullMessage());
212 Throwable cause = getExceptionCause(result);
213 assertTrue("expected TimeoutIOException, got " + cause, cause instanceof TimeoutIOException);
214 }
215
216 public static void assertIsIllegalArgumentException(final ProcedureInfo result) {
217 assertEquals(true, result.isFailed());
218 LOG.info(result.getExceptionFullMessage());
219 Throwable cause = ProcedureTestingUtility.getExceptionCause(result);
220 assertTrue("expected IllegalArgumentIOException, got " + cause,
221 cause instanceof IllegalArgumentIOException);
222 }
223
224 public static Throwable getExceptionCause(final ProcedureInfo procInfo) {
225 assert procInfo.getForeignExceptionMessage() != null;
226 return RemoteProcedureException.fromProto(procInfo.getForeignExceptionMessage()).getCause();
227 }
228
229 public static class TestProcedure extends Procedure<Void> {
230 private byte[] data = null;
231
232 public TestProcedure() {}
233
234 public TestProcedure(long procId) {
235 this(procId, 0);
236 }
237
238 public TestProcedure(long procId, long parentId) {
239 this(procId, parentId, null);
240 }
241
242 public TestProcedure(long procId, long parentId, byte[] data) {
243 setData(data);
244 setProcId(procId);
245 if (parentId > 0) {
246 setParentProcId(parentId);
247 }
248 }
249
250 public void addStackId(final int index) {
251 addStackIndex(index);
252 }
253
254 public void setFinishedState() {
255 setState(ProcedureProtos.ProcedureState.FINISHED);
256 }
257
258 public void setData(final byte[] data) {
259 this.data = data;
260 }
261
262 @Override
263 protected Procedure[] execute(Void env) { return null; }
264
265 @Override
266 protected void rollback(Void env) { }
267
268 @Override
269 protected boolean abort(Void env) { return false; }
270
271 @Override
272 protected void serializeStateData(final OutputStream stream) throws IOException {
273 StreamUtils.writeRawVInt32(stream, data != null ? data.length : 0);
274 if (data != null) stream.write(data);
275 }
276
277 @Override
278 protected void deserializeStateData(final InputStream stream) throws IOException {
279 int len = StreamUtils.readRawVarint32(stream);
280 if (len > 0) {
281 data = new byte[len];
282 stream.read(data);
283 } else {
284 data = null;
285 }
286 }
287 }
288
289 public static class LoadCounter implements ProcedureStore.ProcedureLoader {
290 private final ArrayList<Procedure> corrupted = new ArrayList<Procedure>();
291 private final ArrayList<Procedure> loaded = new ArrayList<Procedure>();
292
293 private Set<Long> procIds;
294 private long maxProcId = 0;
295
296 public LoadCounter() {
297 this(null);
298 }
299
300 public LoadCounter(final Set<Long> procIds) {
301 this.procIds = procIds;
302 }
303
304 public void reset() {
305 reset(null);
306 }
307
308 public void reset(final Set<Long> procIds) {
309 corrupted.clear();
310 loaded.clear();
311 this.procIds = procIds;
312 this.maxProcId = 0;
313 }
314
315 public long getMaxProcId() {
316 return maxProcId;
317 }
318
319 public ArrayList<Procedure> getLoaded() {
320 return loaded;
321 }
322
323 public int getLoadedCount() {
324 return loaded.size();
325 }
326
327 public ArrayList<Procedure> getCorrupted() {
328 return corrupted;
329 }
330
331 public int getCorruptedCount() {
332 return corrupted.size();
333 }
334
335 @Override
336 public void setMaxProcId(long maxProcId) {
337 this.maxProcId = maxProcId;
338 }
339
340 @Override
341 public void load(ProcedureIterator procIter) throws IOException {
342 while (procIter.hasNext()) {
343 Procedure proc = procIter.nextAsProcedure();
344 LOG.debug("loading procId=" + proc.getProcId() + ": " + proc);
345 if (procIds != null) {
346 assertTrue("procId=" + proc.getProcId() + " unexpected",
347 procIds.contains(proc.getProcId()));
348 }
349 loaded.add(proc);
350 }
351 }
352
353 @Override
354 public void handleCorrupted(ProcedureIterator procIter) throws IOException {
355 while (procIter.hasNext()) {
356 Procedure proc = procIter.nextAsProcedure();
357 LOG.debug("corrupted procId=" + proc.getProcId() + ": " + proc);
358 corrupted.add(proc);
359 }
360 }
361 }
362 }