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.store.wal;
20
21 import java.io.IOException;
22 import java.util.Random;
23 import java.util.concurrent.atomic.AtomicLong;
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.FileStatus;
30 import org.apache.hadoop.fs.Path;
31 import org.apache.hadoop.hbase.HBaseCommonTestingUtility;
32 import org.apache.hadoop.hbase.procedure2.Procedure;
33 import org.apache.hadoop.hbase.procedure2.ProcedureTestingUtility;
34 import org.apache.hadoop.hbase.procedure2.ProcedureTestingUtility.LoadCounter;
35 import org.apache.hadoop.hbase.procedure2.ProcedureTestingUtility.TestProcedure;
36 import org.apache.hadoop.hbase.testclassification.LargeTests;
37 import org.apache.hadoop.hbase.testclassification.MasterTests;
38 import org.apache.hadoop.hbase.util.Bytes;
39 import org.apache.hadoop.io.IOUtils;
40
41 import org.junit.After;
42 import org.junit.Before;
43 import org.junit.Test;
44 import org.junit.experimental.categories.Category;
45
46 import static org.junit.Assert.assertEquals;
47 import static org.junit.Assert.assertFalse;
48 import static org.junit.Assert.assertTrue;
49 import static org.junit.Assert.fail;
50
51 @Category({MasterTests.class, LargeTests.class})
52 public class TestStressWALProcedureStore {
53 private static final Log LOG = LogFactory.getLog(TestWALProcedureStore.class);
54
55 private static final int PROCEDURE_STORE_SLOTS = 8;
56
57 private WALProcedureStore procStore;
58
59 private HBaseCommonTestingUtility htu;
60 private FileSystem fs;
61 private Path testDir;
62 private Path logDir;
63
64 private void setupConfiguration(Configuration conf) {
65 conf.setBoolean(WALProcedureStore.USE_HSYNC_CONF_KEY, false);
66 conf.setInt(WALProcedureStore.PERIODIC_ROLL_CONF_KEY, 5000);
67 conf.setInt(WALProcedureStore.ROLL_THRESHOLD_CONF_KEY, 128 * 1024);
68 }
69
70 @Before
71 public void setUp() throws IOException {
72 htu = new HBaseCommonTestingUtility();
73 setupConfiguration(htu.getConfiguration());
74
75 testDir = htu.getDataTestDir();
76 fs = testDir.getFileSystem(htu.getConfiguration());
77 assertTrue(testDir.depth() > 1);
78
79 logDir = new Path(testDir, "proc-logs");
80 procStore = ProcedureTestingUtility.createWalStore(htu.getConfiguration(), fs, logDir);
81 procStore.start(PROCEDURE_STORE_SLOTS);
82 procStore.recoverLease();
83
84 LoadCounter loader = new LoadCounter();
85 procStore.load(loader);
86 assertEquals(0, loader.getMaxProcId());
87 assertEquals(0, loader.getLoadedCount());
88 assertEquals(0, loader.getCorruptedCount());
89 }
90
91 @After
92 public void tearDown() throws IOException {
93 procStore.stop(false);
94 fs.delete(logDir, true);
95 }
96
97 @Test
98 public void testInsertUpdateDelete() throws Exception {
99 final long LAST_PROC_ID = 19999;
100 final Thread[] thread = new Thread[PROCEDURE_STORE_SLOTS];
101 final AtomicLong procCounter = new AtomicLong((long)Math.round(Math.random() * 100));
102 for (int i = 0; i < thread.length; ++i) {
103 thread[i] = new Thread() {
104 @Override
105 public void run() {
106 Random rand = new Random();
107 TestProcedure proc;
108 do {
109 proc = new TestProcedure(procCounter.addAndGet(1));
110
111 procStore.insert(proc, null);
112
113 for (int i = 0, nupdates = rand.nextInt(10); i <= nupdates; ++i) {
114 try { Thread.sleep(0, rand.nextInt(15)); } catch (InterruptedException e) {}
115 procStore.update(proc);
116 }
117
118 procStore.delete(proc.getProcId());
119 } while (proc.getProcId() < LAST_PROC_ID);
120 }
121 };
122 thread[i].start();
123 }
124
125 for (int i = 0; i < thread.length; ++i) {
126 thread[i].join();
127 }
128
129 procStore.getStoreTracker().dump();
130 assertTrue(procCounter.get() >= LAST_PROC_ID);
131 assertTrue(procStore.getStoreTracker().isEmpty());
132 assertEquals(1, procStore.getActiveLogs().size());
133 }
134 }