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.assertFalse;
23 import static org.junit.Assert.assertNotEquals;
24 import static org.junit.Assert.assertNull;
25 import static org.junit.Assert.assertTrue;
26
27 import com.google.common.collect.Sets;
28
29 import java.io.IOException;
30 import java.net.URLEncoder;
31 import java.util.ArrayList;
32 import java.util.Collection;
33 import java.util.HashMap;
34 import java.util.List;
35 import java.util.Map;
36 import java.util.NavigableMap;
37 import java.util.SortedMap;
38 import java.util.SortedSet;
39 import java.util.TreeMap;
40 import java.util.TreeSet;
41 import java.util.UUID;
42 import java.util.concurrent.CountDownLatch;
43
44 import org.apache.commons.logging.Log;
45 import org.apache.commons.logging.LogFactory;
46 import org.apache.hadoop.conf.Configuration;
47 import org.apache.hadoop.fs.FileSystem;
48 import org.apache.hadoop.fs.Path;
49 import org.apache.hadoop.hbase.ChoreService;
50 import org.apache.hadoop.hbase.ClusterId;
51 import org.apache.hadoop.hbase.CoordinatedStateManager;
52 import org.apache.hadoop.hbase.HBaseConfiguration;
53 import org.apache.hadoop.hbase.HBaseTestingUtility;
54 import org.apache.hadoop.hbase.HColumnDescriptor;
55 import org.apache.hadoop.hbase.HConstants;
56 import org.apache.hadoop.hbase.HRegionInfo;
57 import org.apache.hadoop.hbase.HTableDescriptor;
58 import org.apache.hadoop.hbase.KeyValue;
59 import org.apache.hadoop.hbase.Server;
60 import org.apache.hadoop.hbase.ServerName;
61 import org.apache.hadoop.hbase.Stoppable;
62 import org.apache.hadoop.hbase.TableName;
63 import org.apache.hadoop.hbase.Waiter;
64 import org.apache.hadoop.hbase.client.ClusterConnection;
65 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
66 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.BulkLoadDescriptor;
67 import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl;
68 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
69 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
70 import org.apache.hadoop.hbase.replication.ReplicationEndpoint;
71 import org.apache.hadoop.hbase.replication.ReplicationFactory;
72 import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
73 import org.apache.hadoop.hbase.replication.ReplicationPeers;
74 import org.apache.hadoop.hbase.replication.ReplicationQueueInfo;
75 import org.apache.hadoop.hbase.replication.ReplicationQueues;
76 import org.apache.hadoop.hbase.replication.ReplicationQueuesClient;
77 import org.apache.hadoop.hbase.replication.ReplicationSourceDummy;
78 import org.apache.hadoop.hbase.replication.ReplicationStateZKBase;
79 import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceManager.NodeFailoverWorker;
80 import org.apache.hadoop.hbase.testclassification.MediumTests;
81 import org.apache.hadoop.hbase.util.ByteStringer;
82 import org.apache.hadoop.hbase.util.Bytes;
83 import org.apache.hadoop.hbase.util.FSUtils;
84 import org.apache.hadoop.hbase.util.Pair;
85 import org.apache.hadoop.hbase.wal.WAL;
86 import org.apache.hadoop.hbase.wal.WALFactory;
87 import org.apache.hadoop.hbase.wal.WALKey;
88 import org.apache.hadoop.hbase.zookeeper.MetaTableLocator;
89 import org.apache.hadoop.hbase.zookeeper.ZKClusterId;
90 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
91 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
92 import org.junit.After;
93 import org.junit.AfterClass;
94 import org.junit.Before;
95 import org.junit.BeforeClass;
96 import org.junit.Rule;
97 import org.junit.Test;
98 import org.junit.experimental.categories.Category;
99 import org.junit.rules.TestName;
100
101 @Category(MediumTests.class)
102 public class TestReplicationSourceManager {
103
104 private static final Log LOG =
105 LogFactory.getLog(TestReplicationSourceManager.class);
106
107 private static Configuration conf;
108
109 private static HBaseTestingUtility utility;
110
111 private static Replication replication;
112
113 private static ReplicationSourceManager manager;
114
115 private static ZooKeeperWatcher zkw;
116
117 private static HTableDescriptor htd;
118
119 private static HRegionInfo hri;
120
121 private static final byte[] r1 = Bytes.toBytes("r1");
122
123 private static final byte[] r2 = Bytes.toBytes("r2");
124
125 private static final byte[] f1 = Bytes.toBytes("f1");
126
127 private static final byte[] f2 = Bytes.toBytes("f2");
128
129 private static final TableName test =
130 TableName.valueOf("test");
131
132 private static final String slaveId = "1";
133
134 private static FileSystem fs;
135
136 private static Path oldLogDir;
137
138 private static Path logDir;
139
140 private static CountDownLatch latch;
141
142 private static List<String> files = new ArrayList<String>();
143
144 @BeforeClass
145 public static void setUpBeforeClass() throws Exception {
146
147 conf = HBaseConfiguration.create();
148 conf.set("replication.replicationsource.implementation",
149 ReplicationSourceDummy.class.getCanonicalName());
150 conf.setBoolean(HConstants.REPLICATION_ENABLE_KEY,
151 HConstants.REPLICATION_ENABLE_DEFAULT);
152 conf.setLong("replication.sleep.before.failover", 2000);
153 conf.setInt("replication.source.maxretriesmultiplier", 10);
154 utility = new HBaseTestingUtility(conf);
155 utility.startMiniZKCluster();
156
157 zkw = new ZooKeeperWatcher(conf, "test", null);
158 ZKUtil.createWithParents(zkw, "/hbase/replication");
159 ZKUtil.createWithParents(zkw, "/hbase/replication/peers/1");
160 ZKUtil.setData(zkw, "/hbase/replication/peers/1",
161 Bytes.toBytes(conf.get(HConstants.ZOOKEEPER_QUORUM) + ":"
162 + conf.get(HConstants.ZOOKEEPER_CLIENT_PORT) + ":/1"));
163 ZKUtil.createWithParents(zkw, "/hbase/replication/peers/1/peer-state");
164 ZKUtil.setData(zkw, "/hbase/replication/peers/1/peer-state",
165 ReplicationStateZKBase.ENABLED_ZNODE_BYTES);
166 ZKUtil.createWithParents(zkw, "/hbase/replication/state");
167 ZKUtil.setData(zkw, "/hbase/replication/state", ReplicationStateZKBase.ENABLED_ZNODE_BYTES);
168
169 ZKClusterId.setClusterId(zkw, new ClusterId());
170 FSUtils.setRootDir(utility.getConfiguration(), utility.getDataTestDir());
171 fs = FileSystem.get(conf);
172 oldLogDir = new Path(utility.getDataTestDir(),
173 HConstants.HREGION_OLDLOGDIR_NAME);
174 logDir = new Path(utility.getDataTestDir(),
175 HConstants.HREGION_LOGDIR_NAME);
176 replication = new Replication(new DummyServer(), fs, logDir, oldLogDir);
177 manager = replication.getReplicationManager();
178
179 manager.addSource(slaveId);
180
181 htd = new HTableDescriptor(test);
182 HColumnDescriptor col = new HColumnDescriptor(f1);
183 col.setScope(HConstants.REPLICATION_SCOPE_GLOBAL);
184 htd.addFamily(col);
185 col = new HColumnDescriptor(f2);
186 col.setScope(HConstants.REPLICATION_SCOPE_LOCAL);
187 htd.addFamily(col);
188
189 hri = new HRegionInfo(htd.getTableName(), r1, r2);
190 }
191
192 @AfterClass
193 public static void tearDownAfterClass() throws Exception {
194 manager.join();
195 utility.shutdownMiniCluster();
196 }
197
198 @Rule
199 public TestName testName = new TestName();
200
201 private void cleanLogDir() throws IOException {
202 fs.delete(logDir, true);
203 fs.delete(oldLogDir, true);
204 }
205
206 @Before
207 public void setUp() throws Exception {
208 LOG.info("Start " + testName.getMethodName());
209 cleanLogDir();
210 }
211
212 @After
213 public void tearDown() throws Exception {
214 LOG.info("End " + testName.getMethodName());
215 cleanLogDir();
216 }
217
218 @Test
219 public void testLogRoll() throws Exception {
220 long baseline = 1000;
221 long time = baseline;
222 MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl();
223 KeyValue kv = new KeyValue(r1, f1, r1);
224 WALEdit edit = new WALEdit();
225 edit.add(kv);
226
227 List<WALActionsListener> listeners = new ArrayList<WALActionsListener>();
228 listeners.add(replication);
229 final WALFactory wals = new WALFactory(utility.getConfiguration(), listeners,
230 URLEncoder.encode("regionserver:60020", "UTF8"));
231 final WAL wal = wals.getWAL(hri.getEncodedNameAsBytes(), hri.getTable().getNamespace());
232 manager.init();
233 HTableDescriptor htd = new HTableDescriptor();
234 htd.addFamily(new HColumnDescriptor(f1));
235
236 for(long i = 1; i < 101; i++) {
237 if(i > 1 && i % 20 == 0) {
238 wal.rollWriter();
239 }
240 LOG.info(i);
241 final long txid = wal.append(htd,
242 hri,
243 new WALKey(hri.getEncodedNameAsBytes(), test, System.currentTimeMillis(), mvcc),
244 edit,
245 true);
246 wal.sync(txid);
247 }
248
249
250
251 LOG.info(baseline + " and " + time);
252 baseline += 101;
253 time = baseline;
254 LOG.info(baseline + " and " + time);
255
256 for (int i = 0; i < 3; i++) {
257 wal.append(htd, hri,
258 new WALKey(hri.getEncodedNameAsBytes(), test, System.currentTimeMillis(), mvcc),
259 edit,
260 true);
261 }
262 wal.sync();
263
264 int logNumber = 0;
265 for (Map.Entry<String, SortedSet<String>> entry : manager.getWALs().get(slaveId).entrySet()) {
266 logNumber += entry.getValue().size();
267 }
268 assertEquals(6, logNumber);
269
270 wal.rollWriter();
271
272 manager.logPositionAndCleanOldLogs(manager.getSources().get(0).getCurrentPath(),
273 "1", 0, false, false);
274
275 wal.append(htd, hri,
276 new WALKey(hri.getEncodedNameAsBytes(), test, System.currentTimeMillis(), mvcc),
277 edit,
278 true);
279 wal.sync();
280
281 assertEquals(1, manager.getWALs().size());
282
283
284
285 }
286
287 @Test
288 public void testClaimQueues() throws Exception {
289 conf.setBoolean(HConstants.ZOOKEEPER_USEMULTI, true);
290 final Server server = new DummyServer("hostname0.example.org");
291 ReplicationQueues rq =
292 ReplicationFactory.getReplicationQueues(server.getZooKeeper(), server.getConfiguration(),
293 server);
294 rq.init(server.getServerName().toString());
295
296 files.add("log1");
297 files.add("log2");
298 for (String file : files) {
299 rq.addLog("1", file);
300 }
301
302 Server s1 = new DummyServer("dummyserver1.example.org");
303 Server s2 = new DummyServer("dummyserver2.example.org");
304 Server s3 = new DummyServer("dummyserver3.example.org");
305
306
307 DummyNodeFailoverWorker w1 = new DummyNodeFailoverWorker(
308 server.getServerName().getServerName(), s1);
309 DummyNodeFailoverWorker w2 = new DummyNodeFailoverWorker(
310 server.getServerName().getServerName(), s2);
311 DummyNodeFailoverWorker w3 = new DummyNodeFailoverWorker(
312 server.getServerName().getServerName(), s3);
313
314 latch = new CountDownLatch(3);
315
316 w1.start();
317 w2.start();
318 w3.start();
319
320 int populatedMap = 0;
321
322 latch.await();
323 populatedMap += w1.isLogZnodesMapPopulated() + w2.isLogZnodesMapPopulated()
324 + w3.isLogZnodesMapPopulated();
325 assertEquals(1, populatedMap);
326 server.abort("", null);
327 }
328
329 @Test
330 public void testCleanupFailoverQueues() throws Exception {
331 final Server server = new DummyServer("hostname1.example.org");
332 ReplicationQueues rq =
333 ReplicationFactory.getReplicationQueues(server.getZooKeeper(), server.getConfiguration(),
334 server);
335 rq.init(server.getServerName().toString());
336
337 SortedSet<String> files = new TreeSet<String>();
338 String group = "testgroup";
339 String file1 = group + ".log1";
340 String file2 = group + ".log2";
341 files.add(file1);
342 files.add(file2);
343 for (String file : files) {
344 rq.addLog("1", file);
345 }
346 Server s1 = new DummyServer("dummyserver1.example.org");
347 ReplicationQueues rq1 =
348 ReplicationFactory.getReplicationQueues(s1.getZooKeeper(), s1.getConfiguration(), s1);
349 rq1.init(s1.getServerName().toString());
350 ReplicationPeers rp1 =
351 ReplicationFactory.getReplicationPeers(s1.getZooKeeper(), s1.getConfiguration(), s1);
352 rp1.init();
353 NodeFailoverWorker w1 =
354 manager.new NodeFailoverWorker(server.getServerName().getServerName(), rq1, rp1, new UUID(
355 new Long(1), new Long(2)));
356 w1.start();
357 w1.join(10000);
358 assertEquals(1, manager.getWalsByIdRecoveredQueues().size());
359 String id = "1-" + server.getServerName().getServerName();
360 assertEquals(files, manager.getWalsByIdRecoveredQueues().get(id).get(group));
361 manager.cleanOldLogs(file2, id, true);
362
363 assertEquals(Sets.newHashSet(file2), manager.getWalsByIdRecoveredQueues().get(id).get(group));
364 }
365
366 @Test
367 public void testNodeFailoverDeadServerParsing() throws Exception {
368 LOG.debug("testNodeFailoverDeadServerParsing");
369 conf.setBoolean(HConstants.ZOOKEEPER_USEMULTI, true);
370 final Server server = new DummyServer("ec2-54-234-230-108.compute-1.amazonaws.com");
371 ReplicationQueues repQueues =
372 ReplicationFactory.getReplicationQueues(server.getZooKeeper(), conf, server);
373 repQueues.init(server.getServerName().toString());
374
375 files.add("log1");
376 files.add("log2");
377 for (String file : files) {
378 repQueues.addLog("1", file);
379 }
380
381
382 Server s1 = new DummyServer("ip-10-8-101-114.ec2.internal");
383 Server s2 = new DummyServer("ec2-107-20-52-47.compute-1.amazonaws.com");
384 Server s3 = new DummyServer("ec2-23-20-187-167.compute-1.amazonaws.com");
385
386
387 ReplicationQueues rq1 =
388 ReplicationFactory.getReplicationQueues(s1.getZooKeeper(), s1.getConfiguration(), s1);
389 rq1.init(s1.getServerName().toString());
390 String serverName = server.getServerName().getServerName();
391 List<String> unclaimed = rq1.getUnClaimedQueueIds(serverName);
392 rq1.claimQueue(serverName, unclaimed.get(0)).getSecond();
393 rq1.removeReplicatorIfQueueIsEmpty(unclaimed.get(0));
394
395 ReplicationQueues rq2 =
396 ReplicationFactory.getReplicationQueues(s2.getZooKeeper(), s2.getConfiguration(), s2);
397 rq2.init(s2.getServerName().toString());
398 serverName = s1.getServerName().getServerName();
399 unclaimed = rq2.getUnClaimedQueueIds(serverName);
400 rq2.claimQueue(serverName, unclaimed.get(0)).getSecond();
401 rq2.removeReplicatorIfQueueIsEmpty(unclaimed.get(0));
402 ReplicationQueues rq3 =
403 ReplicationFactory.getReplicationQueues(s3.getZooKeeper(), s3.getConfiguration(), s3);
404 rq3.init(s3.getServerName().toString());
405 serverName = s2.getServerName().getServerName();
406 unclaimed = rq3.getUnClaimedQueueIds(serverName);
407 String queue3 = rq3.claimQueue(serverName, unclaimed.get(0)).getFirst();
408 rq3.removeReplicatorIfQueueIsEmpty(unclaimed.get(0));
409 ReplicationQueueInfo replicationQueueInfo = new ReplicationQueueInfo(queue3);
410 List<String> result = replicationQueueInfo.getDeadRegionServers();
411
412
413 assertTrue(result.contains(server.getServerName().getServerName()));
414 assertTrue(result.contains(s1.getServerName().getServerName()));
415 assertTrue(result.contains(s2.getServerName().getServerName()));
416
417 server.abort("", null);
418 }
419
420 @Test
421 public void testFailoverDeadServerCversionChange() throws Exception {
422 LOG.debug("testFailoverDeadServerCversionChange");
423
424 conf.setBoolean(HConstants.ZOOKEEPER_USEMULTI, true);
425 final Server s0 = new DummyServer("cversion-change0.example.org");
426 ReplicationQueues repQueues =
427 ReplicationFactory.getReplicationQueues(s0.getZooKeeper(), conf, s0);
428 repQueues.init(s0.getServerName().toString());
429
430 files.add("log1");
431 files.add("log2");
432 for (String file : files) {
433 repQueues.addLog("1", file);
434 }
435
436 Server s1 = new DummyServer("cversion-change1.example.org");
437 ReplicationQueues rq1 =
438 ReplicationFactory.getReplicationQueues(s1.getZooKeeper(), s1.getConfiguration(), s1);
439 rq1.init(s1.getServerName().toString());
440
441 ReplicationQueuesClient client =
442 ReplicationFactory.getReplicationQueuesClient(s1.getZooKeeper(), s1.getConfiguration(), s1);
443
444 int v0 = client.getQueuesZNodeCversion();
445 List<String> queues = rq1.getUnClaimedQueueIds(s0.getServerName().getServerName());
446 for(String queue : queues) {
447 rq1.claimQueue(s0.getServerName().getServerName(), queue);
448 }
449 rq1.removeReplicatorIfQueueIsEmpty(s0.getServerName().getServerName());
450 int v1 = client.getQueuesZNodeCversion();
451
452 assertEquals(v0 + 1, v1);
453
454 s0.abort("", null);
455 }
456
457 @Test
458 public void testCleanupUnknownPeerZNode() throws Exception {
459 final Server server = new DummyServer("hostname2.example.org");
460 ReplicationQueues rq =
461 ReplicationFactory.getReplicationQueues(server.getZooKeeper(), server.getConfiguration(),
462 server);
463 rq.init(server.getServerName().toString());
464
465
466 String group = "testgroup";
467 rq.addLog("2", group + ".log1");
468 rq.addLog("2", group + ".log2");
469
470 NodeFailoverWorker w1 = manager.new NodeFailoverWorker(server.getServerName().getServerName());
471 w1.run();
472
473
474 for (String peer : manager.getAllQueues()) {
475 assertTrue(peer.startsWith("1"));
476 }
477 }
478
479 @Test
480 public void testBulkLoadWALEditsWithoutBulkLoadReplicationEnabled() throws Exception {
481
482 WALKey logKey = new WALKey();
483
484 WALEdit logEdit = getBulkLoadWALEdit();
485
486
487 Replication.scopeWALEdits(htd, logKey, logEdit, conf, manager);
488
489
490 assertNull("No bulk load entries scope should be added if bulk load replication is diabled.",
491 logKey.getScopes());
492 }
493
494 @Test
495 public void testBulkLoadWALEdits() throws Exception {
496
497 WALKey logKey = new WALKey();
498
499 WALEdit logEdit = getBulkLoadWALEdit();
500
501 Configuration bulkLoadConf = HBaseConfiguration.create(conf);
502 bulkLoadConf.setBoolean(HConstants.REPLICATION_BULKLOAD_ENABLE_KEY, true);
503
504
505 Replication.scopeWALEdits(htd, logKey, logEdit, bulkLoadConf, manager);
506
507 NavigableMap<byte[], Integer> scopes = logKey.getScopes();
508
509 assertTrue("This family scope is set to global, should be part of replication key scopes.",
510 scopes.containsKey(f1));
511
512 assertFalse("This family scope is set to local, should not be part of replication key scopes",
513 scopes.containsKey(f2));
514 }
515
516
517
518
519
520
521
522
523 @Test
524 public void testPeerRemovalCleanup() throws Exception{
525 String replicationSourceImplName = conf.get("replication.replicationsource.implementation");
526 try {
527 DummyServer server = new DummyServer();
528 final ReplicationQueues rq =
529 ReplicationFactory.getReplicationQueues(server.getZooKeeper(), server.getConfiguration(),
530 server);
531 rq.init(server.getServerName().toString());
532
533
534 conf.set("replication.replicationsource.implementation",
535 FailInitializeDummyReplicationSource.class.getName());
536 final ReplicationPeers rp = manager.getReplicationPeers();
537
538 rp.addPeer("FakePeer", new ReplicationPeerConfig().setClusterKey("localhost:1:/hbase"), null);
539
540 Waiter.waitFor(conf, 20000, new Waiter.Predicate<Exception>() {
541 @Override
542 public boolean evaluate() throws Exception {
543 return (rp.getPeer("FakePeer") != null);
544 }
545 });
546
547
548 List<ReplicationSourceInterface> sources = manager.getSources();
549 for (ReplicationSourceInterface source : sources) {
550 assertNotEquals("FakePeer", source.getPeerClusterId());
551 }
552
553
554 manager.removePeer("FakePeer");
555 assertFalse(rq.getAllQueues().contains("FakePeer"));
556 assertNull(rp.getPeer("FakePeer"));
557
558
559 rp.removePeer("FakePeer");
560 Waiter.waitFor(conf, 20000, new Waiter.Predicate<Exception>() {
561 @Override
562 public boolean evaluate() throws Exception {
563 List<String> peers = rp.getAllPeerIds();
564 return (!rq.getAllQueues().contains("FakePeer"))
565 && (rp.getPeer("FakePeer") == null)
566 && (!peers.contains("FakePeer"));
567 }
568 });
569 } finally {
570 conf.set("replication.replicationsource.implementation", replicationSourceImplName);
571 }
572 }
573
574 private WALEdit getBulkLoadWALEdit() {
575
576 Map<byte[], List<Path>> storeFiles = new HashMap<>(1);
577 Map<String, Long> storeFilesSize = new HashMap<>(1);
578 List<Path> p = new ArrayList<>(1);
579 Path hfilePath1 = new Path(Bytes.toString(f1));
580 p.add(hfilePath1);
581 try {
582 storeFilesSize.put(hfilePath1.getName(), fs.getFileStatus(hfilePath1).getLen());
583 } catch (IOException e) {
584 LOG.debug("Failed to calculate the size of hfile " + hfilePath1);
585 storeFilesSize.put(hfilePath1.getName(), 0L);
586 }
587 storeFiles.put(f1, p);
588
589 p = new ArrayList<>(1);
590 Path hfilePath2 = new Path(Bytes.toString(f2));
591 p.add(hfilePath2);
592 try {
593 storeFilesSize.put(hfilePath2.getName(), fs.getFileStatus(hfilePath2).getLen());
594 } catch (IOException e) {
595 LOG.debug("Failed to calculate the size of hfile " + hfilePath2);
596 storeFilesSize.put(hfilePath2.getName(), 0L);
597 }
598 storeFiles.put(f2, p);
599
600
601 BulkLoadDescriptor desc =
602 ProtobufUtil.toBulkLoadDescriptor(hri.getTable(),
603 ByteStringer.wrap(hri.getEncodedNameAsBytes()), storeFiles, storeFilesSize, 1);
604
605
606 WALEdit logEdit = WALEdit.createBulkLoadEvent(hri, desc);
607 return logEdit;
608 }
609
610 static class DummyNodeFailoverWorker extends Thread {
611 private SortedMap<String, SortedSet<String>> logZnodesMap;
612 Server server;
613 private String deadRsZnode;
614 ReplicationQueues rq;
615
616 public DummyNodeFailoverWorker(String znode, Server s) throws Exception {
617 this.deadRsZnode = znode;
618 this.server = s;
619 this.rq =
620 ReplicationFactory.getReplicationQueues(server.getZooKeeper(), server.getConfiguration(),
621 server);
622 this.rq.init(this.server.getServerName().toString());
623 }
624
625 @Override
626 public void run() {
627 try {
628 logZnodesMap = new TreeMap<>();
629 List<String> queues = rq.getUnClaimedQueueIds(deadRsZnode);
630 for(String queue:queues){
631 Pair<String, SortedSet<String>> pair = rq.claimQueue(deadRsZnode, queue);
632 logZnodesMap.put(pair.getFirst(), pair.getSecond());
633 }
634 server.abort("Done with testing", null);
635 } catch (Exception e) {
636 LOG.error("Got exception while running NodeFailoverWorker", e);
637 } finally {
638 latch.countDown();
639 }
640 }
641
642
643
644
645 private int isLogZnodesMapPopulated() {
646 Collection<SortedSet<String>> sets = logZnodesMap.values();
647 if (sets.size() > 1) {
648 throw new RuntimeException("unexpected size of logZnodesMap: " + sets.size());
649 }
650 if (sets.size() == 1) {
651 SortedSet<String> s = sets.iterator().next();
652 for (String file : files) {
653
654 if (!s.contains(file)) {
655 return 0;
656 }
657 }
658 return 1;
659 }
660 return 0;
661 }
662 }
663
664 static class FailInitializeDummyReplicationSource extends ReplicationSourceDummy {
665
666 @Override
667 public void init(Configuration conf, FileSystem fs, ReplicationSourceManager manager,
668 ReplicationQueues rq, ReplicationPeers rp, Stoppable stopper, String peerClusterId,
669 UUID clusterId, ReplicationEndpoint replicationEndpoint, MetricsSource metrics)
670 throws IOException {
671 throw new IOException("Failing deliberately");
672 }
673 }
674
675 static class DummyServer implements Server {
676 String hostname;
677
678 DummyServer() {
679 hostname = "hostname.example.org";
680 }
681
682 DummyServer(String hostname) {
683 this.hostname = hostname;
684 }
685
686 @Override
687 public Configuration getConfiguration() {
688 return conf;
689 }
690
691 @Override
692 public ZooKeeperWatcher getZooKeeper() {
693 return zkw;
694 }
695
696 @Override
697 public CoordinatedStateManager getCoordinatedStateManager() {
698 return null;
699 }
700 @Override
701 public ClusterConnection getConnection() {
702 return null;
703 }
704
705 @Override
706 public MetaTableLocator getMetaTableLocator() {
707 return null;
708 }
709
710 @Override
711 public ServerName getServerName() {
712 return ServerName.valueOf(hostname, 1234, 1L);
713 }
714
715 @Override
716 public void abort(String why, Throwable e) {
717
718 }
719
720 @Override
721 public boolean isAborted() {
722 return false;
723 }
724
725 @Override
726 public void stop(String why) {
727
728 }
729
730 @Override
731 public boolean isStopped() {
732 return false;
733 }
734
735 @Override
736 public ChoreService getChoreService() {
737 return null;
738 }
739 }
740 }