1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.regionserver;
20
21 import static org.junit.Assert.*;
22
23 import java.io.IOException;
24 import java.util.Arrays;
25 import java.util.Collection;
26 import java.util.concurrent.atomic.AtomicBoolean;
27 import java.util.concurrent.atomic.AtomicInteger;
28 import java.util.concurrent.atomic.AtomicReference;
29
30 import org.apache.commons.logging.Log;
31 import org.apache.commons.logging.LogFactory;
32 import org.apache.commons.logging.impl.Log4JLogger;
33 import org.apache.hadoop.conf.Configuration;
34 import org.apache.hadoop.hbase.HBaseTestingUtility;
35 import org.apache.hadoop.hbase.HConstants;
36 import org.apache.hadoop.hbase.HTableDescriptor;
37 import org.apache.hadoop.hbase.TableName;
38 import org.apache.hadoop.hbase.Waiter.Predicate;
39 import org.apache.hadoop.hbase.client.Admin;
40 import org.apache.hadoop.hbase.client.Connection;
41 import org.apache.hadoop.hbase.client.ConnectionFactory;
42 import org.apache.hadoop.hbase.client.Consistency;
43 import org.apache.hadoop.hbase.client.Get;
44 import org.apache.hadoop.hbase.client.RpcRetryingCaller;
45 import org.apache.hadoop.hbase.client.Table;
46 import org.apache.hadoop.hbase.replication.regionserver.TestRegionReplicaReplicationEndpoint;
47 import org.apache.hadoop.hbase.testclassification.LargeTests;
48 import org.apache.hadoop.hbase.util.Bytes;
49 import org.apache.hadoop.hbase.util.ServerRegionReplicaUtil;
50 import org.apache.hadoop.hbase.util.Threads;
51 import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread;
52 import org.apache.log4j.Level;
53 import org.junit.After;
54 import org.junit.Before;
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
63
64
65
66 @RunWith(Parameterized.class)
67 @Category(LargeTests.class)
68 public class TestRegionReplicaFailover {
69
70 private static final Log LOG = LogFactory.getLog(TestRegionReplicaReplicationEndpoint.class);
71
72 static {
73 ((Log4JLogger)RpcRetryingCaller.LOG).getLogger().setLevel(Level.ALL);
74 }
75
76 private static final HBaseTestingUtility HTU = new HBaseTestingUtility();
77
78 private static final int NB_SERVERS = 3;
79
80 protected final byte[][] families = new byte[][] {HBaseTestingUtility.fam1,
81 HBaseTestingUtility.fam2, HBaseTestingUtility.fam3};
82 protected final byte[] fam = HBaseTestingUtility.fam1;
83 protected final byte[] qual1 = Bytes.toBytes("qual1");
84 protected final byte[] value1 = Bytes.toBytes("value1");
85 protected final byte[] row = Bytes.toBytes("rowA");
86 protected final byte[] row2 = Bytes.toBytes("rowB");
87
88 @Rule public TestName name = new TestName();
89
90 private HTableDescriptor htd;
91
92
93
94
95 @Parameters
96 public static Collection<Object[]> getParameters() {
97 Object[][] params =
98 new Boolean[][] { {true}, {false} };
99 return Arrays.asList(params);
100 }
101
102 @Parameterized.Parameter(0)
103 public boolean distributedLogReplay;
104
105 @Before
106 public void before() throws Exception {
107 Configuration conf = HTU.getConfiguration();
108 conf.setBoolean(HConstants.REPLICATION_ENABLE_KEY, true);
109 conf.setBoolean(ServerRegionReplicaUtil.REGION_REPLICA_REPLICATION_CONF_KEY, true);
110 conf.setBoolean(ServerRegionReplicaUtil.REGION_REPLICA_WAIT_FOR_PRIMARY_FLUSH_CONF_KEY, true);
111 conf.setInt("replication.stats.thread.period.seconds", 5);
112 conf.setBoolean("hbase.tests.use.shortcircuit.reads", false);
113 conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, distributedLogReplay);
114
115 HTU.startMiniCluster(NB_SERVERS);
116 htd = HTU.createTableDescriptor(
117 name.getMethodName().substring(0, name.getMethodName().length()-3));
118 htd.setRegionReplication(3);
119 HTU.getHBaseAdmin().createTable(htd);
120 }
121
122 @After
123 public void after() throws Exception {
124 HTU.deleteTableIfAny(htd.getTableName());
125 HTU.shutdownMiniCluster();
126 }
127
128
129
130
131
132 @Test(timeout = 60000)
133 public void testSecondaryRegionWithEmptyRegion() throws IOException {
134
135
136 try (Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
137 Table table = connection.getTable(htd.getTableName())) {
138
139 Get get = new Get(row);
140 get.setConsistency(Consistency.TIMELINE);
141 get.setReplicaId(1);
142 table.get(get);
143 }
144 }
145
146
147
148
149
150
151 @Test(timeout = 60000)
152 public void testSecondaryRegionWithNonEmptyRegion() throws IOException {
153
154
155 try (Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
156 Table table = connection.getTable(htd.getTableName())) {
157
158 HTU.loadNumericRows(table, fam, 0, 1000);
159
160 HTU.getHBaseAdmin().disableTable(htd.getTableName());
161 HTU.getHBaseAdmin().enableTable(htd.getTableName());
162
163 HTU.verifyNumericRows(table, fam, 0, 1000, 1);
164 }
165 }
166
167
168
169
170 @Test (timeout = 120000)
171 public void testPrimaryRegionKill() throws Exception {
172 try (Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
173 Table table = connection.getTable(htd.getTableName())) {
174
175 HTU.loadNumericRows(table, fam, 0, 1000);
176
177
178 verifyNumericRowsWithTimeout(table, fam, 0, 1000, 1, 30000);
179 verifyNumericRowsWithTimeout(table, fam, 0, 1000, 2, 30000);
180
181
182
183
184 boolean aborted = false;
185 for (RegionServerThread rs : HTU.getMiniHBaseCluster().getRegionServerThreads()) {
186 for (Region r : rs.getRegionServer().getOnlineRegions(htd.getTableName())) {
187 if (r.getRegionInfo().getReplicaId() == 0) {
188 LOG.info("Aborting region server hosting primary region replica");
189 rs.getRegionServer().abort("for test");
190 aborted = true;
191 }
192 }
193 }
194 assertTrue(aborted);
195
196
197 verifyNumericRowsWithTimeout(table, fam, 0, 1000, 0, 30000);
198 verifyNumericRowsWithTimeout(table, fam, 0, 1000, 1, 30000);
199 verifyNumericRowsWithTimeout(table, fam, 0, 1000, 2, 30000);
200 }
201
202
203 HTU.getMiniHBaseCluster().startRegionServer();
204 }
205
206
207
208 private void verifyNumericRowsWithTimeout(final Table table, final byte[] f, final int startRow,
209 final int endRow, final int replicaId, final long timeout) throws Exception {
210 try {
211 HTU.waitFor(timeout, new Predicate<Exception>() {
212 @Override
213 public boolean evaluate() throws Exception {
214 try {
215 HTU.verifyNumericRows(table, f, startRow, endRow, replicaId);
216 return true;
217 } catch (AssertionError ae) {
218 return false;
219 }
220 }
221 });
222 } catch (Throwable t) {
223
224 HTU.verifyNumericRows(table, f, startRow, endRow, replicaId);
225 }
226 }
227
228
229
230
231
232 @Test (timeout = 120000)
233 public void testSecondaryRegionKill() throws Exception {
234 try (Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
235 Table table = connection.getTable(htd.getTableName())) {
236 HTU.loadNumericRows(table, fam, 0, 1000);
237
238
239 verifyNumericRowsWithTimeout(table, fam, 0, 1000, 1, 30000);
240 verifyNumericRowsWithTimeout(table, fam, 0, 1000, 2, 30000);
241
242
243
244
245 boolean aborted = false;
246 for (RegionServerThread rs : HTU.getMiniHBaseCluster().getRegionServerThreads()) {
247 for (Region r : rs.getRegionServer().getOnlineRegions(htd.getTableName())) {
248 if (r.getRegionInfo().getReplicaId() == 1) {
249 LOG.info("Aborting region server hosting secondary region replica");
250 rs.getRegionServer().abort("for test");
251 aborted = true;
252 }
253 }
254 }
255 assertTrue(aborted);
256
257 Threads.sleep(5000);
258
259 HTU.verifyNumericRows(table, fam, 0, 1000, 1);
260 HTU.verifyNumericRows(table, fam, 0, 1000, 2);
261 }
262
263
264 HTU.getMiniHBaseCluster().startRegionServer();
265 }
266
267
268
269
270
271
272 @Test (timeout = 120000)
273 public void testSecondaryRegionKillWhilePrimaryIsAcceptingWrites() throws Exception {
274 try (Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
275 Table table = connection.getTable(htd.getTableName());
276 Admin admin = connection.getAdmin()) {
277
278 HTU.loadNumericRows(table, fam, 0, 1000);
279 admin.flush(table.getName());
280 HTU.loadNumericRows(table, fam, 1000, 2000);
281
282 final AtomicReference<Throwable> ex = new AtomicReference<Throwable>(null);
283 final AtomicBoolean done = new AtomicBoolean(false);
284 final AtomicInteger key = new AtomicInteger(2000);
285
286 Thread loader = new Thread() {
287 @Override
288 public void run() {
289 while (!done.get()) {
290 try {
291 HTU.loadNumericRows(table, fam, key.get(), key.get()+1000);
292 key.addAndGet(1000);
293 } catch (Throwable e) {
294 ex.compareAndSet(null, e);
295 }
296 }
297 }
298 };
299 loader.start();
300
301 Thread aborter = new Thread() {
302 @Override
303 public void run() {
304 try {
305 boolean aborted = false;
306 for (RegionServerThread rs : HTU.getMiniHBaseCluster().getRegionServerThreads()) {
307 for (Region r : rs.getRegionServer().getOnlineRegions(htd.getTableName())) {
308 if (r.getRegionInfo().getReplicaId() == 1) {
309 LOG.info("Aborting region server hosting secondary region replica");
310 rs.getRegionServer().abort("for test");
311 aborted = true;
312 }
313 }
314 }
315 assertTrue(aborted);
316 } catch (Throwable e) {
317 ex.compareAndSet(null, e);
318 }
319 };
320 };
321
322 aborter.start();
323 aborter.join();
324 done.set(true);
325 loader.join();
326
327 assertNull(ex.get());
328
329 assertTrue(key.get() > 1000);
330 LOG.info("Loaded up to key :" + key.get());
331 verifyNumericRowsWithTimeout(table, fam, 0, key.get(), 0, 30000);
332 verifyNumericRowsWithTimeout(table, fam, 0, key.get(), 1, 30000);
333 verifyNumericRowsWithTimeout(table, fam, 0, key.get(), 2, 30000);
334 }
335
336
337 HTU.getMiniHBaseCluster().startRegionServer();
338 }
339
340
341
342
343
344 @Test (timeout = 120000)
345 public void testLotsOfRegionReplicas() throws IOException {
346 int numRegions = NB_SERVERS * 20;
347 int regionReplication = 10;
348 String tableName = htd.getTableName().getNameAsString() + "2";
349 htd = HTU.createTableDescriptor(tableName);
350 htd.setRegionReplication(regionReplication);
351
352
353 byte[] startKey = Bytes.toBytes("aaa");
354 byte[] endKey = Bytes.toBytes("zzz");
355 byte[][] splits = HTU.getRegionSplitStartKeys(startKey, endKey, numRegions);
356 HTU.getHBaseAdmin().createTable(htd, startKey, endKey, numRegions);
357
358 try (Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
359 Table table = connection.getTable(htd.getTableName())) {
360
361 for (int i = 1; i < splits.length; i++) {
362 for (int j = 0; j < regionReplication; j++) {
363 Get get = new Get(splits[i]);
364 get.setConsistency(Consistency.TIMELINE);
365 get.setReplicaId(j);
366 table.get(get);
367 }
368 }
369 }
370
371 HTU.deleteTableIfAny(TableName.valueOf(tableName));
372 }
373 }