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.assertArrayEquals;
22 import static org.junit.Assert.assertEquals;
23 import static org.junit.Assert.assertFalse;
24 import static org.junit.Assert.assertNotNull;
25 import static org.junit.Assert.assertNull;
26 import static org.junit.Assert.assertTrue;
27 import static org.junit.Assert.fail;
28
29 import java.io.IOException;
30 import java.lang.reflect.Method;
31 import java.net.BindException;
32 import java.util.List;
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.FSDataInputStream;
38 import org.apache.hadoop.fs.FSDataOutputStream;
39 import org.apache.hadoop.fs.FileStatus;
40 import org.apache.hadoop.fs.FileSystem;
41 import org.apache.hadoop.fs.Path;
42 import org.apache.hadoop.hbase.Cell;
43 import org.apache.hadoop.hbase.Coprocessor;
44 import org.apache.hadoop.hbase.HBaseTestingUtility;
45 import org.apache.hadoop.hbase.HColumnDescriptor;
46 import org.apache.hadoop.hbase.HConstants;
47 import org.apache.hadoop.hbase.HRegionInfo;
48 import org.apache.hadoop.hbase.HTableDescriptor;
49 import org.apache.hadoop.hbase.KeyValue;
50 import org.apache.hadoop.hbase.TableName;
51 import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
52 import org.apache.hadoop.hbase.coprocessor.SampleRegionWALObserver;
53 import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl;
54 import org.apache.hadoop.hbase.regionserver.wal.HLogKey;
55 import org.apache.hadoop.hbase.regionserver.wal.SequenceFileLogReader;
56 import org.apache.hadoop.hbase.regionserver.wal.SequenceFileLogWriter;
57 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
58 import org.apache.hadoop.hbase.regionserver.wal.WALCoprocessorHost;
59 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
60 import org.apache.hadoop.hbase.testclassification.MediumTests;
61 import org.apache.hadoop.hbase.util.Bytes;
62 import org.apache.hadoop.hbase.util.FSUtils;
63 import org.apache.hadoop.hbase.util.Threads;
64 import org.apache.hadoop.hdfs.DistributedFileSystem;
65 import org.apache.hadoop.hdfs.MiniDFSCluster;
66 import org.apache.hadoop.hdfs.protocol.HdfsConstants;
67 import org.junit.After;
68 import org.junit.AfterClass;
69 import org.junit.Before;
70 import org.junit.BeforeClass;
71 import org.junit.Rule;
72 import org.junit.Test;
73 import org.junit.experimental.categories.Category;
74 import org.junit.rules.TestName;
75
76
77
78
79 @Category(MediumTests.class)
80 public class TestWALFactory {
81 private static final Log LOG = LogFactory.getLog(TestWALFactory.class);
82
83 protected static Configuration conf;
84 private static MiniDFSCluster cluster;
85 protected final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
86 protected static Path hbaseDir;
87
88 protected FileSystem fs;
89 protected Path dir;
90 protected WALFactory wals;
91
92 @Rule
93 public final TestName currentTest = new TestName();
94
95 @Before
96 public void setUp() throws Exception {
97 fs = cluster.getFileSystem();
98 dir = new Path(hbaseDir, currentTest.getMethodName());
99 wals = new WALFactory(conf, null, currentTest.getMethodName());
100 }
101
102 @After
103 public void tearDown() throws Exception {
104
105 try {
106 wals.close();
107 } catch (IOException exception) {
108 LOG.warn("Encountered exception while closing wal factory. If you have other errors, this" +
109 " may be the cause. Message: " + exception);
110 LOG.debug("Exception details for failure to close wal factory.", exception);
111 }
112 FileStatus[] entries = fs.listStatus(new Path("/"));
113 for (FileStatus dir : entries) {
114 fs.delete(dir.getPath(), true);
115 }
116 }
117
118 @BeforeClass
119 public static void setUpBeforeClass() throws Exception {
120
121 TEST_UTIL.getConfiguration().setInt("dfs.blocksize", 1024 * 1024);
122
123 TEST_UTIL.getConfiguration().setBoolean("dfs.support.broken.append", true);
124 TEST_UTIL.getConfiguration().setBoolean("dfs.support.append", true);
125
126 TEST_UTIL.getConfiguration().setInt("dfs.namenode.heartbeat.recheck-interval", 5000);
127 TEST_UTIL.getConfiguration().setInt("dfs.heartbeat.interval", 1);
128 TEST_UTIL.getConfiguration().setInt("dfs.client.socket-timeout", 5000);
129
130
131 TEST_UTIL.getConfiguration()
132 .setInt("hbase.ipc.client.connect.max.retries", 1);
133 TEST_UTIL.getConfiguration().setInt(
134 "dfs.client.block.recovery.retries", 1);
135 TEST_UTIL.getConfiguration().setInt(
136 "hbase.ipc.client.connection.maxidletime", 500);
137 TEST_UTIL.getConfiguration().set(CoprocessorHost.WAL_COPROCESSOR_CONF_KEY,
138 SampleRegionWALObserver.class.getName());
139 TEST_UTIL.startMiniDFSCluster(3);
140
141 conf = TEST_UTIL.getConfiguration();
142 cluster = TEST_UTIL.getDFSCluster();
143
144 hbaseDir = TEST_UTIL.createRootDir();
145 }
146
147 @AfterClass
148 public static void tearDownAfterClass() throws Exception {
149 TEST_UTIL.shutdownMiniCluster();
150 }
151
152 @Test
153 public void canCloseSingleton() throws IOException {
154 WALFactory.getInstance(conf).close();
155 }
156
157
158
159
160
161
162 @Test
163 public void testSplit() throws IOException {
164 final TableName tableName = TableName.valueOf(currentTest.getMethodName());
165 final byte [] rowName = tableName.getName();
166 final MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl(1);
167 final Path logdir = new Path(hbaseDir,
168 DefaultWALProvider.getWALDirectoryName(currentTest.getMethodName()));
169 Path oldLogDir = new Path(hbaseDir, HConstants.HREGION_OLDLOGDIR_NAME);
170 final int howmany = 3;
171 HRegionInfo[] infos = new HRegionInfo[3];
172 Path tabledir = FSUtils.getTableDir(hbaseDir, tableName);
173 fs.mkdirs(tabledir);
174 for(int i = 0; i < howmany; i++) {
175 infos[i] = new HRegionInfo(tableName,
176 Bytes.toBytes("" + i), Bytes.toBytes("" + (i+1)), false);
177 fs.mkdirs(new Path(tabledir, infos[i].getEncodedName()));
178 LOG.info("allo " + new Path(tabledir, infos[i].getEncodedName()).toString());
179 }
180 HTableDescriptor htd = new HTableDescriptor(tableName);
181 htd.addFamily(new HColumnDescriptor("column"));
182
183
184 for (int ii = 0; ii < howmany; ii++) {
185 for (int i = 0; i < howmany; i++) {
186 final WAL log =
187 wals.getWAL(infos[i].getEncodedNameAsBytes(), infos[i].getTable().getNamespace());
188 for (int j = 0; j < howmany; j++) {
189 WALEdit edit = new WALEdit();
190 byte [] family = Bytes.toBytes("column");
191 byte [] qualifier = Bytes.toBytes(Integer.toString(j));
192 byte [] column = Bytes.toBytes("column:" + Integer.toString(j));
193 edit.add(new KeyValue(rowName, family, qualifier,
194 System.currentTimeMillis(), column));
195 LOG.info("Region " + i + ": " + edit);
196 WALKey walKey = new WALKey(infos[i].getEncodedNameAsBytes(), tableName,
197 System.currentTimeMillis(), mvcc);
198 log.append(htd, infos[i], walKey, edit, true);
199 walKey.getWriteEntry();
200 }
201 log.sync();
202 log.rollWriter(true);
203 }
204 }
205 wals.shutdown();
206 List<Path> splits = WALSplitter.split(hbaseDir, logdir, oldLogDir, fs, conf, wals);
207 verifySplits(splits, howmany);
208 }
209
210
211
212
213
214 @Test
215 public void Broken_testSync() throws Exception {
216 TableName tableName = TableName.valueOf(currentTest.getMethodName());
217 MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl(1);
218
219 Path p = new Path(dir, currentTest.getMethodName() + ".fsdos");
220 FSDataOutputStream out = fs.create(p);
221 out.write(tableName.getName());
222 Method syncMethod = null;
223 try {
224 syncMethod = out.getClass().getMethod("hflush", new Class<?> []{});
225 } catch (NoSuchMethodException e) {
226 try {
227 syncMethod = out.getClass().getMethod("sync", new Class<?> []{});
228 } catch (NoSuchMethodException ex) {
229 fail("This version of Hadoop supports neither Syncable.sync() " +
230 "nor Syncable.hflush().");
231 }
232 }
233 syncMethod.invoke(out, new Object[]{});
234 FSDataInputStream in = fs.open(p);
235 assertTrue(in.available() > 0);
236 byte [] buffer = new byte [1024];
237 int read = in.read(buffer);
238 assertEquals(tableName.getName().length, read);
239 out.close();
240 in.close();
241
242 final int total = 20;
243 WAL.Reader reader = null;
244
245 try {
246 HRegionInfo info = new HRegionInfo(tableName,
247 null,null, false);
248 HTableDescriptor htd = new HTableDescriptor();
249 htd.addFamily(new HColumnDescriptor(tableName.getName()));
250 final WAL wal = wals.getWAL(info.getEncodedNameAsBytes(), info.getTable().getNamespace());
251
252 for (int i = 0; i < total; i++) {
253 WALEdit kvs = new WALEdit();
254 kvs.add(new KeyValue(Bytes.toBytes(i), tableName.getName(), tableName.getName()));
255 wal.append(htd, info, new WALKey(info.getEncodedNameAsBytes(), tableName,
256 System.currentTimeMillis(), mvcc), kvs, true);
257 }
258
259
260 wal.sync();
261
262 Path walPath = DefaultWALProvider.getCurrentFileName(wal);
263 reader = wals.createReader(fs, walPath);
264 int count = 0;
265 WAL.Entry entry = new WAL.Entry();
266 while ((entry = reader.next(entry)) != null) count++;
267 assertEquals(total, count);
268 reader.close();
269
270
271 for (int i = 0; i < total; i++) {
272 WALEdit kvs = new WALEdit();
273 kvs.add(new KeyValue(Bytes.toBytes(i), tableName.getName(), tableName.getName()));
274 wal.append(htd, info, new WALKey(info.getEncodedNameAsBytes(), tableName,
275 System.currentTimeMillis(), mvcc), kvs, true);
276 }
277 wal.sync();
278 reader = wals.createReader(fs, walPath);
279 count = 0;
280 while((entry = reader.next(entry)) != null) count++;
281 assertTrue(count >= total);
282 reader.close();
283
284 wal.sync();
285 reader = wals.createReader(fs, walPath);
286 count = 0;
287 while((entry = reader.next(entry)) != null) count++;
288 assertEquals(total * 2, count);
289 reader.close();
290
291
292 final byte [] value = new byte[1025 * 1024];
293 for (int i = 0; i < total; i++) {
294 WALEdit kvs = new WALEdit();
295 kvs.add(new KeyValue(Bytes.toBytes(i), tableName.getName(), value));
296 wal.append(htd, info, new WALKey(info.getEncodedNameAsBytes(), tableName,
297 System.currentTimeMillis(), mvcc), kvs, true);
298 }
299
300 wal.sync();
301 reader = wals.createReader(fs, walPath);
302 count = 0;
303 while((entry = reader.next(entry)) != null) count++;
304 assertEquals(total * 3, count);
305 reader.close();
306
307 wal.shutdown();
308 reader = wals.createReader(fs, walPath);
309 count = 0;
310 while((entry = reader.next(entry)) != null) count++;
311 assertEquals(total * 3, count);
312 reader.close();
313 } finally {
314 if (reader != null) reader.close();
315 }
316 }
317
318 private void verifySplits(final List<Path> splits, final int howmany)
319 throws IOException {
320 assertEquals(howmany * howmany, splits.size());
321 for (int i = 0; i < splits.size(); i++) {
322 LOG.info("Verifying=" + splits.get(i));
323 WAL.Reader reader = wals.createReader(fs, splits.get(i));
324 try {
325 int count = 0;
326 String previousRegion = null;
327 long seqno = -1;
328 WAL.Entry entry = new WAL.Entry();
329 while((entry = reader.next(entry)) != null) {
330 WALKey key = entry.getKey();
331 String region = Bytes.toString(key.getEncodedRegionName());
332
333 if (previousRegion != null) {
334 assertEquals(previousRegion, region);
335 }
336 LOG.info("oldseqno=" + seqno + ", newseqno=" + key.getLogSeqNum());
337 assertTrue(seqno < key.getLogSeqNum());
338 seqno = key.getLogSeqNum();
339 previousRegion = region;
340 count++;
341 }
342 assertEquals(howmany, count);
343 } finally {
344 reader.close();
345 }
346 }
347 }
348
349
350
351
352
353
354
355
356
357
358 @Test (timeout=300000)
359 public void testAppendClose() throws Exception {
360 TableName tableName =
361 TableName.valueOf(currentTest.getMethodName());
362 HRegionInfo regioninfo = new HRegionInfo(tableName,
363 HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW, false);
364
365 final WAL wal =
366 wals.getWAL(regioninfo.getEncodedNameAsBytes(), regioninfo.getTable().getNamespace());
367 final int total = 20;
368
369 HTableDescriptor htd = new HTableDescriptor();
370 htd.addFamily(new HColumnDescriptor(tableName.getName()));
371
372 for (int i = 0; i < total; i++) {
373 WALEdit kvs = new WALEdit();
374 kvs.add(new KeyValue(Bytes.toBytes(i), tableName.getName(), tableName.getName()));
375 wal.append(htd, regioninfo, new WALKey(regioninfo.getEncodedNameAsBytes(), tableName,
376 System.currentTimeMillis()), kvs, true);
377 }
378
379 wal.sync();
380 int namenodePort = cluster.getNameNodePort();
381 final Path walPath = DefaultWALProvider.getCurrentFileName(wal);
382
383
384
385 try {
386 DistributedFileSystem dfs = (DistributedFileSystem) cluster.getFileSystem();
387 dfs.setSafeMode(HdfsConstants.SafeModeAction.SAFEMODE_ENTER);
388 TEST_UTIL.shutdownMiniDFSCluster();
389 try {
390
391
392 wal.shutdown();
393 } catch (IOException e) {
394 LOG.info(e);
395 }
396 fs.close();
397 LOG.info("STOPPED first instance of the cluster");
398 } finally {
399
400 while (cluster.isClusterUp()){
401 LOG.error("Waiting for cluster to go down");
402 Thread.sleep(1000);
403 }
404 assertFalse(cluster.isClusterUp());
405 cluster = null;
406 for (int i = 0; i < 100; i++) {
407 try {
408 cluster = TEST_UTIL.startMiniDFSClusterForTestWAL(namenodePort);
409 break;
410 } catch (BindException e) {
411 LOG.info("Sleeping. BindException bringing up new cluster");
412 Threads.sleep(1000);
413 }
414 }
415 cluster.waitActive();
416 fs = cluster.getFileSystem();
417 LOG.info("STARTED second instance.");
418 }
419
420
421
422 Method setLeasePeriod = cluster.getClass()
423 .getDeclaredMethod("setLeasePeriod", new Class[]{Long.TYPE, Long.TYPE});
424 setLeasePeriod.setAccessible(true);
425 setLeasePeriod.invoke(cluster, 1000L, 1000L);
426 try {
427 Thread.sleep(1000);
428 } catch (InterruptedException e) {
429 LOG.info(e);
430 }
431
432
433 final FileSystem recoveredFs = fs;
434 final Configuration rlConf = conf;
435
436 class RecoverLogThread extends Thread {
437 public Exception exception = null;
438 public void run() {
439 try {
440 FSUtils.getInstance(fs, rlConf)
441 .recoverFileLease(recoveredFs, walPath, rlConf, null);
442 } catch (IOException e) {
443 exception = e;
444 }
445 }
446 }
447
448 RecoverLogThread t = new RecoverLogThread();
449 t.start();
450
451 t.join(60 * 1000);
452 if(t.isAlive()) {
453 t.interrupt();
454 throw new Exception("Timed out waiting for WAL.recoverLog()");
455 }
456
457 if (t.exception != null)
458 throw t.exception;
459
460
461 WAL.Reader reader = wals.createReader(fs, walPath);
462 int count = 0;
463 WAL.Entry entry = new WAL.Entry();
464 while (reader.next(entry) != null) {
465 count++;
466 assertTrue("Should be one KeyValue per WALEdit",
467 entry.getEdit().getCells().size() == 1);
468 }
469 assertEquals(total, count);
470 reader.close();
471
472
473 setLeasePeriod.invoke(cluster, new Object[]{new Long(60000), new Long(3600000)});
474 }
475
476
477
478
479
480 @Test
481 public void testEditAdd() throws IOException {
482 final int COL_COUNT = 10;
483 final HTableDescriptor htd =
484 new HTableDescriptor(TableName.valueOf("tablename")).addFamily(new HColumnDescriptor(
485 "column"));
486 final byte [] row = Bytes.toBytes("row");
487 WAL.Reader reader = null;
488 try {
489 final MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl(1);
490
491
492
493 long timestamp = System.currentTimeMillis();
494 WALEdit cols = new WALEdit();
495 for (int i = 0; i < COL_COUNT; i++) {
496 cols.add(new KeyValue(row, Bytes.toBytes("column"),
497 Bytes.toBytes(Integer.toString(i)),
498 timestamp, new byte[] { (byte)(i + '0') }));
499 }
500 HRegionInfo info = new HRegionInfo(htd.getTableName(),
501 row,Bytes.toBytes(Bytes.toString(row) + "1"), false);
502 final WAL log = wals.getWAL(info.getEncodedNameAsBytes(), info.getTable().getNamespace());
503
504 final long txid = log.append(htd, info,
505 new WALKey(info.getEncodedNameAsBytes(), htd.getTableName(), System.currentTimeMillis(),
506 mvcc),
507 cols, true);
508 log.sync(txid);
509 log.startCacheFlush(info.getEncodedNameAsBytes(), htd.getFamiliesKeys());
510 log.completeCacheFlush(info.getEncodedNameAsBytes());
511 log.shutdown();
512 Path filename = DefaultWALProvider.getCurrentFileName(log);
513
514 reader = wals.createReader(fs, filename);
515
516
517 for (int i = 0; i < 1; i++) {
518 WAL.Entry entry = reader.next(null);
519 if (entry == null) break;
520 WALKey key = entry.getKey();
521 WALEdit val = entry.getEdit();
522 assertTrue(Bytes.equals(info.getEncodedNameAsBytes(), key.getEncodedRegionName()));
523 assertTrue(htd.getTableName().equals(key.getTablename()));
524 Cell cell = val.getCells().get(0);
525 assertTrue(Bytes.equals(row, cell.getRow()));
526 assertEquals((byte)(i + '0'), cell.getValue()[0]);
527 System.out.println(key + " " + val);
528 }
529 } finally {
530 if (reader != null) {
531 reader.close();
532 }
533 }
534 }
535
536
537
538
539 @Test
540 public void testAppend() throws IOException {
541 final int COL_COUNT = 10;
542 final HTableDescriptor htd =
543 new HTableDescriptor(TableName.valueOf("tablename")).addFamily(new HColumnDescriptor(
544 "column"));
545 final byte [] row = Bytes.toBytes("row");
546 WAL.Reader reader = null;
547 final MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl(1);
548 try {
549
550
551 long timestamp = System.currentTimeMillis();
552 WALEdit cols = new WALEdit();
553 for (int i = 0; i < COL_COUNT; i++) {
554 cols.add(new KeyValue(row, Bytes.toBytes("column"),
555 Bytes.toBytes(Integer.toString(i)),
556 timestamp, new byte[] { (byte)(i + '0') }));
557 }
558 HRegionInfo hri = new HRegionInfo(htd.getTableName(),
559 HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW);
560 final WAL log = wals.getWAL(hri.getEncodedNameAsBytes(), hri.getTable().getNamespace());
561 final long txid = log.append(htd, hri,
562 new WALKey(hri.getEncodedNameAsBytes(), htd.getTableName(), System.currentTimeMillis(),
563 mvcc),
564 cols, true);
565 log.sync(txid);
566 log.startCacheFlush(hri.getEncodedNameAsBytes(), htd.getFamiliesKeys());
567 log.completeCacheFlush(hri.getEncodedNameAsBytes());
568 log.shutdown();
569 Path filename = DefaultWALProvider.getCurrentFileName(log);
570
571 reader = wals.createReader(fs, filename);
572 WAL.Entry entry = reader.next();
573 assertEquals(COL_COUNT, entry.getEdit().size());
574 int idx = 0;
575 for (Cell val : entry.getEdit().getCells()) {
576 assertTrue(Bytes.equals(hri.getEncodedNameAsBytes(),
577 entry.getKey().getEncodedRegionName()));
578 assertTrue(htd.getTableName().equals(entry.getKey().getTablename()));
579 assertTrue(Bytes.equals(row, val.getRow()));
580 assertEquals((byte)(idx + '0'), val.getValue()[0]);
581 System.out.println(entry.getKey() + " " + val);
582 idx++;
583 }
584 } finally {
585 if (reader != null) {
586 reader.close();
587 }
588 }
589 }
590
591
592
593
594
595 @Test
596 public void testVisitors() throws Exception {
597 final int COL_COUNT = 10;
598 final TableName tableName =
599 TableName.valueOf("tablename");
600 final byte [] row = Bytes.toBytes("row");
601 final DumbWALActionsListener visitor = new DumbWALActionsListener();
602 final MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl(1);
603 long timestamp = System.currentTimeMillis();
604 HTableDescriptor htd = new HTableDescriptor();
605 htd.addFamily(new HColumnDescriptor("column"));
606
607 HRegionInfo hri = new HRegionInfo(tableName,
608 HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW);
609 final WAL log = wals.getWAL(hri.getEncodedNameAsBytes(), hri.getTable().getNamespace());
610 log.registerWALActionsListener(visitor);
611 for (int i = 0; i < COL_COUNT; i++) {
612 WALEdit cols = new WALEdit();
613 cols.add(new KeyValue(row, Bytes.toBytes("column"),
614 Bytes.toBytes(Integer.toString(i)),
615 timestamp, new byte[]{(byte) (i + '0')}));
616 log.append(htd, hri, new WALKey(hri.getEncodedNameAsBytes(), tableName,
617 System.currentTimeMillis(), mvcc), cols, true);
618 }
619 log.sync();
620 assertEquals(COL_COUNT, visitor.increments);
621 log.unregisterWALActionsListener(visitor);
622 WALEdit cols = new WALEdit();
623 cols.add(new KeyValue(row, Bytes.toBytes("column"),
624 Bytes.toBytes(Integer.toString(11)),
625 timestamp, new byte[]{(byte) (11 + '0')}));
626 log.append(htd, hri, new WALKey(hri.getEncodedNameAsBytes(), tableName,
627 System.currentTimeMillis(), mvcc), cols, true);
628 log.sync();
629 assertEquals(COL_COUNT, visitor.increments);
630 }
631
632
633
634
635 @Test
636 public void testWALCoprocessorLoaded() throws Exception {
637
638 WALCoprocessorHost host = wals.getWAL(UNSPECIFIED_REGION, null).getCoprocessorHost();
639 Coprocessor c = host.findCoprocessor(SampleRegionWALObserver.class.getName());
640 assertNotNull(c);
641 }
642
643
644
645
646 @Test
647 public void testReadLegacyLog() throws IOException {
648 final int columnCount = 5;
649 final int recordCount = 5;
650 final TableName tableName =
651 TableName.valueOf("tablename");
652 final byte[] row = Bytes.toBytes("row");
653 long timestamp = System.currentTimeMillis();
654 Path path = new Path(dir, "tempwal");
655 SequenceFileLogWriter sflw = null;
656 WAL.Reader reader = null;
657 try {
658 HRegionInfo hri = new HRegionInfo(tableName,
659 HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW);
660 HTableDescriptor htd = new HTableDescriptor(tableName);
661 fs.mkdirs(dir);
662
663 sflw = new SequenceFileLogWriter();
664 sflw.init(fs, path, conf, false);
665 for (int i = 0; i < recordCount; ++i) {
666 WALKey key = new HLogKey(
667 hri.getEncodedNameAsBytes(), tableName, i, timestamp, HConstants.DEFAULT_CLUSTER_ID);
668 WALEdit edit = new WALEdit();
669 for (int j = 0; j < columnCount; ++j) {
670 if (i == 0) {
671 htd.addFamily(new HColumnDescriptor("column" + j));
672 }
673 String value = i + "" + j;
674 edit.add(new KeyValue(row, row, row, timestamp, Bytes.toBytes(value)));
675 }
676 sflw.append(new WAL.Entry(key, edit));
677 }
678 sflw.sync();
679 sflw.close();
680
681
682 reader = wals.createReader(fs, path);
683 assertTrue(reader instanceof SequenceFileLogReader);
684 for (int i = 0; i < recordCount; ++i) {
685 WAL.Entry entry = reader.next();
686 assertNotNull(entry);
687 assertEquals(columnCount, entry.getEdit().size());
688 assertArrayEquals(hri.getEncodedNameAsBytes(), entry.getKey().getEncodedRegionName());
689 assertEquals(tableName, entry.getKey().getTablename());
690 int idx = 0;
691 for (Cell val : entry.getEdit().getCells()) {
692 assertTrue(Bytes.equals(row, val.getRow()));
693 String value = i + "" + idx;
694 assertArrayEquals(Bytes.toBytes(value), val.getValue());
695 idx++;
696 }
697 }
698 WAL.Entry entry = reader.next();
699 assertNull(entry);
700 } finally {
701 if (sflw != null) {
702 sflw.close();
703 }
704 if (reader != null) {
705 reader.close();
706 }
707 }
708 }
709
710 static class DumbWALActionsListener extends WALActionsListener.Base {
711 int increments = 0;
712
713 @Override
714 public void visitLogEntryBeforeWrite(HRegionInfo info, WALKey logKey,
715 WALEdit logEdit) {
716 increments++;
717 }
718
719 @Override
720 public void visitLogEntryBeforeWrite(HTableDescriptor htd, WALKey logKey, WALEdit logEdit) {
721
722 increments++;
723 }
724 }
725
726 private static final byte[] UNSPECIFIED_REGION = new byte[]{};
727
728 }