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.util.Bytes;
48 import org.apache.hadoop.hbase.util.FSUtils;
49 import org.junit.After;
50 import org.junit.AfterClass;
51 import org.junit.Before;
52 import org.junit.BeforeClass;
53 import org.junit.Rule;
54 import org.junit.Test;
55 import org.junit.experimental.categories.Category;
56 import org.junit.rules.TestName;
57
58
59 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
60
61 @Category(MediumTests.class)
62 public class TestDefaultWALProvider {
63 protected static final Log LOG = LogFactory.getLog(TestDefaultWALProvider.class);
64
65 protected static Configuration conf;
66 protected static FileSystem fs;
67 protected static FileSystem walFs;
68 protected static Path walRootDir;
69 protected final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
70
71 @Rule
72 public final TestName currentTest = new TestName();
73
74 @Before
75 public void setUp() throws Exception {
76 FileStatus[] entries = fs.listStatus(new Path("/"));
77 for (FileStatus dir : entries) {
78 fs.delete(dir.getPath(), true);
79 }
80 walFs.delete(walRootDir, true);
81 }
82
83 @After
84 public void tearDown() throws Exception {
85 }
86
87 @BeforeClass
88 public static void setUpBeforeClass() throws Exception {
89
90 TEST_UTIL.getConfiguration().setInt("dfs.blocksize", 1024 * 1024);
91
92 TEST_UTIL.getConfiguration().setInt("dfs.namenode.heartbeat.recheck-interval", 5000);
93 TEST_UTIL.getConfiguration().setInt("dfs.heartbeat.interval", 1);
94 TEST_UTIL.getConfiguration().setInt("dfs.client.socket-timeout", 5000);
95
96
97 TEST_UTIL.getConfiguration()
98 .setInt("hbase.ipc.client.connect.max.retries", 1);
99 TEST_UTIL.getConfiguration().setInt(
100 "dfs.client.block.recovery.retries", 1);
101 TEST_UTIL.getConfiguration().setInt(
102 "hbase.ipc.client.connection.maxidletime", 500);
103 TEST_UTIL.startMiniDFSCluster(3);
104
105
106 walRootDir = TEST_UTIL.createWALRootDir();
107 conf = TEST_UTIL.getConfiguration();
108 fs = FSUtils.getRootDirFileSystem(conf);
109 walFs = FSUtils.getWALFileSystem(conf);
110 }
111
112 @AfterClass
113 public static void tearDownAfterClass() throws Exception {
114 walFs.delete(walRootDir, true);
115 TEST_UTIL.shutdownMiniCluster();
116 }
117
118 static String getName() {
119 return "TestDefaultWALProvider";
120 }
121
122 @Test
123 public void testGetServerNameFromWALDirectoryName() throws IOException {
124 ServerName sn = ServerName.valueOf("hn", 450, 1398);
125 String hl = walRootDir + "/" +
126 DefaultWALProvider.getWALDirectoryName(sn.toString());
127
128
129 assertNull(DefaultWALProvider.getServerNameFromWALDirectoryName(conf, null));
130 assertNull(DefaultWALProvider.getServerNameFromWALDirectoryName(conf, walRootDir.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 walRootDir.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 walRootDir.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, AtomicLong sequenceId) 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, sequenceId, true, null);
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);
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, sequenceId);
209 log.rollWriter();
210 assertEquals(1, DefaultWALProvider.getNumRolledLogFiles(log));
211
212
213 addEdits(log, hri, htd, 2, sequenceId);
214 log.rollWriter();
215 assertEquals(2, DefaultWALProvider.getNumRolledLogFiles(log));
216
217
218 addEdits(log, hri, htd, 1, sequenceId);
219 addEdits(log, hri2, htd2, 1, sequenceId);
220 addEdits(log, hri, htd, 1, sequenceId);
221 addEdits(log, hri2, htd2, 1, sequenceId);
222 log.rollWriter();
223 assertEquals(3, DefaultWALProvider.getNumRolledLogFiles(log));
224
225
226
227 addEdits(log, hri2, htd2, 1, sequenceId);
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, sequenceId);
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 final AtomicLong sequenceId1 = new AtomicLong(1);
284 final AtomicLong sequenceId2 = new AtomicLong(1);
285
286 addEdits(wal, hri1, table1, 1, sequenceId1);
287 wal.rollWriter();
288
289 assertEquals(1, DefaultWALProvider.getNumRolledLogFiles(wal));
290
291 addEdits(wal, hri1, table1, 1, sequenceId1);
292 wal.rollWriter();
293
294 assertEquals(2, DefaultWALProvider.getNumRolledLogFiles(wal));
295
296 addEdits(wal, hri1, table1, 3, sequenceId1);
297 flushRegion(wal, hri1.getEncodedNameAsBytes(), table1.getFamiliesKeys());
298
299 wal.rollWriter();
300 assertEquals(0, DefaultWALProvider.getNumRolledLogFiles(wal));
301
302 addEdits(wal, hri2, table2, 1, sequenceId2);
303 wal.rollWriter();
304 assertEquals(1, DefaultWALProvider.getNumRolledLogFiles(wal));
305
306 addEdits(wal, hri1, table1, 2, sequenceId1);
307 wal.rollWriter();
308 assertEquals(2, DefaultWALProvider.getNumRolledLogFiles(wal));
309
310 addEdits(wal, hri2, table2, 2, sequenceId2);
311 flushRegion(wal, hri1.getEncodedNameAsBytes(), table2.getFamiliesKeys());
312
313
314
315
316
317 wal.rollWriter();
318 assertEquals(2, DefaultWALProvider.getNumRolledLogFiles(wal));
319
320 addEdits(wal, hri2, table2, 2, sequenceId2);
321 flushRegion(wal, hri2.getEncodedNameAsBytes(), table2.getFamiliesKeys());
322 wal.rollWriter();
323 assertEquals(0, DefaultWALProvider.getNumRolledLogFiles(wal));
324 } finally {
325 if (wals != null) {
326 wals.close();
327 }
328 }
329 }
330
331
332
333
334
335 @Test
336 public void testConcurrentWrites() throws Exception {
337
338
339 int errCode = WALPerformanceEvaluation.
340 innerMain(new Configuration(TEST_UTIL.getConfiguration()),
341 new String [] {"-threads", "3", "-verify", "-noclosefs", "-iterations", "3000"});
342 assertEquals(0, errCode);
343 }
344
345
346
347
348 @Test
349 public void setMembershipDedups() throws IOException {
350 final Configuration localConf = new Configuration(conf);
351 localConf.set(WALFactory.WAL_PROVIDER, DefaultWALProvider.class.getName());
352 final WALFactory wals = new WALFactory(localConf, null, currentTest.getMethodName());
353 try {
354 final Set<WAL> seen = new HashSet<WAL>(1);
355 final Random random = new Random();
356 assertTrue("first attempt to add WAL from default provider should work.",
357 seen.add(wals.getWAL(Bytes.toBytes(random.nextInt()))));
358 for (int i = 0; i < 1000; i++) {
359 assertFalse("default wal provider is only supposed to return a single wal, which should " +
360 "compare as .equals itself.", seen.add(wals.getWAL(Bytes.toBytes(random.nextInt()))));
361 }
362 } finally {
363 wals.close();
364 }
365 }
366 }