View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
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   * WAL tests that can be reused across providers.
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     // testAppendClose closes the FileSystem, which will prevent us from closing cleanly here.
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     // Make block sizes small.
121     TEST_UTIL.getConfiguration().setInt("dfs.blocksize", 1024 * 1024);
122     // needed for testAppendClose()
123     TEST_UTIL.getConfiguration().setBoolean("dfs.support.broken.append", true);
124     TEST_UTIL.getConfiguration().setBoolean("dfs.support.append", true);
125     // quicker heartbeat interval for faster DN death notification
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     // faster failover with cluster.shutdown();fs.close() idiom
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    * Just write multiple logs then split.  Before fix for HADOOP-2283, this
159    * would fail.
160    * @throws IOException
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     // Add edits for three regions.
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    * Test new HDFS-265 sync.
212    * @throws Exception
213    */
214   @Test
215   public void Broken_testSync() throws Exception {
216     TableName tableName = TableName.valueOf(currentTest.getMethodName());
217     MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl(1);
218     // First verify that using streams all works.
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       // Now call sync and try reading.  Opening a Reader before you sync just
259       // gives you EOFE.
260       wal.sync();
261       // Open a Reader.
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       // Add test that checks to see that an open of a Reader works on a file
270       // that has had a sync done on it.
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       // If I sync, should see double the edits.
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       // Now do a test that ensures stuff works when we go over block boundary,
291       // especially that we return good length on file.
292       final byte [] value = new byte[1025 * 1024];  // Make a 1M value.
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       // Now I should have written out lots of blocks.  Sync then read.
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       // shutdown and ensure that Reader gets right length also.
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           // Assert that all edits are for same region.
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    * We pass different values to recoverFileLease() so that different code paths are covered
351    *
352    * For this test to pass, requires:
353    * 1. HDFS-200 (append support)
354    * 2. HDFS-988 (SafeMode should freeze file operations
355    *              [FSNamesystem.nextGenerationStampForBlock])
356    * 3. HDFS-142 (on restart, maintain pendingCreates)
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     // Now call sync to send the data to HDFS datanodes
379     wal.sync();
380      int namenodePort = cluster.getNameNodePort();
381     final Path walPath = DefaultWALProvider.getCurrentFileName(wal);
382 
383 
384     // Stop the cluster.  (ensure restart since we're sharing MiniDFSCluster)
385     try {
386       DistributedFileSystem dfs = (DistributedFileSystem) cluster.getFileSystem();
387       dfs.setSafeMode(HdfsConstants.SafeModeAction.SAFEMODE_ENTER);
388       TEST_UTIL.shutdownMiniDFSCluster();
389       try {
390         // wal.writer.close() will throw an exception,
391         // but still call this since it closes the LogSyncer thread first
392         wal.shutdown();
393       } catch (IOException e) {
394         LOG.info(e);
395       }
396       fs.close(); // closing FS last so DFSOutputStream can't call close
397       LOG.info("STOPPED first instance of the cluster");
398     } finally {
399       // Restart the cluster
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     // set the lease period to be 1 second so that the
421     // namenode triggers lease recovery upon append request
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     // Now try recovering the log, like the HMaster would do
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     // Timeout after 60 sec. Without correct patches, would be an infinite loop
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     // Make sure you can read all the content
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     // Reset the lease period
473     setLeasePeriod.invoke(cluster, new Object[]{new Long(60000), new Long(3600000)});
474   }
475 
476   /**
477    * Tests that we can write out an edit, close, and then read it back in again.
478    * @throws IOException
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       // Write columns named 1, 2, 3, etc. and then values of single byte
492       // 1, 2, 3...
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       // Now open a reader on the log and assert append worked.
514       reader = wals.createReader(fs, filename);
515       // Above we added all columns on a single row so we only read one
516       // entry in the below... thats why we have '1'.
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    * @throws IOException
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       // Write columns named 1, 2, 3, etc. and then values of single byte
550       // 1, 2, 3...
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       // Now open a reader on the log and assert append worked.
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    * Test that we can visit entries before they are appended
593    * @throws Exception
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    * A loaded WAL coprocessor won't break existing WAL test cases.
634    */
635   @Test
636   public void testWALCoprocessorLoaded() throws Exception {
637     // test to see whether the coprocessor is loaded or not.
638     WALCoprocessorHost host = wals.getWAL(UNSPECIFIED_REGION, null).getCoprocessorHost();
639     Coprocessor c = host.findCoprocessor(SampleRegionWALObserver.class.getName());
640     assertNotNull(c);
641   }
642 
643   /**
644    * @throws IOException
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       // Write log in pre-PB format.
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       // Now read the log using standard means.
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       //To change body of implemented methods use File | Settings | File Templates.
722       increments++;
723     }
724   }
725 
726   private static final byte[] UNSPECIFIED_REGION = new byte[]{};
727 
728 }