View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  
19  package org.apache.hadoop.hbase.replication.regionserver;
20  
21  import static org.junit.Assert.assertEquals;
22  import static org.junit.Assert.assertNotEquals;
23  import static org.junit.Assert.assertNotNull;
24  import static org.junit.Assert.assertNull;
25  import static org.junit.Assert.fail;
26  
27  import java.io.IOException;
28  import java.util.ArrayList;
29  import java.util.Collection;
30  import java.util.List;
31  import java.util.concurrent.atomic.AtomicLong;
32  
33  import org.apache.hadoop.conf.Configuration;
34  import org.apache.hadoop.fs.FileSystem;
35  import org.apache.hadoop.fs.Path;
36  import org.apache.hadoop.hbase.HBaseTestingUtility;
37  import org.apache.hadoop.hbase.HConstants;
38  import org.apache.hadoop.hbase.HRegionInfo;
39  import org.apache.hadoop.hbase.HTableDescriptor;
40  import org.apache.hadoop.hbase.KeyValue;
41  import org.apache.hadoop.hbase.TableName;
42  import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl;
43  import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
44  import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
45  import org.apache.hadoop.hbase.testclassification.LargeTests;
46  import org.apache.hadoop.hbase.util.Bytes;
47  import org.apache.hadoop.hbase.wal.WAL;
48  import org.apache.hadoop.hbase.wal.WALFactory;
49  import org.apache.hadoop.hbase.wal.WALKey;
50  import org.apache.hadoop.hdfs.MiniDFSCluster;
51  import org.junit.After;
52  import org.junit.AfterClass;
53  import org.junit.Before;
54  import org.junit.BeforeClass;
55  import org.junit.Rule;
56  import org.junit.Test;
57  import org.junit.experimental.categories.Category;
58  import org.junit.rules.TestName;
59  import org.junit.runner.RunWith;
60  import org.junit.runners.Parameterized;
61  import org.junit.runners.Parameterized.Parameters;
62  @Category({LargeTests.class})
63  @RunWith(Parameterized.class)
64  public class TestReplicationWALReaderManager {
65  
66    private static HBaseTestingUtility TEST_UTIL;
67    private static Configuration conf;
68    private static FileSystem fs;
69    private static MiniDFSCluster cluster;
70    private static final TableName tableName = TableName.valueOf("tablename");
71    private static final byte [] family = Bytes.toBytes("column");
72    private static final byte [] qualifier = Bytes.toBytes("qualifier");
73    private static final HRegionInfo info = new HRegionInfo(tableName,
74        HConstants.EMPTY_START_ROW, HConstants.LAST_ROW, false);
75    private static final HTableDescriptor htd = new HTableDescriptor(tableName);
76  
77    private WAL log;
78    private ReplicationWALReaderManager logManager;
79    private PathWatcher pathWatcher;
80    private int nbRows;
81    private int walEditKVs;
82    @Rule public TestName tn = new TestName();
83    private final MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl();
84  
85    @Parameters
86    public static Collection<Object[]> parameters() {
87      // Try out different combinations of row count and KeyValue count
88      int[] NB_ROWS = { 1500, 60000 };
89      int[] NB_KVS = { 1, 100 };
90      // whether compression is used
91      Boolean[] BOOL_VALS = { false, true };
92      List<Object[]> parameters = new ArrayList<Object[]>();
93      for (int nbRows : NB_ROWS) {
94        for (int walEditKVs : NB_KVS) {
95          for (boolean b : BOOL_VALS) {
96            Object[] arr = new Object[3];
97            arr[0] = nbRows;
98            arr[1] = walEditKVs;
99            arr[2] = b;
100           parameters.add(arr);
101         }
102       }
103     }
104     return parameters;
105   }
106 
107   public TestReplicationWALReaderManager(int nbRows, int walEditKVs, boolean enableCompression) {
108     this.nbRows = nbRows;
109     this.walEditKVs = walEditKVs;
110     TEST_UTIL.getConfiguration().setBoolean(HConstants.ENABLE_WAL_COMPRESSION,
111       enableCompression);
112     mvcc.advanceTo(1);
113   }
114   
115   @BeforeClass
116   public static void setUpBeforeClass() throws Exception {
117     TEST_UTIL = new HBaseTestingUtility();
118     conf = TEST_UTIL.getConfiguration();
119     TEST_UTIL.startMiniDFSCluster(3);
120 
121     cluster = TEST_UTIL.getDFSCluster();
122     fs = cluster.getFileSystem();
123   }
124 
125   @AfterClass
126   public static void tearDownAfterClass() throws Exception {
127     TEST_UTIL.shutdownMiniCluster();
128   }
129 
130   @Before
131   public void setUp() throws Exception {
132     logManager = new ReplicationWALReaderManager(fs, conf);
133     List<WALActionsListener> listeners = new ArrayList<WALActionsListener>();
134     pathWatcher = new PathWatcher();
135     listeners.add(pathWatcher);
136     final WALFactory wals = new WALFactory(conf, listeners, tn.getMethodName());
137     log = wals.getWAL(info.getEncodedNameAsBytes(), info.getTable().getNamespace());
138   }
139 
140   @After
141   public void tearDown() throws Exception {
142     log.close();
143   }
144 
145   @Test
146   public void test() throws Exception {
147     // Grab the path that was generated when the log rolled as part of its creation
148     Path path = pathWatcher.currentPath;
149 
150     assertEquals(0, logManager.getPosition());
151 
152     appendToLog();
153 
154     // There's one edit in the log, read it. Reading past it needs to return nulls
155     assertNotNull(logManager.openReader(path));
156     logManager.seek();
157     WAL.Entry entry = logManager.readNextAndSetPosition();
158     assertNotNull(entry);
159     entry = logManager.readNextAndSetPosition();
160     assertNull(entry);
161     logManager.closeReader();
162     long oldPos = logManager.getPosition();
163 
164     appendToLog();
165 
166     // Read the newly added entry, make sure we made progress
167     assertNotNull(logManager.openReader(path));
168     logManager.seek();
169     entry = logManager.readNextAndSetPosition();
170     assertNotEquals(oldPos, logManager.getPosition());
171     assertNotNull(entry);
172     logManager.closeReader();
173     oldPos = logManager.getPosition();
174 
175     log.rollWriter();
176 
177     // We rolled but we still should see the end of the first log and not get data
178     assertNotNull(logManager.openReader(path));
179     logManager.seek();
180     entry = logManager.readNextAndSetPosition();
181     assertEquals(oldPos, logManager.getPosition());
182     assertNull(entry);
183     logManager.finishCurrentFile();
184 
185     path = pathWatcher.currentPath;
186 
187     for (int i = 0; i < nbRows; i++) { appendToLogPlus(walEditKVs); }
188     log.rollWriter();
189     logManager.openReader(path);
190     logManager.seek();
191     for (int i = 0; i < nbRows; i++) {
192       WAL.Entry e = logManager.readNextAndSetPosition();
193       if (e == null) {
194         fail("Should have enough entries");
195       }
196     }
197   }
198 
199   private void appendToLog() throws IOException {
200     appendToLogPlus(1);
201   }
202 
203   private void appendToLogPlus(int count) throws IOException {
204     final long txid = log.append(htd, info,
205         new WALKey(info.getEncodedNameAsBytes(), tableName, System.currentTimeMillis(), mvcc),
206         getWALEdits(count), true);
207     log.sync(txid);
208   }
209 
210   private WALEdit getWALEdits(int count) {
211     WALEdit edit = new WALEdit();
212     for (int i = 0; i < count; i++) {
213       edit.add(new KeyValue(Bytes.toBytes(System.currentTimeMillis()), family, qualifier,
214         System.currentTimeMillis(), qualifier));
215     }
216     return edit;
217   }
218 
219   class PathWatcher extends WALActionsListener.Base {
220 
221     Path currentPath;
222 
223     @Override
224     public void preLogRoll(Path oldPath, Path newPath) throws IOException {
225       currentPath = newPath;
226     }
227   }
228 }