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 public void testCleanupUnknownPeerZNode() throws Exception {
460 final Server server = new DummyServer("hostname2.example.org");
461 ReplicationQueues rq =
462 ReplicationFactory.getReplicationQueues(server.getZooKeeper(), server.getConfiguration(),
463 server);
464 rq.init(server.getServerName().toString());
465
466
467 String group = "testgroup";
468 rq.addLog("2", group + ".log1");
469 rq.addLog("2", group + ".log2");
470
471 NodeFailoverWorker w1 = manager.new NodeFailoverWorker(server.getServerName().getServerName());
472 w1.run();
473
474
475 for (String peer : manager.getAllQueues()) {
476 assertTrue(peer.startsWith("1"));
477 }
478 }
479
480 @Test
481 public void testBulkLoadWALEditsWithoutBulkLoadReplicationEnabled() throws Exception {
482
483 WALKey logKey = new WALKey();
484
485 WALEdit logEdit = getBulkLoadWALEdit();
486
487
488 Replication.scopeWALEdits(htd, logKey, logEdit, conf, manager);
489
490
491 assertNull("No bulk load entries scope should be added if bulk load replication is diabled.",
492 logKey.getScopes());
493 }
494
495 @Test
496 public void testBulkLoadWALEdits() throws Exception {
497
498 WALKey logKey = new WALKey();
499
500 WALEdit logEdit = getBulkLoadWALEdit();
501
502 Configuration bulkLoadConf = HBaseConfiguration.create(conf);
503 bulkLoadConf.setBoolean(HConstants.REPLICATION_BULKLOAD_ENABLE_KEY, true);
504
505
506 Replication.scopeWALEdits(htd, logKey, logEdit, bulkLoadConf, manager);
507
508 NavigableMap<byte[], Integer> scopes = logKey.getScopes();
509
510 assertTrue("This family scope is set to global, should be part of replication key scopes.",
511 scopes.containsKey(f1));
512
513 assertFalse("This family scope is set to local, should not be part of replication key scopes",
514 scopes.containsKey(f2));
515 }
516
517
518
519
520
521
522
523
524 @Test
525 public void testPeerRemovalCleanup() throws Exception{
526 String replicationSourceImplName = conf.get("replication.replicationsource.implementation");
527 final String peerId = "FakePeer";
528 final ReplicationPeerConfig peerConfig =
529 new ReplicationPeerConfig().setClusterKey("localhost:1:/hbase");
530 try {
531 DummyServer server = new DummyServer();
532 final ReplicationQueues rq =
533 ReplicationFactory.getReplicationQueues(server.getZooKeeper(), server.getConfiguration(),
534 server);
535 rq.init(server.getServerName().toString());
536
537
538 conf.set("replication.replicationsource.implementation",
539 FailInitializeDummyReplicationSource.class.getName());
540 final ReplicationPeers rp = manager.getReplicationPeers();
541
542
543 addPeerAndWait(peerId, peerConfig, false);
544
545
546 assertNull(manager.getSource(peerId));
547
548
549
550 rq.addLog(peerId, "FakeFile");
551
552
553 removePeerAndWait(peerId);
554 assertFalse(rq.getAllQueues().contains(peerId));
555 } finally {
556 conf.set("replication.replicationsource.implementation", replicationSourceImplName);
557 removePeerAndWait(peerId);
558 }
559 }
560
561 @Test
562 public void testRemovePeerMetricsCleanup() throws Exception {
563 final String peerId = "DummyPeer";
564 final ReplicationPeerConfig peerConfig =
565 new ReplicationPeerConfig().setClusterKey("localhost:1:/hbase");
566 try {
567 addPeerAndWait(peerId, peerConfig, true);
568
569 ReplicationSourceInterface source = manager.getSource(peerId);
570
571 assertNotNull(source);
572
573 Field f = MetricsSource.class.getDeclaredField("globalSourceSource");
574 f.setAccessible(true);
575 MetricsReplicationSourceSource globalSource =
576 (MetricsReplicationSourceSource)f.get(source.getSourceMetrics());
577 int globalLogQueueSizeInitial = globalSource.getSizeOfLogQueue();
578
579
580 source.enqueueLog(new Path("abc"));
581 assertEquals(1, source.getSourceMetrics().getSizeOfLogQueue());
582 assertEquals(1 + globalLogQueueSizeInitial, globalSource.getSizeOfLogQueue());
583
584
585 removePeerAndWait(peerId);
586 assertEquals(globalLogQueueSizeInitial, globalSource.getSizeOfLogQueue());
587
588
589 addPeerAndWait(peerId, peerConfig, true);
590 source = manager.getSource(peerId);
591 assertNotNull(source);
592 assertEquals(0, source.getSourceMetrics().getSizeOfLogQueue());
593 assertEquals(globalLogQueueSizeInitial, globalSource.getSizeOfLogQueue());
594 } finally {
595 removePeerAndWait(peerId);
596 }
597 }
598
599
600
601
602
603
604
605
606 private void addPeerAndWait(final String peerId, final ReplicationPeerConfig peerConfig,
607 final boolean waitForSource) throws Exception {
608 final ReplicationPeers rp = manager.getReplicationPeers();
609 rp.addPeer(peerId, peerConfig, null);
610 Waiter.waitFor(conf, 20000, new Waiter.Predicate<Exception>() {
611 @Override public boolean evaluate() throws Exception {
612 if (waitForSource) {
613 return (manager.getSource(peerId) != null);
614 } else {
615 return (rp.getPeer(peerId) != null);
616 }
617 }
618 });
619 }
620
621
622
623
624
625
626 private void removePeerAndWait(final String peerId) throws Exception {
627 final ReplicationPeers rp = manager.getReplicationPeers();
628 if (rp.getAllPeerIds().contains(peerId)) {
629 rp.removePeer(peerId);
630 }
631 Waiter.waitFor(conf, 20000, new Waiter.Predicate<Exception>() {
632 @Override public boolean evaluate() throws Exception {
633 List<String> peers = rp.getAllPeerIds();
634 return (!manager.getAllQueues().contains(peerId)) && (rp.getPeer(peerId) == null)
635 && (!peers.contains(peerId));
636 }
637 });
638 }
639
640
641 private WALEdit getBulkLoadWALEdit() {
642
643 Map<byte[], List<Path>> storeFiles = new HashMap<>(1);
644 Map<String, Long> storeFilesSize = new HashMap<>(1);
645 List<Path> p = new ArrayList<>(1);
646 Path hfilePath1 = new Path(Bytes.toString(f1));
647 p.add(hfilePath1);
648 try {
649 storeFilesSize.put(hfilePath1.getName(), fs.getFileStatus(hfilePath1).getLen());
650 } catch (IOException e) {
651 LOG.debug("Failed to calculate the size of hfile " + hfilePath1);
652 storeFilesSize.put(hfilePath1.getName(), 0L);
653 }
654 storeFiles.put(f1, p);
655
656 p = new ArrayList<>(1);
657 Path hfilePath2 = new Path(Bytes.toString(f2));
658 p.add(hfilePath2);
659 try {
660 storeFilesSize.put(hfilePath2.getName(), fs.getFileStatus(hfilePath2).getLen());
661 } catch (IOException e) {
662 LOG.debug("Failed to calculate the size of hfile " + hfilePath2);
663 storeFilesSize.put(hfilePath2.getName(), 0L);
664 }
665 storeFiles.put(f2, p);
666
667
668 BulkLoadDescriptor desc =
669 ProtobufUtil.toBulkLoadDescriptor(hri.getTable(),
670 ByteStringer.wrap(hri.getEncodedNameAsBytes()), storeFiles, storeFilesSize, 1);
671
672
673 WALEdit logEdit = WALEdit.createBulkLoadEvent(hri, desc);
674 return logEdit;
675 }
676
677 static class DummyNodeFailoverWorker extends Thread {
678 private SortedMap<String, SortedSet<String>> logZnodesMap;
679 Server server;
680 private String deadRsZnode;
681 ReplicationQueues rq;
682
683 public DummyNodeFailoverWorker(String znode, Server s) throws Exception {
684 this.deadRsZnode = znode;
685 this.server = s;
686 this.rq =
687 ReplicationFactory.getReplicationQueues(server.getZooKeeper(), server.getConfiguration(),
688 server);
689 this.rq.init(this.server.getServerName().toString());
690 }
691
692 @Override
693 public void run() {
694 try {
695 logZnodesMap = new TreeMap<>();
696 List<String> queues = rq.getUnClaimedQueueIds(deadRsZnode);
697 for(String queue:queues){
698 Pair<String, SortedSet<String>> pair = rq.claimQueue(deadRsZnode, queue);
699 logZnodesMap.put(pair.getFirst(), pair.getSecond());
700 }
701 server.abort("Done with testing", null);
702 } catch (Exception e) {
703 LOG.error("Got exception while running NodeFailoverWorker", e);
704 } finally {
705 latch.countDown();
706 }
707 }
708
709
710
711
712 private int isLogZnodesMapPopulated() {
713 Collection<SortedSet<String>> sets = logZnodesMap.values();
714 if (sets.size() > 1) {
715 throw new RuntimeException("unexpected size of logZnodesMap: " + sets.size());
716 }
717 if (sets.size() == 1) {
718 SortedSet<String> s = sets.iterator().next();
719 for (String file : files) {
720
721 if (!s.contains(file)) {
722 return 0;
723 }
724 }
725 return 1;
726 }
727 return 0;
728 }
729 }
730
731 static class FailInitializeDummyReplicationSource extends ReplicationSourceDummy {
732
733 @Override
734 public void init(Configuration conf, FileSystem fs, ReplicationSourceManager manager,
735 ReplicationQueues rq, ReplicationPeers rp, Stoppable stopper, String peerClusterId,
736 UUID clusterId, ReplicationEndpoint replicationEndpoint, MetricsSource metrics)
737 throws IOException {
738 throw new IOException("Failing deliberately");
739 }
740 }
741
742 static class DummyServer implements Server {
743 String hostname;
744
745 DummyServer() {
746 hostname = "hostname.example.org";
747 }
748
749 DummyServer(String hostname) {
750 this.hostname = hostname;
751 }
752
753 @Override
754 public Configuration getConfiguration() {
755 return conf;
756 }
757
758 @Override
759 public ZooKeeperWatcher getZooKeeper() {
760 return zkw;
761 }
762
763 @Override
764 public CoordinatedStateManager getCoordinatedStateManager() {
765 return null;
766 }
767 @Override
768 public ClusterConnection getConnection() {
769 return null;
770 }
771
772 @Override
773 public MetaTableLocator getMetaTableLocator() {
774 return null;
775 }
776
777 @Override
778 public ServerName getServerName() {
779 return ServerName.valueOf(hostname, 1234, 1L);
780 }
781
782 @Override
783 public void abort(String why, Throwable e) {
784
785 }
786
787 @Override
788 public boolean isAborted() {
789 return false;
790 }
791
792 @Override
793 public void stop(String why) {
794
795 }
796
797 @Override
798 public boolean isStopped() {
799 return false;
800 }
801
802 @Override
803 public ChoreService getChoreService() {
804 return null;
805 }
806 }
807 }