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.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
91 final NonceKey nonceKey = procExecutor.createNonceKey(nonceGroup, nonce);
92 assertFalse(procExecutor.registerNonce(nonceKey) >= 0);
93
94
95 Procedure proc = new TestSingleStepProcedure();
96 long procId = procExecutor.submitProcedure(proc, nonceKey);
97 ProcedureTestingUtility.waitProcedure(procExecutor, procId);
98
99
100 ProcedureTestingUtility.restart(procExecutor);
101 ProcedureTestingUtility.waitProcedure(procExecutor, procId);
102
103
104
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
117 final NonceKey nonceKey = procExecutor.createNonceKey(nonceGroup, nonce);
118 assertFalse(procExecutor.registerNonce(nonceKey) >= 0);
119
120
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
128
129 assertEquals(procId, procExecutor.registerNonce(nonceKey));
130
131
132 latch.countDown();
133
134
135 ProcedureTestingUtility.restart(procExecutor);
136 ProcedureTestingUtility.waitProcedure(procExecutor, procId);
137
138
139
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
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
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
189 assertFalse("unexpected already registered nonce",
190 procExecutor.registerNonce(nonceKey) >= 0);
191 t1NonceRegisteredLatch.countDown();
192
193
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
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
224 t1NonceRegisteredLatch.await();
225
226
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
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 }