View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
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          // no-op
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      // stop
81      procExecutor.stop();
82      procExecutor.join();
83      procStore.stop(false);
84      // nothing running...
85      if (beforeStartAction != null) {
86        beforeStartAction.run();
87      }
88      // re-start
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 }