1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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
88 int[] NB_ROWS = { 1500, 60000 };
89 int[] NB_KVS = { 1, 100 };
90
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
148 Path path = pathWatcher.currentPath;
149
150 assertEquals(0, logManager.getPosition());
151
152 appendToLog();
153
154
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
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
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 }