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    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     // testAppendClose closes the FileSystem, which will prevent us from closing cleanly here.
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     // Make block sizes small.
123     TEST_UTIL.getConfiguration().setInt("dfs.blocksize", 1024 * 1024);
124     // needed for testAppendClose()
125     TEST_UTIL.getConfiguration().setBoolean("dfs.support.broken.append", true);
126     TEST_UTIL.getConfiguration().setBoolean("dfs.support.append", true);
127     // quicker heartbeat interval for faster DN death notification
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     // faster failover with cluster.shutdown();fs.close() idiom
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    * Just write multiple logs then split.  Before fix for HADOOP-2283, this
162    * would fail.
163    * @throws IOException
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     // Add edits for three regions.
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    * Test new HDFS-265 sync.
215    * @throws Exception
216    */
217   @Test
218   public void Broken_testSync() throws Exception {
219     TableName tableName = TableName.valueOf(currentTest.getMethodName());
220     MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl(1);
221     // First verify that using streams all works.
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       // Now call sync and try reading.  Opening a Reader before you sync just
262       // gives you EOFE.
263       wal.sync();
264       // Open a Reader.
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       // Add test that checks to see that an open of a Reader works on a file
273       // that has had a sync done on it.
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       // If I sync, should see double the edits.
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       // Now do a test that ensures stuff works when we go over block boundary,
294       // especially that we return good length on file.
295       final byte [] value = new byte[1025 * 1024];  // Make a 1M value.
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       // Now I should have written out lots of blocks.  Sync then read.
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       // shutdown and ensure that Reader gets right length also.
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           // Assert that all edits are for same region.
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    * We pass different values to recoverFileLease() so that different code paths are covered
354    *
355    * For this test to pass, requires:
356    * 1. HDFS-200 (append support)
357    * 2. HDFS-988 (SafeMode should freeze file operations
358    *              [FSNamesystem.nextGenerationStampForBlock])
359    * 3. HDFS-142 (on restart, maintain pendingCreates)
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     MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl();
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(), mvcc), kvs,  true);
380     }
381     // Now call sync to send the data to HDFS datanodes
382     wal.sync();
383      int namenodePort = cluster.getNameNodePort();
384     final Path walPath = DefaultWALProvider.getCurrentFileName(wal);
385 
386 
387     // Stop the cluster.  (ensure restart since we're sharing MiniDFSCluster)
388     try {
389       DistributedFileSystem dfs = (DistributedFileSystem) cluster.getFileSystem();
390       dfs.setSafeMode(HdfsConstants.SafeModeAction.SAFEMODE_ENTER);
391       TEST_UTIL.shutdownMiniDFSCluster();
392       try {
393         // wal.writer.close() will throw an exception,
394         // but still call this since it closes the LogSyncer thread first
395         wal.shutdown();
396       } catch (IOException e) {
397         LOG.info(e);
398       }
399       fs.close(); // closing FS last so DFSOutputStream can't call close
400       LOG.info("STOPPED first instance of the cluster");
401     } finally {
402       // Restart the cluster
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     // set the lease period to be 1 second so that the
424     // namenode triggers lease recovery upon append request
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     // Now try recovering the log, like the HMaster would do
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     // Timeout after 60 sec. Without correct patches, would be an infinite loop
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     // Make sure you can read all the content
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     // Reset the lease period
476     setLeasePeriod.invoke(cluster, new Object[]{new Long(60000), new Long(3600000)});
477   }
478 
479   /**
480    * Tests that we can write out an edit, close, and then read it back in again.
481    * @throws IOException
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       // Write columns named 1, 2, 3, etc. and then values of single byte
495       // 1, 2, 3...
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       // Now open a reader on the log and assert append worked.
517       reader = wals.createReader(fs, filename);
518       // Above we added all columns on a single row so we only read one
519       // entry in the below... thats why we have '1'.
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    * @throws IOException
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       // Write columns named 1, 2, 3, etc. and then values of single byte
553       // 1, 2, 3...
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       // Now open a reader on the log and assert append worked.
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    * Test that we can visit entries before they are appended
596    * @throws Exception
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    * A loaded WAL coprocessor won't break existing WAL test cases.
637    */
638   @Test
639   public void testWALCoprocessorLoaded() throws Exception {
640     // test to see whether the coprocessor is loaded or not.
641     WALCoprocessorHost host = wals.getWAL(UNSPECIFIED_REGION, null).getCoprocessorHost();
642     Coprocessor c = host.findCoprocessor(SampleRegionWALObserver.class.getName());
643     assertNotNull(c);
644   }
645 
646   /**
647    * @throws IOException
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       // Write log in pre-PB format.
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       // Now read the log using standard means.
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       //To change body of implemented methods use File | Settings | File Templates.
725       increments++;
726     }
727   }
728 
729   private static final byte[] UNSPECIFIED_REGION = new byte[]{};
730 
731 }