@Test public void testNodeFailoverDeadServerParsing() throws Exception { LOG.debug("testNodeFailoverDeadServerParsing"); conf.setBoolean(HConstants.ZOOKEEPER_USEMULTI, true); final Server server = new DummyServer("ec2-54-234-230-108.compute-1.amazonaws.com"); ReplicationQueues repQueues = ReplicationFactory.getReplicationQueues(server.getZooKeeper(), conf, server); repQueues.init(server.getServerName().toString()); files.add("log1"); files.add("log2"); for (String file : files) { repQueues.addLog("1", file); } Server s1 = new DummyServer("ip-10-8-101-114.ec2.internal"); Server s2 = new DummyServer("ec2-107-20-52-47.compute-1.amazonaws.com"); Server s3 = new DummyServer("ec2-23-20-187-167.compute-1.amazonaws.com"); ReplicationQueues rq1 = ReplicationFactory.getReplicationQueues(s1.getZooKeeper(), s1.getConfiguration(), s1); rq1.init(s1.getServerName().toString()); SortedMap<String, SortedSet<String>> testMap = rq1.claimQueues(server.getServerName().getServerName()); ReplicationQueues rq2 = ReplicationFactory.getReplicationQueues(s2.getZooKeeper(), s2.getConfiguration(), s2); rq2.init(s2.getServerName().toString()); testMap = rq2.claimQueues(s1.getServerName().getServerName()); ReplicationQueues rq3 = ReplicationFactory.getReplicationQueues(s3.getZooKeeper(), s3.getConfiguration(), s3); rq3.init(s3.getServerName().toString()); testMap = rq3.claimQueues(s2.getServerName().getServerName()); ReplicationQueueInfo replicationQueueInfo = new ReplicationQueueInfo(testMap.firstKey()); List<String> result = replicationQueueInfo.getDeadRegionServers(); assertTrue(result.contains(server.getServerName().getServerName())); assertTrue(result.contains(s1.getServerName().getServerName())); assertTrue(result.contains(s2.getServerName().getServerName())); server.abort("", null); }
protected void init() throws IOException, ReplicationException { for (String id : this.replicationPeers.getConnectedPeers()) { addSource(id); } List<String> currentReplicators = this.replicationQueues.getListOfReplicators(); if (currentReplicators == null || currentReplicators.size() == 0) { return; } List<String> otherRegionServers = replicationTracker.getListOfRegionServers(); LOG.info("Current list of replicators: " + currentReplicators + " other RSs: " + otherRegionServers); for (String rs : currentReplicators) { if (!otherRegionServers.contains(rs)) { transferQueues(rs); } } }
ReplicationSourceManager implements ReplicationListener { protected void init() throws IOException, ReplicationException { for (String id : this.replicationPeers.getConnectedPeers()) { addSource(id); } List<String> currentReplicators = this.replicationQueues.getListOfReplicators(); if (currentReplicators == null || currentReplicators.size() == 0) { return; } List<String> otherRegionServers = replicationTracker.getListOfRegionServers(); LOG.info("Current list of replicators: " + currentReplicators + " other RSs: " + otherRegionServers); for (String rs : currentReplicators) { if (!otherRegionServers.contains(rs)) { transferQueues(rs); } } } }
ReplicationSourceManager implements ReplicationListener { protected void init() throws IOException, ReplicationException { for (String id : this.replicationPeers.getConnectedPeers()) { addSource(id); } List<String> currentReplicators = this.replicationQueues.getListOfReplicators(); if (currentReplicators == null || currentReplicators.size() == 0) { return; } List<String> otherRegionServers = replicationTracker.getListOfRegionServers(); LOG.info("Current list of replicators: " + currentReplicators + " other RSs: " + otherRegionServers); for (String rs : currentReplicators) { if (!otherRegionServers.contains(rs)) { transferQueues(rs); } } } ReplicationSourceManager(final ReplicationQueues replicationQueues,
      final ReplicationPeers replicationPeers, final ReplicationTracker replicationTracker,
      final Configuration conf, final Stoppable stopper, final FileSystem fs, final Path logDir,
      final Path oldLogDir, final UUID clusterId); }
ReplicationSourceManager implements ReplicationListener { protected void init() throws IOException, ReplicationException { for (String id : this.replicationPeers.getConnectedPeers()) { addSource(id); } List<String> currentReplicators = this.replicationQueues.getListOfReplicators(); if (currentReplicators == null || currentReplicators.size() == 0) { return; } List<String> otherRegionServers = replicationTracker.getListOfRegionServers(); LOG.info("Current list of replicators: " + currentReplicators + " other RSs: " + otherRegionServers); for (String rs : currentReplicators) { if (!otherRegionServers.contains(rs)) { transferQueues(rs); } } } ReplicationSourceManager(final ReplicationQueues replicationQueues,
      final ReplicationPeers replicationPeers, final ReplicationTracker replicationTracker,
      final Configuration conf, final Stoppable stopper, final FileSystem fs, final Path logDir,
      final Path oldLogDir, final UUID clusterId); void logPositionAndCleanOldLogs(Path log, String id, long position,
      boolean queueRecovered, boolean holdLogInZK); void cleanOldLogs(String key,
                           String id,
                           boolean queueRecovered); void deleteSource(String peerId, boolean closeConnection); void join(); List<ReplicationSourceInterface> getSources(); List<ReplicationSourceInterface> getOldSources(); void closeRecoveredQueue(ReplicationSourceInterface src); void removePeer(String id); @Override void regionServerRemoved(String regionserver); @Override void peerRemoved(String peerId); @Override void peerListChanged(List<String> peerIds); Path getOldLogDir(); Path getLogDir(); FileSystem getFs(); String getStats(); }
ReplicationSourceManager implements ReplicationListener { protected void init() throws IOException, ReplicationException { for (String id : this.replicationPeers.getConnectedPeers()) { addSource(id); } List<String> currentReplicators = this.replicationQueues.getListOfReplicators(); if (currentReplicators == null || currentReplicators.size() == 0) { return; } List<String> otherRegionServers = replicationTracker.getListOfRegionServers(); LOG.info("Current list of replicators: " + currentReplicators + " other RSs: " + otherRegionServers); for (String rs : currentReplicators) { if (!otherRegionServers.contains(rs)) { transferQueues(rs); } } } ReplicationSourceManager(final ReplicationQueues replicationQueues,
      final ReplicationPeers replicationPeers, final ReplicationTracker replicationTracker,
      final Configuration conf, final Stoppable stopper, final FileSystem fs, final Path logDir,
      final Path oldLogDir, final UUID clusterId); void logPositionAndCleanOldLogs(Path log, String id, long position,
      boolean queueRecovered, boolean holdLogInZK); void cleanOldLogs(String key,
                           String id,
                           boolean queueRecovered); void deleteSource(String peerId, boolean closeConnection); void join(); List<ReplicationSourceInterface> getSources(); List<ReplicationSourceInterface> getOldSources(); void closeRecoveredQueue(ReplicationSourceInterface src); void removePeer(String id); @Override void regionServerRemoved(String regionserver); @Override void peerRemoved(String peerId); @Override void peerListChanged(List<String> peerIds); Path getOldLogDir(); Path getLogDir(); FileSystem getFs(); String getStats();  }