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.assertTrue;
23
24 import java.net.URLEncoder;
25 import java.util.ArrayList;
26 import java.util.Collection;
27 import java.util.List;
28 import java.util.SortedMap;
29 import java.util.SortedSet;
30 import java.util.TreeSet;
31 import java.util.UUID;
32 import java.util.concurrent.CountDownLatch;
33 import java.util.concurrent.atomic.AtomicLong;
34
35 import org.apache.commons.logging.Log;
36 import org.apache.commons.logging.LogFactory;
37 import org.apache.hadoop.conf.Configuration;
38 import org.apache.hadoop.fs.FileSystem;
39 import org.apache.hadoop.fs.Path;
40 import org.apache.hadoop.hbase.ChoreService;
41 import org.apache.hadoop.hbase.ClusterId;
42 import org.apache.hadoop.hbase.CoordinatedStateManager;
43 import org.apache.hadoop.hbase.HBaseConfiguration;
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.Server;
51 import org.apache.hadoop.hbase.ServerName;
52 import org.apache.hadoop.hbase.TableName;
53 import org.apache.hadoop.hbase.client.ClusterConnection;
54 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
55 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
56 import org.apache.hadoop.hbase.replication.ReplicationFactory;
57 import org.apache.hadoop.hbase.replication.ReplicationPeers;
58 import org.apache.hadoop.hbase.replication.ReplicationQueueInfo;
59 import org.apache.hadoop.hbase.replication.ReplicationQueues;
60 import org.apache.hadoop.hbase.replication.ReplicationQueuesClient;
61 import org.apache.hadoop.hbase.replication.ReplicationSourceDummy;
62 import org.apache.hadoop.hbase.replication.ReplicationStateZKBase;
63 import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceManager.NodeFailoverWorker;
64 import org.apache.hadoop.hbase.testclassification.MediumTests;
65 import org.apache.hadoop.hbase.util.Bytes;
66 import org.apache.hadoop.hbase.util.FSUtils;
67 import org.apache.hadoop.hbase.wal.WAL;
68 import org.apache.hadoop.hbase.wal.WALFactory;
69 import org.apache.hadoop.hbase.wal.WALKey;
70 import org.apache.hadoop.hbase.zookeeper.MetaTableLocator;
71 import org.apache.hadoop.hbase.zookeeper.ZKClusterId;
72 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
73 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
74 import org.junit.After;
75 import org.junit.AfterClass;
76 import org.junit.Before;
77 import org.junit.BeforeClass;
78 import org.junit.Test;
79 import org.junit.experimental.categories.Category;
80
81 import com.google.common.collect.Sets;
82
83 @Category(MediumTests.class)
84 public class TestReplicationSourceManager {
85
86 private static final Log LOG =
87 LogFactory.getLog(TestReplicationSourceManager.class);
88
89 private static Configuration conf;
90
91 private static HBaseTestingUtility utility;
92
93 private static Replication replication;
94
95 private static ReplicationSourceManager manager;
96
97 private static ZooKeeperWatcher zkw;
98
99 private static HTableDescriptor htd;
100
101 private static HRegionInfo hri;
102
103 private static final byte[] r1 = Bytes.toBytes("r1");
104
105 private static final byte[] r2 = Bytes.toBytes("r2");
106
107 private static final byte[] f1 = Bytes.toBytes("f1");
108
109 private static final TableName test =
110 TableName.valueOf("test");
111
112 private static final String slaveId = "1";
113
114 private static FileSystem fs;
115
116 private static Path oldLogDir;
117
118 private static Path logDir;
119
120 private static CountDownLatch latch;
121
122 private static List<String> files = new ArrayList<String>();
123
124 @BeforeClass
125 public static void setUpBeforeClass() throws Exception {
126
127 conf = HBaseConfiguration.create();
128 conf.set("replication.replicationsource.implementation",
129 ReplicationSourceDummy.class.getCanonicalName());
130 conf.setBoolean(HConstants.REPLICATION_ENABLE_KEY,
131 HConstants.REPLICATION_ENABLE_DEFAULT);
132 conf.setLong("replication.sleep.before.failover", 2000);
133 conf.setInt("replication.source.maxretriesmultiplier", 10);
134 utility = new HBaseTestingUtility(conf);
135 utility.startMiniZKCluster();
136
137 zkw = new ZooKeeperWatcher(conf, "test", null);
138 ZKUtil.createWithParents(zkw, "/hbase/replication");
139 ZKUtil.createWithParents(zkw, "/hbase/replication/peers/1");
140 ZKUtil.setData(zkw, "/hbase/replication/peers/1",
141 Bytes.toBytes(conf.get(HConstants.ZOOKEEPER_QUORUM) + ":"
142 + conf.get(HConstants.ZOOKEEPER_CLIENT_PORT) + ":/1"));
143 ZKUtil.createWithParents(zkw, "/hbase/replication/peers/1/peer-state");
144 ZKUtil.setData(zkw, "/hbase/replication/peers/1/peer-state",
145 ReplicationStateZKBase.ENABLED_ZNODE_BYTES);
146 ZKUtil.createWithParents(zkw, "/hbase/replication/state");
147 ZKUtil.setData(zkw, "/hbase/replication/state", ReplicationStateZKBase.ENABLED_ZNODE_BYTES);
148
149 ZKClusterId.setClusterId(zkw, new ClusterId());
150 FSUtils.setRootDir(utility.getConfiguration(), utility.getDataTestDir());
151 fs = FileSystem.get(conf);
152 oldLogDir = new Path(utility.getDataTestDir(),
153 HConstants.HREGION_OLDLOGDIR_NAME);
154 logDir = new Path(utility.getDataTestDir(),
155 HConstants.HREGION_LOGDIR_NAME);
156 replication = new Replication(new DummyServer(), fs, logDir, oldLogDir);
157 manager = replication.getReplicationManager();
158
159 manager.addSource(slaveId);
160
161 htd = new HTableDescriptor(test);
162 HColumnDescriptor col = new HColumnDescriptor("f1");
163 col.setScope(HConstants.REPLICATION_SCOPE_GLOBAL);
164 htd.addFamily(col);
165 col = new HColumnDescriptor("f2");
166 col.setScope(HConstants.REPLICATION_SCOPE_LOCAL);
167 htd.addFamily(col);
168
169 hri = new HRegionInfo(htd.getTableName(), r1, r2);
170 }
171
172 @AfterClass
173 public static void tearDownAfterClass() throws Exception {
174 manager.join();
175 utility.shutdownMiniCluster();
176 }
177
178 @Before
179 public void setUp() throws Exception {
180 fs.delete(logDir, true);
181 fs.delete(oldLogDir, true);
182 }
183
184 @After
185 public void tearDown() throws Exception {
186 setUp();
187 }
188
189 @Test
190 public void testLogRoll() throws Exception {
191 long seq = 0;
192 long baseline = 1000;
193 long time = baseline;
194 KeyValue kv = new KeyValue(r1, f1, r1);
195 WALEdit edit = new WALEdit();
196 edit.add(kv);
197
198 List<WALActionsListener> listeners = new ArrayList<WALActionsListener>();
199 listeners.add(replication);
200 final WALFactory wals = new WALFactory(utility.getConfiguration(), listeners,
201 URLEncoder.encode("regionserver:60020", "UTF8"));
202 final WAL wal = wals.getWAL(hri.getEncodedNameAsBytes());
203 final AtomicLong sequenceId = new AtomicLong(1);
204 manager.init();
205 HTableDescriptor htd = new HTableDescriptor();
206 htd.addFamily(new HColumnDescriptor(f1));
207
208 for(long i = 1; i < 101; i++) {
209 if(i > 1 && i % 20 == 0) {
210 wal.rollWriter();
211 }
212 LOG.info(i);
213 final long txid = wal.append(htd, hri, new WALKey(hri.getEncodedNameAsBytes(), test,
214 System.currentTimeMillis()), edit, sequenceId, true ,null);
215 wal.sync(txid);
216 }
217
218
219
220 LOG.info(baseline + " and " + time);
221 baseline += 101;
222 time = baseline;
223 LOG.info(baseline + " and " + time);
224
225 for (int i = 0; i < 3; i++) {
226 wal.append(htd, hri, new WALKey(hri.getEncodedNameAsBytes(), test,
227 System.currentTimeMillis()), edit, sequenceId, true, null);
228 }
229 wal.sync();
230
231 assertEquals(6, manager.getWALs().get(slaveId).size());
232
233 wal.rollWriter();
234
235 manager.logPositionAndCleanOldLogs(manager.getSources().get(0).getCurrentPath(),
236 "1", 0, false, false);
237
238 wal.append(htd, hri, new WALKey(hri.getEncodedNameAsBytes(), test,
239 System.currentTimeMillis()), edit, sequenceId, true, null);
240 wal.sync();
241
242 assertEquals(1, manager.getWALs().size());
243
244
245
246 }
247
248 @Test
249 public void testClaimQueues() throws Exception {
250 LOG.debug("testNodeFailoverWorkerCopyQueuesFromRSUsingMulti");
251 conf.setBoolean(HConstants.ZOOKEEPER_USEMULTI, true);
252 final Server server = new DummyServer("hostname0.example.org");
253 ReplicationQueues rq =
254 ReplicationFactory.getReplicationQueues(server.getZooKeeper(), server.getConfiguration(),
255 server);
256 rq.init(server.getServerName().toString());
257
258 files.add("log1");
259 files.add("log2");
260 for (String file : files) {
261 rq.addLog("1", file);
262 }
263
264 Server s1 = new DummyServer("dummyserver1.example.org");
265 Server s2 = new DummyServer("dummyserver2.example.org");
266 Server s3 = new DummyServer("dummyserver3.example.org");
267
268
269 DummyNodeFailoverWorker w1 = new DummyNodeFailoverWorker(
270 server.getServerName().getServerName(), s1);
271 DummyNodeFailoverWorker w2 = new DummyNodeFailoverWorker(
272 server.getServerName().getServerName(), s2);
273 DummyNodeFailoverWorker w3 = new DummyNodeFailoverWorker(
274 server.getServerName().getServerName(), s3);
275
276 latch = new CountDownLatch(3);
277
278 w1.start();
279 w2.start();
280 w3.start();
281
282 int populatedMap = 0;
283
284 latch.await();
285 populatedMap += w1.isLogZnodesMapPopulated() + w2.isLogZnodesMapPopulated()
286 + w3.isLogZnodesMapPopulated();
287 assertEquals(1, populatedMap);
288 server.abort("", null);
289 }
290
291 @Test
292 public void testCleanupFailoverQueues() throws Exception {
293 final Server server = new DummyServer("hostname1.example.org");
294 ReplicationQueues rq =
295 ReplicationFactory.getReplicationQueues(server.getZooKeeper(), server.getConfiguration(),
296 server);
297 rq.init(server.getServerName().toString());
298
299 SortedSet<String> files = new TreeSet<String>();
300 files.add("log1");
301 files.add("log2");
302 for (String file : files) {
303 rq.addLog("1", file);
304 }
305 Server s1 = new DummyServer("dummyserver1.example.org");
306 ReplicationQueues rq1 =
307 ReplicationFactory.getReplicationQueues(s1.getZooKeeper(), s1.getConfiguration(), s1);
308 rq1.init(s1.getServerName().toString());
309 ReplicationPeers rp1 =
310 ReplicationFactory.getReplicationPeers(s1.getZooKeeper(), s1.getConfiguration(), s1);
311 rp1.init();
312 NodeFailoverWorker w1 =
313 manager.new NodeFailoverWorker(server.getServerName().getServerName(), rq1, rp1, new UUID(
314 new Long(1), new Long(2)));
315 w1.start();
316 w1.join(5000);
317 assertEquals(1, manager.getWalsByIdRecoveredQueues().size());
318 String id = "1-" + server.getServerName().getServerName();
319 assertEquals(files, manager.getWalsByIdRecoveredQueues().get(id));
320 manager.cleanOldLogs("log2", id, true);
321
322 assertEquals(Sets.newHashSet("log2"), manager.getWalsByIdRecoveredQueues().get(id));
323 }
324
325 @Test
326 public void testNodeFailoverDeadServerParsing() throws Exception {
327 LOG.debug("testNodeFailoverDeadServerParsing");
328 conf.setBoolean(HConstants.ZOOKEEPER_USEMULTI, true);
329 final Server server = new DummyServer("ec2-54-234-230-108.compute-1.amazonaws.com");
330 ReplicationQueues repQueues =
331 ReplicationFactory.getReplicationQueues(server.getZooKeeper(), conf, server);
332 repQueues.init(server.getServerName().toString());
333
334 files.add("log1");
335 files.add("log2");
336 for (String file : files) {
337 repQueues.addLog("1", file);
338 }
339
340
341 Server s1 = new DummyServer("ip-10-8-101-114.ec2.internal");
342 Server s2 = new DummyServer("ec2-107-20-52-47.compute-1.amazonaws.com");
343 Server s3 = new DummyServer("ec2-23-20-187-167.compute-1.amazonaws.com");
344
345
346 ReplicationQueues rq1 =
347 ReplicationFactory.getReplicationQueues(s1.getZooKeeper(), s1.getConfiguration(), s1);
348 rq1.init(s1.getServerName().toString());
349 SortedMap<String, SortedSet<String>> testMap =
350 rq1.claimQueues(server.getServerName().getServerName());
351 ReplicationQueues rq2 =
352 ReplicationFactory.getReplicationQueues(s2.getZooKeeper(), s2.getConfiguration(), s2);
353 rq2.init(s2.getServerName().toString());
354 testMap = rq2.claimQueues(s1.getServerName().getServerName());
355 ReplicationQueues rq3 =
356 ReplicationFactory.getReplicationQueues(s3.getZooKeeper(), s3.getConfiguration(), s3);
357 rq3.init(s3.getServerName().toString());
358 testMap = rq3.claimQueues(s2.getServerName().getServerName());
359
360 ReplicationQueueInfo replicationQueueInfo = new ReplicationQueueInfo(testMap.firstKey());
361 List<String> result = replicationQueueInfo.getDeadRegionServers();
362
363
364 assertTrue(result.contains(server.getServerName().getServerName()));
365 assertTrue(result.contains(s1.getServerName().getServerName()));
366 assertTrue(result.contains(s2.getServerName().getServerName()));
367
368 server.abort("", null);
369 }
370
371 @Test
372 public void testFailoverDeadServerCversionChange() throws Exception {
373 LOG.debug("testFailoverDeadServerCversionChange");
374
375 conf.setBoolean(HConstants.ZOOKEEPER_USEMULTI, true);
376 final Server s0 = new DummyServer("cversion-change0.example.org");
377 ReplicationQueues repQueues =
378 ReplicationFactory.getReplicationQueues(s0.getZooKeeper(), conf, s0);
379 repQueues.init(s0.getServerName().toString());
380
381 files.add("log1");
382 files.add("log2");
383 for (String file : files) {
384 repQueues.addLog("1", file);
385 }
386
387 Server s1 = new DummyServer("cversion-change1.example.org");
388 ReplicationQueues rq1 =
389 ReplicationFactory.getReplicationQueues(s1.getZooKeeper(), s1.getConfiguration(), s1);
390 rq1.init(s1.getServerName().toString());
391
392 ReplicationQueuesClient client =
393 ReplicationFactory.getReplicationQueuesClient(s1.getZooKeeper(), s1.getConfiguration(), s1);
394
395 int v0 = client.getQueuesZNodeCversion();
396 rq1.claimQueues(s0.getServerName().getServerName());
397 int v1 = client.getQueuesZNodeCversion();
398
399 assertEquals(v0 + 1, v1);
400
401 s0.abort("", null);
402 }
403
404 static class DummyNodeFailoverWorker extends Thread {
405 private SortedMap<String, SortedSet<String>> logZnodesMap;
406 Server server;
407 private String deadRsZnode;
408 ReplicationQueues rq;
409
410 public DummyNodeFailoverWorker(String znode, Server s) throws Exception {
411 this.deadRsZnode = znode;
412 this.server = s;
413 this.rq =
414 ReplicationFactory.getReplicationQueues(server.getZooKeeper(), server.getConfiguration(),
415 server);
416 this.rq.init(this.server.getServerName().toString());
417 }
418
419 @Override
420 public void run() {
421 try {
422 logZnodesMap = rq.claimQueues(deadRsZnode);
423 server.abort("Done with testing", null);
424 } catch (Exception e) {
425 LOG.error("Got exception while running NodeFailoverWorker", e);
426 } finally {
427 latch.countDown();
428 }
429 }
430
431
432
433
434 private int isLogZnodesMapPopulated() {
435 Collection<SortedSet<String>> sets = logZnodesMap.values();
436 if (sets.size() > 1) {
437 throw new RuntimeException("unexpected size of logZnodesMap: " + sets.size());
438 }
439 if (sets.size() == 1) {
440 SortedSet<String> s = sets.iterator().next();
441 for (String file : files) {
442
443 if (!s.contains(file)) {
444 return 0;
445 }
446 }
447 return 1;
448 }
449 return 0;
450 }
451 }
452
453 static class DummyServer implements Server {
454 String hostname;
455
456 DummyServer() {
457 hostname = "hostname.example.org";
458 }
459
460 DummyServer(String hostname) {
461 this.hostname = hostname;
462 }
463
464 @Override
465 public Configuration getConfiguration() {
466 return conf;
467 }
468
469 @Override
470 public ZooKeeperWatcher getZooKeeper() {
471 return zkw;
472 }
473
474 @Override
475 public CoordinatedStateManager getCoordinatedStateManager() {
476 return null;
477 }
478 @Override
479 public ClusterConnection getConnection() {
480 return null;
481 }
482
483 @Override
484 public MetaTableLocator getMetaTableLocator() {
485 return null;
486 }
487
488 @Override
489 public ServerName getServerName() {
490 return ServerName.valueOf(hostname, 1234, 1L);
491 }
492
493 @Override
494 public void abort(String why, Throwable e) {
495
496 }
497
498 @Override
499 public boolean isAborted() {
500 return false;
501 }
502
503 @Override
504 public void stop(String why) {
505
506 }
507
508 @Override
509 public boolean isStopped() {
510 return false;
511 }
512
513 @Override
514 public ChoreService getChoreService() {
515 return null;
516 }
517 }
518 }