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