1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.regionserver.wal;
20
21 import static org.junit.Assert.assertEquals;
22 import static org.junit.Assert.assertFalse;
23 import static org.junit.Assert.assertNotNull;
24 import static org.junit.Assert.assertTrue;
25
26 import java.io.EOFException;
27 import java.io.IOException;
28 import java.util.ArrayList;
29 import java.util.HashSet;
30 import java.util.List;
31 import java.util.Set;
32 import java.util.concurrent.atomic.AtomicBoolean;
33
34 import org.apache.commons.logging.Log;
35 import org.apache.commons.logging.LogFactory;
36 import org.apache.hadoop.conf.Configuration;
37 import org.apache.hadoop.fs.FileSystem;
38 import org.apache.hadoop.fs.Path;
39 import org.apache.hadoop.hbase.Cell;
40 import org.apache.hadoop.hbase.HBaseTestingUtility;
41 import org.apache.hadoop.hbase.HColumnDescriptor;
42 import org.apache.hadoop.hbase.HConstants;
43 import org.apache.hadoop.hbase.HRegionInfo;
44 import org.apache.hadoop.hbase.HTableDescriptor;
45 import org.apache.hadoop.hbase.testclassification.LargeTests;
46 import org.apache.hadoop.hbase.MiniHBaseCluster;
47 import org.apache.hadoop.hbase.ServerName;
48 import org.apache.hadoop.hbase.TableName;
49 import org.apache.hadoop.hbase.client.Admin;
50 import org.apache.hadoop.hbase.client.Get;
51 import org.apache.hadoop.hbase.client.HTable;
52 import org.apache.hadoop.hbase.client.Put;
53 import org.apache.hadoop.hbase.client.Result;
54 import org.apache.hadoop.hbase.client.ResultScanner;
55 import org.apache.hadoop.hbase.client.Scan;
56 import org.apache.hadoop.hbase.client.Table;
57 import org.apache.hadoop.hbase.fs.HFileSystem;
58 import org.apache.hadoop.hbase.regionserver.HRegionServer;
59 import org.apache.hadoop.hbase.regionserver.Region;
60 import org.apache.hadoop.hbase.regionserver.Store;
61 import org.apache.hadoop.hbase.util.Bytes;
62 import org.apache.hadoop.hbase.util.FSUtils;
63 import org.apache.hadoop.hbase.util.JVMClusterUtil;
64 import org.apache.hadoop.hbase.util.Threads;
65 import org.apache.hadoop.hbase.wal.DefaultWALProvider;
66 import org.apache.hadoop.hbase.wal.WAL;
67 import org.apache.hadoop.hbase.wal.WALFactory;
68 import org.apache.hadoop.hdfs.MiniDFSCluster;
69 import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
70 import org.apache.hadoop.hdfs.server.datanode.DataNode;
71 import org.junit.After;
72 import org.junit.Assert;
73 import org.junit.Before;
74 import org.junit.BeforeClass;
75 import org.junit.Rule;
76 import org.junit.Test;
77 import org.junit.experimental.categories.Category;
78 import org.junit.rules.TestName;
79
80
81
82
83 @Category(LargeTests.class)
84 public class TestLogRolling {
85 private static final Log LOG = LogFactory.getLog(TestLogRolling.class);
86 private HRegionServer server;
87 private String tableName;
88 private byte[] value;
89 private FileSystem fs;
90 private MiniDFSCluster dfsCluster;
91 private Admin admin;
92 private MiniHBaseCluster cluster;
93 private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
94 @Rule public final TestName name = new TestName();
95
96 public TestLogRolling() {
97 this.server = null;
98 this.tableName = null;
99
100 String className = this.getClass().getName();
101 StringBuilder v = new StringBuilder(className);
102 while (v.length() < 1000) {
103 v.append(className);
104 }
105 this.value = Bytes.toBytes(v.toString());
106 }
107
108
109
110 @BeforeClass
111 public static void setUpBeforeClass() throws Exception {
112
113
114 System.setProperty("hbase.tests.use.shortcircuit.reads", "false");
115
116
117
118 TEST_UTIL.getConfiguration().setLong(HConstants.HREGION_MAX_FILESIZE, 768L * 1024L);
119
120
121 TEST_UTIL.getConfiguration().setInt("hbase.regionserver.maxlogentries", 32);
122
123 TEST_UTIL.getConfiguration().setInt("hbase.regionserver.logroll.errors.tolerated", 2);
124 TEST_UTIL.getConfiguration().setInt("hbase.rpc.timeout", 10 * 1000);
125
126
127 TEST_UTIL.getConfiguration().setInt("hbase.hregion.memstore.optionalflushcount", 2);
128
129
130 TEST_UTIL.getConfiguration().setInt(
131 HConstants.HREGION_MEMSTORE_FLUSH_SIZE, 8192);
132
133
134 TEST_UTIL.getConfiguration().setLong("hbase.client.pause", 10 * 1000);
135
136
137
138 TEST_UTIL.getConfiguration().setInt(HConstants.THREAD_WAKE_FREQUENCY, 2 * 1000);
139
140
141
142 TEST_UTIL.getConfiguration().setBoolean("dfs.support.append", true);
143
144
145 TEST_UTIL.getConfiguration().setInt("dfs.namenode.heartbeat.recheck-interval", 5000);
146 TEST_UTIL.getConfiguration().setInt("dfs.heartbeat.interval", 1);
147
148
149 TEST_UTIL.getConfiguration().setInt("dfs.client.block.write.retries", 30);
150 TEST_UTIL.getConfiguration().setInt("hbase.regionserver.hlog.tolerable.lowreplication", 2);
151 TEST_UTIL.getConfiguration().setInt("hbase.regionserver.hlog.lowreplication.rolllimit", 3);
152 }
153
154 @Before
155 public void setUp() throws Exception {
156 TEST_UTIL.startMiniCluster(1, 1, 2);
157
158 cluster = TEST_UTIL.getHBaseCluster();
159 dfsCluster = TEST_UTIL.getDFSCluster();
160 fs = TEST_UTIL.getTestFileSystem();
161 admin = TEST_UTIL.getHBaseAdmin();
162
163
164 cluster.getMaster().balanceSwitch(false);
165 }
166
167 @After
168 public void tearDown() throws Exception {
169 TEST_UTIL.shutdownMiniCluster();
170 }
171
172 private void startAndWriteData() throws IOException, InterruptedException {
173
174 new HTable(TEST_UTIL.getConfiguration(), TableName.META_TABLE_NAME);
175 this.server = cluster.getRegionServerThreads().get(0).getRegionServer();
176
177 Table table = createTestTable(this.tableName);
178
179 server = TEST_UTIL.getRSForFirstRegionInTable(table.getName());
180 for (int i = 1; i <= 256; i++) {
181 doPut(table, i);
182 if (i % 32 == 0) {
183
184 try {
185 Thread.sleep(2000);
186 } catch (InterruptedException e) {
187
188 }
189 }
190 }
191 }
192
193
194
195
196 @Test(timeout=120000)
197 public void testLogRollOnNothingWritten() throws Exception {
198 final Configuration conf = TEST_UTIL.getConfiguration();
199 final WALFactory wals = new WALFactory(conf, null,
200 ServerName.valueOf("test.com",8080, 1).toString());
201 final WAL newLog = wals.getWAL(new byte[]{}, null);
202 try {
203
204 newLog.rollWriter(true);
205 } finally {
206 wals.close();
207 }
208 }
209
210
211
212
213
214
215 @Test
216 public void testLogRolling() throws Exception {
217 this.tableName = getName();
218
219 startAndWriteData();
220 HRegionInfo region =
221 server.getOnlineRegions(TableName.valueOf(tableName)).get(0).getRegionInfo();
222 final WAL log = server.getWAL(region);
223 LOG.info("after writing there are " + DefaultWALProvider.getNumRolledLogFiles(log) +
224 " log files");
225
226
227 for (Region r: server.getOnlineRegionsLocalContext()) {
228 r.flush(true);
229 }
230
231
232 log.rollWriter();
233
234 int count = DefaultWALProvider.getNumRolledLogFiles(log);
235 LOG.info("after flushing all regions and rolling logs there are " + count + " log files");
236 assertTrue(("actual count: " + count), count <= 2);
237 }
238
239 private String getName() {
240 return "TestLogRolling-" + name.getMethodName();
241 }
242
243 void writeData(Table table, int rownum) throws IOException {
244 doPut(table, rownum);
245
246
247 try {
248 Thread.sleep(2000);
249 } catch (InterruptedException e) {
250
251 }
252 }
253
254 void validateData(Table table, int rownum) throws IOException {
255 String row = "row" + String.format("%1$04d", rownum);
256 Get get = new Get(Bytes.toBytes(row));
257 get.addFamily(HConstants.CATALOG_FAMILY);
258 Result result = table.get(get);
259 assertTrue(result.size() == 1);
260 assertTrue(Bytes.equals(value,
261 result.getValue(HConstants.CATALOG_FAMILY, null)));
262 LOG.info("Validated row " + row);
263 }
264
265 void batchWriteAndWait(Table table, final FSHLog log, int start, boolean expect, int timeout)
266 throws IOException {
267 for (int i = 0; i < 10; i++) {
268 Put put = new Put(Bytes.toBytes("row"
269 + String.format("%1$04d", (start + i))));
270 put.add(HConstants.CATALOG_FAMILY, null, value);
271 table.put(put);
272 }
273 Put tmpPut = new Put(Bytes.toBytes("tmprow"));
274 tmpPut.add(HConstants.CATALOG_FAMILY, null, value);
275 long startTime = System.currentTimeMillis();
276 long remaining = timeout;
277 while (remaining > 0) {
278 if (log.isLowReplicationRollEnabled() == expect) {
279 break;
280 } else {
281
282 table.put(tmpPut);
283 try {
284 Thread.sleep(200);
285 } catch (InterruptedException e) {
286
287 }
288 remaining = timeout - (System.currentTimeMillis() - startTime);
289 }
290 }
291 }
292
293
294
295
296
297 @Test
298 public void testLogRollOnDatanodeDeath() throws Exception {
299 TEST_UTIL.ensureSomeRegionServersAvailable(2);
300 assertTrue("This test requires WAL file replication set to 2.",
301 fs.getDefaultReplication(TEST_UTIL.getDataTestDirOnTestFS()) == 2);
302 LOG.info("Replication=" +
303 fs.getDefaultReplication(TEST_UTIL.getDataTestDirOnTestFS()));
304
305 this.server = cluster.getRegionServer(0);
306
307
308 HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(getName()));
309 desc.addFamily(new HColumnDescriptor(HConstants.CATALOG_FAMILY));
310
311 admin.createTable(desc);
312 Table table = TEST_UTIL.getConnection().getTable(desc.getTableName());
313 assertTrue(((HTable) table).isAutoFlush());
314
315 server = TEST_UTIL.getRSForFirstRegionInTable(desc.getTableName());
316 HRegionInfo region = server.getOnlineRegions(desc.getTableName()).get(0).getRegionInfo();
317 final FSHLog log = (FSHLog) server.getWAL(region);
318 final AtomicBoolean lowReplicationHookCalled = new AtomicBoolean(false);
319
320 log.registerWALActionsListener(new WALActionsListener.Base() {
321 @Override
322 public void logRollRequested(boolean lowReplication) {
323 if (lowReplication) {
324 lowReplicationHookCalled.lazySet(true);
325 }
326 }
327 });
328
329
330 assertTrue("Need append support for this test", FSUtils
331 .isAppendSupported(TEST_UTIL.getConfiguration()));
332
333
334
335
336
337 List<DataNode> existingNodes = dfsCluster.getDataNodes();
338 int numDataNodes = 3;
339 dfsCluster.startDataNodes(TEST_UTIL.getConfiguration(), numDataNodes, true,
340 null, null);
341 List<DataNode> allNodes = dfsCluster.getDataNodes();
342 for (int i = allNodes.size()-1; i >= 0; i--) {
343 if (existingNodes.contains(allNodes.get(i))) {
344 dfsCluster.stopDataNode( i );
345 }
346 }
347
348 assertTrue("DataNodes " + dfsCluster.getDataNodes().size() +
349 " default replication " +
350 fs.getDefaultReplication(TEST_UTIL.getDataTestDirOnTestFS()),
351 dfsCluster.getDataNodes().size() >=
352 fs.getDefaultReplication(TEST_UTIL.getDataTestDirOnTestFS()) + 1);
353
354 writeData(table, 2);
355
356 long curTime = System.currentTimeMillis();
357 LOG.info("log.getCurrentFileName(): " + log.getCurrentFileName());
358 long oldFilenum = DefaultWALProvider.extractFileNumFromWAL(log);
359 assertTrue("Log should have a timestamp older than now",
360 curTime > oldFilenum && oldFilenum != -1);
361
362 assertTrue("The log shouldn't have rolled yet",
363 oldFilenum == DefaultWALProvider.extractFileNumFromWAL(log));
364 final DatanodeInfo[] pipeline = log.getPipeLine();
365 assertTrue(pipeline.length ==
366 fs.getDefaultReplication(TEST_UTIL.getDataTestDirOnTestFS()));
367
368
369
370 assertTrue(dfsCluster.stopDataNode(pipeline[0].getName()) != null);
371
372
373 writeData(table, 2);
374 long newFilenum = DefaultWALProvider.extractFileNumFromWAL(log);
375
376 assertTrue("Missing datanode should've triggered a log roll",
377 newFilenum > oldFilenum && newFilenum > curTime);
378
379 assertTrue("The log rolling hook should have been called with the low replication flag",
380 lowReplicationHookCalled.get());
381
382
383 writeData(table, 3);
384 assertTrue("The log should not roll again.",
385 DefaultWALProvider.extractFileNumFromWAL(log) == newFilenum);
386
387
388 assertTrue(dfsCluster.stopDataNode(pipeline[1].getName()) != null);
389
390 batchWriteAndWait(table, log, 3, false, 14000);
391 int replication = log.getLogReplication();
392 assertTrue("LowReplication Roller should've been disabled, current replication="
393 + replication, !log.isLowReplicationRollEnabled());
394
395 dfsCluster
396 .startDataNodes(TEST_UTIL.getConfiguration(), 1, true, null, null);
397
398
399
400 log.rollWriter(true);
401 batchWriteAndWait(table, log, 13, true, 10000);
402 replication = log.getLogReplication();
403 assertTrue("New log file should have the default replication instead of " +
404 replication,
405 replication == fs.getDefaultReplication(TEST_UTIL.getDataTestDirOnTestFS()));
406 assertTrue("LowReplication Roller should've been enabled", log.isLowReplicationRollEnabled());
407 }
408
409
410
411
412
413
414 @Test
415 public void testLogRollOnPipelineRestart() throws Exception {
416 LOG.info("Starting testLogRollOnPipelineRestart");
417 assertTrue("This test requires WAL file replication.",
418 fs.getDefaultReplication(TEST_UTIL.getDataTestDirOnTestFS()) > 1);
419 LOG.info("Replication=" +
420 fs.getDefaultReplication(TEST_UTIL.getDataTestDirOnTestFS()));
421
422 Table t = new HTable(TEST_UTIL.getConfiguration(), TableName.META_TABLE_NAME);
423 try {
424 this.server = cluster.getRegionServer(0);
425
426
427 HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(getName()));
428 desc.addFamily(new HColumnDescriptor(HConstants.CATALOG_FAMILY));
429
430 admin.createTable(desc);
431 Table table = new HTable(TEST_UTIL.getConfiguration(), desc.getTableName());
432
433 server = TEST_UTIL.getRSForFirstRegionInTable(desc.getTableName());
434 HRegionInfo region = server.getOnlineRegions(desc.getTableName()).get(0).getRegionInfo();
435 final WAL log = server.getWAL(region);
436 final List<Path> paths = new ArrayList<Path>();
437 final List<Integer> preLogRolledCalled = new ArrayList<Integer>();
438
439 paths.add(DefaultWALProvider.getCurrentFileName(log));
440 log.registerWALActionsListener(new WALActionsListener.Base() {
441
442 @Override
443 public void preLogRoll(Path oldFile, Path newFile) {
444 LOG.debug("preLogRoll: oldFile="+oldFile+" newFile="+newFile);
445 preLogRolledCalled.add(new Integer(1));
446 }
447 @Override
448 public void postLogRoll(Path oldFile, Path newFile) {
449 paths.add(newFile);
450 }
451 });
452
453
454 assertTrue("Need append support for this test", FSUtils
455 .isAppendSupported(TEST_UTIL.getConfiguration()));
456
457 writeData(table, 1002);
458
459 long curTime = System.currentTimeMillis();
460 LOG.info("log.getCurrentFileName()): " + DefaultWALProvider.getCurrentFileName(log));
461 long oldFilenum = DefaultWALProvider.extractFileNumFromWAL(log);
462 assertTrue("Log should have a timestamp older than now",
463 curTime > oldFilenum && oldFilenum != -1);
464
465 assertTrue("The log shouldn't have rolled yet", oldFilenum ==
466 DefaultWALProvider.extractFileNumFromWAL(log));
467
468
469 dfsCluster.restartDataNodes();
470 Thread.sleep(1000);
471 dfsCluster.waitActive();
472 LOG.info("Data Nodes restarted");
473 validateData(table, 1002);
474
475
476 writeData(table, 1003);
477 long newFilenum = DefaultWALProvider.extractFileNumFromWAL(log);
478
479 assertTrue("Missing datanode should've triggered a log roll",
480 newFilenum > oldFilenum && newFilenum > curTime);
481 validateData(table, 1003);
482
483 writeData(table, 1004);
484
485
486 dfsCluster.restartDataNodes();
487 Thread.sleep(1000);
488 dfsCluster.waitActive();
489 LOG.info("Data Nodes restarted");
490 validateData(table, 1004);
491
492
493 writeData(table, 1005);
494
495
496 log.rollWriter(true);
497 assertTrue("preLogRolledCalled has size of " + preLogRolledCalled.size(),
498 preLogRolledCalled.size() >= 1);
499
500
501 Set<String> loggedRows = new HashSet<String>();
502 FSUtils fsUtils = FSUtils.getInstance(fs, TEST_UTIL.getConfiguration());
503 for (Path p : paths) {
504 LOG.debug("recovering lease for " + p);
505 fsUtils.recoverFileLease(((HFileSystem)fs).getBackingFs(), p,
506 TEST_UTIL.getConfiguration(), null);
507
508 LOG.debug("Reading WAL "+FSUtils.getPath(p));
509 WAL.Reader reader = null;
510 try {
511 reader = WALFactory.createReader(fs, p, TEST_UTIL.getConfiguration());
512 WAL.Entry entry;
513 while ((entry = reader.next()) != null) {
514 LOG.debug("#"+entry.getKey().getLogSeqNum()+": "+entry.getEdit().getCells());
515 for (Cell cell : entry.getEdit().getCells()) {
516 loggedRows.add(Bytes.toStringBinary(cell.getRow()));
517 }
518 }
519 } catch (EOFException e) {
520 LOG.debug("EOF reading file "+FSUtils.getPath(p));
521 } finally {
522 if (reader != null) reader.close();
523 }
524 }
525
526
527 assertTrue(loggedRows.contains("row1002"));
528 assertTrue(loggedRows.contains("row1003"));
529 assertTrue(loggedRows.contains("row1004"));
530 assertTrue(loggedRows.contains("row1005"));
531
532
533 for (Region r: server.getOnlineRegionsLocalContext()) {
534 try {
535 r.flush(true);
536 } catch (Exception e) {
537
538
539
540
541
542 LOG.info(e);
543 }
544 }
545
546 ResultScanner scanner = table.getScanner(new Scan());
547 try {
548 for (int i=2; i<=5; i++) {
549 Result r = scanner.next();
550 assertNotNull(r);
551 assertFalse(r.isEmpty());
552 assertEquals("row100"+i, Bytes.toString(r.getRow()));
553 }
554 } finally {
555 scanner.close();
556 }
557
558
559 for (JVMClusterUtil.RegionServerThread rsThread:
560 TEST_UTIL.getHBaseCluster().getRegionServerThreads()) {
561 assertFalse(rsThread.getRegionServer().isAborted());
562 }
563 } finally {
564 if (t != null) t.close();
565 }
566 }
567
568
569
570
571
572 @Test
573 public void testCompactionRecordDoesntBlockRolling() throws Exception {
574 Table table = null;
575
576
577 Table t = new HTable(TEST_UTIL.getConfiguration(), TableName.META_TABLE_NAME);
578 try {
579 table = createTestTable(getName());
580
581 server = TEST_UTIL.getRSForFirstRegionInTable(table.getName());
582 Region region = server.getOnlineRegions(table.getName()).get(0);
583 final WAL log = server.getWAL(region.getRegionInfo());
584 Store s = region.getStore(HConstants.CATALOG_FAMILY);
585
586
587 admin.flush(TableName.NAMESPACE_TABLE_NAME);
588
589
590 for (int i = 1; i <= 2; ++i) {
591 doPut(table, i);
592 admin.flush(table.getName());
593 }
594 doPut(table, 3);
595 assertEquals("Should have no WAL after initial writes", 0,
596 DefaultWALProvider.getNumRolledLogFiles(log));
597 assertEquals(2, s.getStorefilesCount());
598
599
600 log.rollWriter();
601 assertEquals("Should have WAL; one table is not flushed", 1,
602 DefaultWALProvider.getNumRolledLogFiles(log));
603 admin.flush(table.getName());
604 region.compact(false);
605
606 Assert.assertNotNull(s);
607 for (int waitTime = 3000; s.getStorefilesCount() > 1 && waitTime > 0; waitTime -= 200) {
608 Threads.sleepWithoutInterrupt(200);
609 }
610 assertEquals("Compaction didn't happen", 1, s.getStorefilesCount());
611
612
613 doPut(table, 0);
614 log.rollWriter();
615 assertEquals("Should have WAL; one table is not flushed", 1,
616 DefaultWALProvider.getNumRolledLogFiles(log));
617
618
619 admin.flush(table.getName());
620 doPut(table, 1);
621 log.rollWriter();
622 assertEquals("Should have 1 WALs at the end", 1,
623 DefaultWALProvider.getNumRolledLogFiles(log));
624 } finally {
625 if (t != null) t.close();
626 if (table != null) table.close();
627 }
628 }
629
630 private void doPut(Table table, int i) throws IOException {
631 Put put = new Put(Bytes.toBytes("row" + String.format("%1$04d", i)));
632 put.add(HConstants.CATALOG_FAMILY, null, value);
633 table.put(put);
634 }
635
636 private Table createTestTable(String tableName) throws IOException {
637
638 HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
639 desc.addFamily(new HColumnDescriptor(HConstants.CATALOG_FAMILY));
640 admin.createTable(desc);
641 return new HTable(TEST_UTIL.getConfiguration(), desc.getTableName());
642 }
643 }
644