1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.wal;
20
21 import static org.junit.Assert.assertEquals;
22 import static org.junit.Assert.assertFalse;
23 import static org.junit.Assert.assertNull;
24 import static org.junit.Assert.assertTrue;
25
26 import java.io.IOException;
27 import java.util.HashSet;
28 import java.util.Random;
29 import java.util.Set;
30 import java.util.concurrent.atomic.AtomicLong;
31
32 import org.apache.commons.logging.Log;
33 import org.apache.commons.logging.LogFactory;
34 import org.apache.hadoop.conf.Configuration;
35 import org.apache.hadoop.fs.FileStatus;
36 import org.apache.hadoop.fs.FileSystem;
37 import org.apache.hadoop.fs.Path;
38 import org.apache.hadoop.hbase.HBaseTestingUtility;
39 import org.apache.hadoop.hbase.HColumnDescriptor;
40 import org.apache.hadoop.hbase.HConstants;
41 import org.apache.hadoop.hbase.HRegionInfo;
42 import org.apache.hadoop.hbase.HTableDescriptor;
43 import org.apache.hadoop.hbase.KeyValue;
44 import org.apache.hadoop.hbase.testclassification.MediumTests;
45 import org.apache.hadoop.hbase.ServerName;
46 import org.apache.hadoop.hbase.TableName;
47 import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl;
48 import org.apache.hadoop.hbase.testclassification.MediumTests;
49 import org.apache.hadoop.hbase.util.Bytes;
50 import org.apache.hadoop.hbase.util.FSUtils;
51 import org.junit.After;
52 import org.junit.AfterClass;
53 import org.junit.Before;
54 import org.junit.BeforeClass;
55 import org.junit.Rule;
56 import org.junit.Test;
57 import org.junit.experimental.categories.Category;
58 import org.junit.rules.TestName;
59
60
61 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
62
63 @Category(MediumTests.class)
64 public class TestDefaultWALProvider {
65 private static final Log LOG = LogFactory.getLog(TestDefaultWALProvider.class);
66
67 protected static Configuration conf;
68 protected static FileSystem fs;
69 protected final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
70 protected MultiVersionConcurrencyControl mvcc;
71
72 @Rule
73 public final TestName currentTest = new TestName();
74
75 @Before
76 public void setUp() throws Exception {
77 mvcc = new MultiVersionConcurrencyControl();
78 FileStatus[] entries = fs.listStatus(new Path("/"));
79 for (FileStatus dir : entries) {
80 fs.delete(dir.getPath(), true);
81 }
82 }
83
84 @After
85 public void tearDown() throws Exception {
86 }
87
88 @BeforeClass
89 public static void setUpBeforeClass() throws Exception {
90
91 TEST_UTIL.getConfiguration().setInt("dfs.blocksize", 1024 * 1024);
92
93 TEST_UTIL.getConfiguration().setInt("dfs.namenode.heartbeat.recheck-interval", 5000);
94 TEST_UTIL.getConfiguration().setInt("dfs.heartbeat.interval", 1);
95 TEST_UTIL.getConfiguration().setInt("dfs.client.socket-timeout", 5000);
96
97
98 TEST_UTIL.getConfiguration()
99 .setInt("hbase.ipc.client.connect.max.retries", 1);
100 TEST_UTIL.getConfiguration().setInt(
101 "dfs.client.block.recovery.retries", 1);
102 TEST_UTIL.getConfiguration().setInt(
103 "hbase.ipc.client.connection.maxidletime", 500);
104 TEST_UTIL.startMiniDFSCluster(3);
105
106
107 TEST_UTIL.createRootDir();
108 conf = TEST_UTIL.getConfiguration();
109 fs = TEST_UTIL.getDFSCluster().getFileSystem();
110 }
111
112 @AfterClass
113 public static void tearDownAfterClass() throws Exception {
114 TEST_UTIL.shutdownMiniCluster();
115 }
116
117 static String getName() {
118 return "TestDefaultWALProvider";
119 }
120
121 @Test
122 public void testGetServerNameFromWALDirectoryName() throws IOException {
123 ServerName sn = ServerName.valueOf("hn", 450, 1398);
124 String hl = FSUtils.getRootDir(conf) + "/" +
125 DefaultWALProvider.getWALDirectoryName(sn.toString());
126
127
128 assertNull(DefaultWALProvider.getServerNameFromWALDirectoryName(conf, null));
129 assertNull(DefaultWALProvider.getServerNameFromWALDirectoryName(conf,
130 FSUtils.getRootDir(conf).toUri().toString()));
131 assertNull(DefaultWALProvider.getServerNameFromWALDirectoryName(conf, ""));
132 assertNull(DefaultWALProvider.getServerNameFromWALDirectoryName(conf, " "));
133 assertNull(DefaultWALProvider.getServerNameFromWALDirectoryName(conf, hl));
134 assertNull(DefaultWALProvider.getServerNameFromWALDirectoryName(conf, hl + "qdf"));
135 assertNull(DefaultWALProvider.getServerNameFromWALDirectoryName(conf, "sfqf" + hl + "qdf"));
136
137 final String wals = "/WALs/";
138 ServerName parsed = DefaultWALProvider.getServerNameFromWALDirectoryName(conf,
139 FSUtils.getRootDir(conf).toUri().toString() + wals + sn +
140 "/localhost%2C32984%2C1343316388997.1343316390417");
141 assertEquals("standard", sn, parsed);
142
143 parsed = DefaultWALProvider.getServerNameFromWALDirectoryName(conf, hl + "/qdf");
144 assertEquals("subdir", sn, parsed);
145
146 parsed = DefaultWALProvider.getServerNameFromWALDirectoryName(conf,
147 FSUtils.getRootDir(conf).toUri().toString() + wals + sn +
148 "-splitting/localhost%3A57020.1340474893931");
149 assertEquals("split", sn, parsed);
150 }
151
152
153 protected void addEdits(WAL log, HRegionInfo hri, HTableDescriptor htd,
154 int times) throws IOException {
155 final byte[] row = Bytes.toBytes("row");
156 for (int i = 0; i < times; i++) {
157 long timestamp = System.currentTimeMillis();
158 WALEdit cols = new WALEdit();
159 cols.add(new KeyValue(row, row, row, timestamp, row));
160 log.append(htd, hri, getWalKey(hri.getEncodedNameAsBytes(), htd.getTableName(), timestamp),
161 cols, true);
162 }
163 log.sync();
164 }
165
166
167
168
169 WALKey getWalKey(final byte[] info, final TableName tableName, final long timestamp) {
170 return new WALKey(info, tableName, timestamp, mvcc);
171 }
172
173
174
175
176
177
178 protected void flushRegion(WAL wal, byte[] regionEncodedName, Set<byte[]> flushedFamilyNames) {
179 wal.startCacheFlush(regionEncodedName, flushedFamilyNames);
180 wal.completeCacheFlush(regionEncodedName);
181 }
182
183 private static final byte[] UNSPECIFIED_REGION = new byte[]{};
184
185 @Test
186 public void testLogCleaning() throws Exception {
187 LOG.info("testLogCleaning");
188 final HTableDescriptor htd =
189 new HTableDescriptor(TableName.valueOf("testLogCleaning")).addFamily(new HColumnDescriptor(
190 "row"));
191 final HTableDescriptor htd2 =
192 new HTableDescriptor(TableName.valueOf("testLogCleaning2"))
193 .addFamily(new HColumnDescriptor("row"));
194 final Configuration localConf = new Configuration(conf);
195 localConf.set(WALFactory.WAL_PROVIDER, DefaultWALProvider.class.getName());
196 final WALFactory wals = new WALFactory(localConf, null, currentTest.getMethodName());
197 final AtomicLong sequenceId = new AtomicLong(1);
198 try {
199 HRegionInfo hri = new HRegionInfo(htd.getTableName(),
200 HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW);
201 HRegionInfo hri2 = new HRegionInfo(htd2.getTableName(),
202 HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW);
203
204 final WAL log = wals.getWAL(UNSPECIFIED_REGION);
205
206
207
208 addEdits(log, hri, htd, 1);
209 log.rollWriter();
210 assertEquals(1, DefaultWALProvider.getNumRolledLogFiles(log));
211
212
213 addEdits(log, hri, htd, 2);
214 log.rollWriter();
215 assertEquals(2, DefaultWALProvider.getNumRolledLogFiles(log));
216
217
218 addEdits(log, hri, htd, 1);
219 addEdits(log, hri2, htd2, 1);
220 addEdits(log, hri, htd, 1);
221 addEdits(log, hri2, htd2, 1);
222 log.rollWriter();
223 assertEquals(3, DefaultWALProvider.getNumRolledLogFiles(log));
224
225
226
227 addEdits(log, hri2, htd2, 1);
228 log.startCacheFlush(hri.getEncodedNameAsBytes(), htd.getFamiliesKeys());
229 log.completeCacheFlush(hri.getEncodedNameAsBytes());
230 log.rollWriter();
231 assertEquals(2, DefaultWALProvider.getNumRolledLogFiles(log));
232
233
234
235
236 addEdits(log, hri2, htd2, 1);
237 log.startCacheFlush(hri2.getEncodedNameAsBytes(), htd2.getFamiliesKeys());
238 log.completeCacheFlush(hri2.getEncodedNameAsBytes());
239 log.rollWriter();
240 assertEquals(0, DefaultWALProvider.getNumRolledLogFiles(log));
241 } finally {
242 if (wals != null) {
243 wals.close();
244 }
245 }
246 }
247
248
249
250
251
252
253
254
255
256
257
258
259
260 @Test
261 public void testWALArchiving() throws IOException {
262 LOG.debug("testWALArchiving");
263 HTableDescriptor table1 =
264 new HTableDescriptor(TableName.valueOf("t1")).addFamily(new HColumnDescriptor("row"));
265 HTableDescriptor table2 =
266 new HTableDescriptor(TableName.valueOf("t2")).addFamily(new HColumnDescriptor("row"));
267 final Configuration localConf = new Configuration(conf);
268 localConf.set(WALFactory.WAL_PROVIDER, DefaultWALProvider.class.getName());
269 final WALFactory wals = new WALFactory(localConf, null, currentTest.getMethodName());
270 try {
271 final WAL wal = wals.getWAL(UNSPECIFIED_REGION);
272 assertEquals(0, DefaultWALProvider.getNumRolledLogFiles(wal));
273 HRegionInfo hri1 =
274 new HRegionInfo(table1.getTableName(), HConstants.EMPTY_START_ROW,
275 HConstants.EMPTY_END_ROW);
276 HRegionInfo hri2 =
277 new HRegionInfo(table2.getTableName(), HConstants.EMPTY_START_ROW,
278 HConstants.EMPTY_END_ROW);
279
280 hri1.setSplit(false);
281 hri2.setSplit(false);
282
283
284 addEdits(wal, hri1, table1, 1);
285 wal.rollWriter();
286
287 assertEquals(1, DefaultWALProvider.getNumRolledLogFiles(wal));
288
289 addEdits(wal, hri1, table1, 1);
290 wal.rollWriter();
291
292 assertEquals(2, DefaultWALProvider.getNumRolledLogFiles(wal));
293
294 addEdits(wal, hri1, table1, 3);
295 flushRegion(wal, hri1.getEncodedNameAsBytes(), table1.getFamiliesKeys());
296
297 wal.rollWriter();
298 assertEquals(0, DefaultWALProvider.getNumRolledLogFiles(wal));
299
300 addEdits(wal, hri2, table2, 1);
301 wal.rollWriter();
302 assertEquals(1, DefaultWALProvider.getNumRolledLogFiles(wal));
303
304 addEdits(wal, hri1, table1, 2);
305 wal.rollWriter();
306 assertEquals(2, DefaultWALProvider.getNumRolledLogFiles(wal));
307
308 addEdits(wal, hri2, table2, 2);
309 flushRegion(wal, hri1.getEncodedNameAsBytes(), table2.getFamiliesKeys());
310
311
312
313
314
315 wal.rollWriter();
316 assertEquals(2, DefaultWALProvider.getNumRolledLogFiles(wal));
317
318 addEdits(wal, hri2, table2, 2);
319 flushRegion(wal, hri2.getEncodedNameAsBytes(), table2.getFamiliesKeys());
320 wal.rollWriter();
321 assertEquals(0, DefaultWALProvider.getNumRolledLogFiles(wal));
322 } finally {
323 if (wals != null) {
324 wals.close();
325 }
326 }
327 }
328
329
330
331
332
333 @Test
334 public void testConcurrentWrites() throws Exception {
335
336
337 int errCode = WALPerformanceEvaluation.
338 innerMain(new Configuration(TEST_UTIL.getConfiguration()),
339 new String [] {"-threads", "3", "-verify", "-noclosefs", "-iterations", "3000"});
340 assertEquals(0, errCode);
341 }
342
343
344
345
346 @Test
347 public void setMembershipDedups() throws IOException {
348 final Configuration localConf = new Configuration(conf);
349 localConf.set(WALFactory.WAL_PROVIDER, DefaultWALProvider.class.getName());
350 final WALFactory wals = new WALFactory(localConf, null, currentTest.getMethodName());
351 try {
352 final Set<WAL> seen = new HashSet<WAL>(1);
353 final Random random = new Random();
354 assertTrue("first attempt to add WAL from default provider should work.",
355 seen.add(wals.getWAL(Bytes.toBytes(random.nextInt()))));
356 for (int i = 0; i < 1000; i++) {
357 assertFalse("default wal provider is only supposed to return a single wal, which should " +
358 "compare as .equals itself.", seen.add(wals.getWAL(Bytes.toBytes(random.nextInt()))));
359 }
360 } finally {
361 wals.close();
362 }
363 }
364 }