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