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.util.concurrent.atomic.AtomicReference;
23  import java.util.concurrent.CountDownLatch;
24  
25  import org.apache.commons.logging.Log;
26  import org.apache.commons.logging.LogFactory;
27  import org.apache.hadoop.fs.FileSystem;
28  import org.apache.hadoop.fs.Path;
29  import org.apache.hadoop.hbase.HBaseCommonTestingUtility;
30  import org.apache.hadoop.hbase.ProcedureInfo;
31  import org.apache.hadoop.hbase.procedure2.store.ProcedureStore;
32  import org.apache.hadoop.hbase.testclassification.SmallTests;
33  import org.apache.hadoop.hbase.testclassification.MasterTests;
34  import org.apache.hadoop.hbase.security.User;
35  import org.apache.hadoop.hbase.util.Bytes;
36  import org.apache.hadoop.hbase.util.NonceKey;
37  import org.apache.hadoop.hbase.util.Threads;
38  
39  import org.junit.After;
40  import org.junit.Before;
41  import org.junit.Test;
42  import org.junit.experimental.categories.Category;
43  
44  import static org.junit.Assert.assertEquals;
45  import static org.junit.Assert.assertFalse;
46  import static org.junit.Assert.assertTrue;
47  
48  @Category({MasterTests.class, SmallTests.class})
49  public class TestProcedureNonce {
50    private static final Log LOG = LogFactory.getLog(TestProcedureNonce.class);
51  
52    private static final int PROCEDURE_EXECUTOR_SLOTS = 2;
53  
54    private static TestProcEnv procEnv;
55    private static ProcedureExecutor<TestProcEnv> procExecutor;
56    private static ProcedureStore procStore;
57  
58    private HBaseCommonTestingUtility htu;
59    private FileSystem fs;
60    private Path logDir;
61  
62    @Before
63    public void setUp() throws IOException {
64      htu = new HBaseCommonTestingUtility();
65      Path testDir = htu.getDataTestDir();
66      fs = testDir.getFileSystem(htu.getConfiguration());
67      assertTrue(testDir.depth() > 1);
68  
69      logDir = new Path(testDir, "proc-logs");
70      procEnv = new TestProcEnv();
71      procStore = ProcedureTestingUtility.createStore(htu.getConfiguration(), fs, logDir);
72      procExecutor = new ProcedureExecutor(htu.getConfiguration(), procEnv, procStore);
73      procExecutor.testing = new ProcedureExecutor.Testing();
74      procStore.start(PROCEDURE_EXECUTOR_SLOTS);
75      procExecutor.start(PROCEDURE_EXECUTOR_SLOTS, true);
76    }
77  
78    @After
79    public void tearDown() throws IOException {
80      procExecutor.stop();
81      procStore.stop(false);
82      fs.delete(logDir, true);
83    }
84  
85    @Test(timeout=30000)
86    public void testCompletedProcWithSameNonce() throws Exception {
87      final long nonceGroup = 123;
88      final long nonce = 2222;
89  
90      // register the nonce
91      final NonceKey nonceKey = procExecutor.createNonceKey(nonceGroup, nonce);
92      assertFalse(procExecutor.registerNonce(nonceKey) >= 0);
93  
94      // Submit a proc and wait for its completion
95      Procedure proc = new TestSingleStepProcedure();
96      long procId = procExecutor.submitProcedure(proc, nonceKey);
97      ProcedureTestingUtility.waitProcedure(procExecutor, procId);
98  
99      // Restart
100     ProcedureTestingUtility.restart(procExecutor);
101     ProcedureTestingUtility.waitProcedure(procExecutor, procId);
102 
103     // try to register a procedure with the same nonce
104     // we should get back the old procId
105     assertEquals(procId, procExecutor.registerNonce(nonceKey));
106 
107     ProcedureInfo result = procExecutor.getResult(procId);
108     ProcedureTestingUtility.assertProcNotFailed(result);
109   }
110 
111   @Test(timeout=30000)
112   public void testRunningProcWithSameNonce() throws Exception {
113     final long nonceGroup = 456;
114     final long nonce = 33333;
115 
116     // register the nonce
117     final NonceKey nonceKey = procExecutor.createNonceKey(nonceGroup, nonce);
118     assertFalse(procExecutor.registerNonce(nonceKey) >= 0);
119 
120     // Submit a proc and use a latch to prevent the step execution until we submitted proc2
121     CountDownLatch latch = new CountDownLatch(1);
122     TestSingleStepProcedure proc = new TestSingleStepProcedure();
123     procEnv.setWaitLatch(latch);
124     long procId = procExecutor.submitProcedure(proc, nonceKey);
125     while (proc.step != 1) Threads.sleep(25);
126 
127     // try to register a procedure with the same nonce
128     // we should get back the old procId
129     assertEquals(procId, procExecutor.registerNonce(nonceKey));
130 
131     // complete the procedure
132     latch.countDown();
133 
134     // Restart, the procedure is not completed yet
135     ProcedureTestingUtility.restart(procExecutor);
136     ProcedureTestingUtility.waitProcedure(procExecutor, procId);
137 
138     // try to register a procedure with the same nonce
139     // we should get back the old procId
140     assertEquals(procId, procExecutor.registerNonce(nonceKey));
141 
142     ProcedureInfo result = procExecutor.getResult(procId);
143     ProcedureTestingUtility.assertProcNotFailed(result);
144   }
145 
146   @Test
147   public void testSetFailureResultForNonce() throws IOException {
148     final long nonceGroup = 234;
149     final long nonce = 55555;
150 
151     // check and register the request nonce
152     final NonceKey nonceKey = procExecutor.createNonceKey(nonceGroup, nonce);
153     assertFalse(procExecutor.registerNonce(nonceKey) >= 0);
154 
155     procExecutor.setFailureResultForNonce(nonceKey, "testProc", User.getCurrent(),
156       new IOException("test failure"));
157 
158     final long procId = procExecutor.registerNonce(nonceKey);
159     ProcedureInfo result = procExecutor.getResult(procId);
160     ProcedureTestingUtility.assertProcFailed(result);
161   }
162 
163   @Test(timeout=30000)
164   public void testConcurrentNonceRegistration() throws IOException {
165     testConcurrentNonceRegistration(true, 567, 44444);
166   }
167 
168   @Test(timeout=30000)
169   public void testConcurrentNonceRegistrationWithRollback() throws IOException {
170     testConcurrentNonceRegistration(false, 890, 55555);
171   }
172 
173   private void testConcurrentNonceRegistration(final boolean submitProcedure,
174       final long nonceGroup, final long nonce) throws IOException {
175     // register the nonce
176     final NonceKey nonceKey = procExecutor.createNonceKey(nonceGroup, nonce);
177 
178     final AtomicReference<Throwable> t1Exception = new AtomicReference();
179     final AtomicReference<Throwable> t2Exception = new AtomicReference();
180 
181     final CountDownLatch t1NonceRegisteredLatch = new CountDownLatch(1);
182     final CountDownLatch t2BeforeNonceRegisteredLatch = new CountDownLatch(1);
183     final Thread[] threads = new Thread[2];
184     threads[0] = new Thread() {
185       @Override
186       public void run() {
187         try {
188           // release the nonce and wake t2
189           assertFalse("unexpected already registered nonce",
190             procExecutor.registerNonce(nonceKey) >= 0);
191           t1NonceRegisteredLatch.countDown();
192 
193           // hold the submission until t2 is registering the nonce
194           t2BeforeNonceRegisteredLatch.await();
195           Threads.sleep(1000);
196 
197           if (submitProcedure) {
198             CountDownLatch latch = new CountDownLatch(1);
199             TestSingleStepProcedure proc = new TestSingleStepProcedure();
200             procEnv.setWaitLatch(latch);
201 
202             procExecutor.submitProcedure(proc, nonceKey);
203             Threads.sleep(100);
204 
205             // complete the procedure
206             latch.countDown();
207           } else {
208             procExecutor.unregisterNonceIfProcedureWasNotSubmitted(nonceKey);
209           }
210         } catch (Throwable e) {
211           t1Exception.set(e);
212         } finally {
213           t1NonceRegisteredLatch.countDown();
214           t2BeforeNonceRegisteredLatch.countDown();
215         }
216       }
217     };
218 
219     threads[1] = new Thread() {
220       @Override
221       public void run() {
222         try {
223           // wait until t1 has registered the nonce
224           t1NonceRegisteredLatch.await();
225 
226           // register the nonce
227           t2BeforeNonceRegisteredLatch.countDown();
228           assertFalse("unexpected non registered nonce",
229             procExecutor.registerNonce(nonceKey) < 0);
230         } catch (Throwable e) {
231           t2Exception.set(e);
232         } finally {
233           t1NonceRegisteredLatch.countDown();
234           t2BeforeNonceRegisteredLatch.countDown();
235         }
236       }
237     };
238 
239     for (int i = 0; i < threads.length; ++i) threads[i].start();
240     for (int i = 0; i < threads.length; ++i) Threads.shutdown(threads[i]);
241     ProcedureTestingUtility.waitNoProcedureRunning(procExecutor);
242     assertEquals(null, t1Exception.get());
243     assertEquals(null, t2Exception.get());
244   }
245 
246   public static class TestSingleStepProcedure extends SequentialProcedure<TestProcEnv> {
247     private int step = 0;
248 
249     public TestSingleStepProcedure() { }
250 
251     @Override
252     protected Procedure[] execute(TestProcEnv env) throws InterruptedException {
253       step++;
254       env.waitOnLatch();
255       LOG.debug("execute procedure " + this + " step=" + step);
256       step++;
257       setResult(Bytes.toBytes(step));
258       return null;
259     }
260 
261     @Override
262     protected void rollback(TestProcEnv env) { }
263 
264     @Override
265     protected boolean abort(TestProcEnv env) { return true; }
266   }
267 
268   private static class TestProcEnv {
269     private CountDownLatch latch = null;
270 
271     /**
272      * set/unset a latch. every procedure execute() step will wait on the latch if any.
273      */
274     public void setWaitLatch(CountDownLatch latch) {
275       this.latch = latch;
276     }
277 
278     public void waitOnLatch() throws InterruptedException {
279       if (latch != null) {
280         latch.await();
281       }
282     }
283   }
284 }