import org.apache.hadoop.hbase.replication.ReplicationZookeeper; //导入依赖的package包/类
public void initialize(final Server server, final FileSystem fs,
    final Path logDir, final Path oldLogDir) throws IOException {
  this.server = server;
  this.conf = this.server.getConfiguration();
  this.replication = isReplication(this.conf);
  if (replication) {
    try {
      this.zkHelper = new ReplicationZookeeper(server, this.replicating);
    } catch (KeeperException ke) {
      throw new IOException("Failed replication handler create " +
         "(replicating=" + this.replicating, ke);
    }
    this.replicationManager = new ReplicationSourceManager(zkHelper, conf,
        this.server, fs, this.replicating, logDir, oldLogDir) ;
  } else {
    this.replicationManager = null;
    this.zkHelper = null;
  }
}
 

import org.apache.hadoop.hbase.replication.ReplicationZookeeper; //导入依赖的package包/类
/**
 * Instantiate the replication management (if rep is enabled).
 * @param server Hosting server
 * @param fs handle to the filesystem
 * @param logDir
 * @param oldLogDir directory where logs are archived
 * @throws IOException
 * @throws KeeperException 
 */
public Replication(final Server server, final FileSystem fs,
    final Path logDir, final Path oldLogDir)
throws IOException, KeeperException {
  this.server = server;
  this.conf = this.server.getConfiguration();
  this.replication = isReplication(this.conf);
  if (replication) {
    this.zkHelper = new ReplicationZookeeper(server, this.replicating);
    this.replicationManager = new ReplicationSourceManager(zkHelper, conf,
        this.server, fs, this.replicating, logDir, oldLogDir) ;
  } else {
    this.replicationManager = null;
    this.zkHelper = null;
  }
}
 

import org.apache.hadoop.hbase.replication.ReplicationZookeeper; //导入依赖的package包/类
/**
 * Creates a replication manager and sets the watch on all the other
 * registered region servers
 * @param zkHelper the zk helper for replication
 * @param conf the configuration to use
 * @param stopper the stopper object for this region server
 * @param fs the file system to use
 * @param replicating the status of the replication on this cluster
 * @param logDir the directory that contains all hlog directories of live RSs
 * @param oldLogDir the directory where old logs are archived
 */
public ReplicationSourceManager(final ReplicationZookeeper zkHelper,
                                final Configuration conf,
                                final Stoppable stopper,
                                final FileSystem fs,
                                final AtomicBoolean replicating,
                                final Path logDir,
                                final Path oldLogDir) {
  this.sources = new ArrayList<ReplicationSourceInterface>();
  this.replicating = replicating;
  this.zkHelper = zkHelper;
  this.stopper = stopper;
  this.hlogsById = new HashMap<String, SortedSet<String>>();
  this.oldsources = new ArrayList<ReplicationSourceInterface>();
  this.conf = conf;
  this.fs = fs;
  this.logDir = logDir;
  this.oldLogDir = oldLogDir;
  this.sleepBeforeFailover = conf.getLong("replication.sleep.before.failover", 2000);
  this.zkHelper.registerRegionServerListener(
      new OtherRegionServerWatcher(this.zkHelper.getZookeeperWatcher()));
  this.zkHelper.registerRegionServerListener(
      new PeersWatcher(this.zkHelper.getZookeeperWatcher()));
  this.zkHelper.listPeersIdsAndWatch();
  // It's preferable to failover 1 RS at a time, but with good zk servers
  // more could be processed at the same time.
  int nbWorkers = conf.getInt("replication.executor.workers", 1);
  // use a short 100ms sleep since this could be done inline with a RS startup
  // even if we fail, other region servers can take care of it
  this.executor = new ThreadPoolExecutor(nbWorkers, nbWorkers,
      100, TimeUnit.MILLISECONDS,
      new LinkedBlockingQueue<Runnable>());
  ThreadFactoryBuilder tfb = new ThreadFactoryBuilder();
  tfb.setNameFormat("ReplicationExecutor-%d");
  this.executor.setThreadFactory(tfb.build());
  this.rand = new Random();
}
 

import org.apache.hadoop.hbase.replication.ReplicationZookeeper; //导入依赖的package包/类
/**
 * Called when a node has been deleted
 * @param path full path of the deleted node
 */
public void nodeDeleted(String path) {
  if (stopper.isStopped()) {
    return;
  }
  boolean cont = refreshListIfRightPath(path);
  if (!cont) {
    return;
  }
  LOG.info(path + " znode expired, trying to lock it");
  transferQueues(ReplicationZookeeper.getZNodeName(path));
}
 

import org.apache.hadoop.hbase.replication.ReplicationZookeeper; //导入依赖的package包/类
/**
 * Constructor that creates a connection to the local ZooKeeper ensemble.
 * @param conf Configuration to use
 * @throws IOException if the connection to ZK cannot be made
 * @throws RuntimeException if replication isn't enabled.
 */
public ReplicationAdmin(Configuration conf) throws IOException {
  if (!conf.getBoolean(HConstants.REPLICATION_ENABLE_KEY, false)) {
    throw new RuntimeException("hbase.replication isn't true, please " +
        "enable it in order to use replication");
  }
  this.connection = HConnectionManager.getConnection(conf);
  ZooKeeperWatcher zkw = this.connection.getZooKeeperWatcher();
  try {
    this.replicationZk = new ReplicationZookeeper(this.connection, conf, zkw);
  } catch (KeeperException e) {
    throw new IOException("Unable setup the ZooKeeper connection", e);
  }
}
 

import org.apache.hadoop.hbase.replication.ReplicationZookeeper; //导入依赖的package包/类
@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");
  AtomicBoolean replicating = new AtomicBoolean(true);
  ReplicationZookeeper rz = new ReplicationZookeeper(server, replicating);
  // populate some znodes in the peer znode
  files.add("log1");
  files.add("log2");
  for (String file : files) {
    rz.addLogToList(file, "1");
  }
  // create 3 DummyServers
  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");

  // simulate three server fail sequentially
  ReplicationZookeeper rz1 = new ReplicationZookeeper(s1, new AtomicBoolean(true));
  SortedMap<String, SortedSet<String>> testMap =
      rz1.copyQueuesFromRSUsingMulti(server.getServerName().getServerName());
  ReplicationZookeeper rz2 = new ReplicationZookeeper(s2, new AtomicBoolean(true));
  testMap = rz2.copyQueuesFromRSUsingMulti(s1.getServerName().getServerName());
  ReplicationZookeeper rz3 = new ReplicationZookeeper(s3, new AtomicBoolean(true));
  testMap = rz3.copyQueuesFromRSUsingMulti(s2.getServerName().getServerName());

  ReplicationSource s = new ReplicationSource();
  s.checkIfQueueRecovered(testMap.firstKey());
  List<String> result = s.getDeadRegionServers();

  // verify
  assertTrue(result.contains(server.getServerName().getServerName()));
  assertTrue(result.contains(s1.getServerName().getServerName()));
  assertTrue(result.contains(s2.getServerName().getServerName()));

  server.abort("", null);
}
 

import org.apache.hadoop.hbase.replication.ReplicationZookeeper; //导入依赖的package包/类
/**
 * Creates a replication manager and sets the watch on all the other
 * registered region servers
 * @param zkHelper the zk helper for replication
 * @param conf the configuration to use
 * @param stopper the stopper object for this region server
 * @param fs the file system to use
 * @param replicating the status of the replication on this cluster
 * @param logDir the directory that contains all hlog directories of live RSs
 * @param oldLogDir the directory where old logs are archived
 */
public ReplicationSourceManager(final ReplicationZookeeper zkHelper,
                                final Configuration conf,
                                final Stoppable stopper,
                                final FileSystem fs,
                                final AtomicBoolean replicating,
                                final Path logDir,
                                final Path oldLogDir) {
  this.sources = new ArrayList<ReplicationSourceInterface>();
  this.replicating = replicating;
  this.zkHelper = zkHelper;
  this.stopper = stopper;
  this.hlogsById = new HashMap<String, SortedSet<String>>();
  this.oldsources = new ArrayList<ReplicationSourceInterface>();
  this.conf = conf;
  this.fs = fs;
  this.logDir = logDir;
  this.oldLogDir = oldLogDir;
  this.sleepBeforeFailover = conf.getLong("replication.sleep.before.failover", 2000);
  this.zkHelper.registerRegionServerListener(
      new OtherRegionServerWatcher(this.zkHelper.getZookeeperWatcher()));
  List<String> otherRSs =
      this.zkHelper.getRegisteredRegionServers();
  this.zkHelper.registerRegionServerListener(
      new PeersWatcher(this.zkHelper.getZookeeperWatcher()));
  this.zkHelper.listPeersIdsAndWatch();
  this.otherRegionServers = otherRSs == null ? new ArrayList<String>() : otherRSs;
  // It's preferable to failover 1 RS at a time, but with good zk servers
  // more could be processed at the same time.
  int nbWorkers = conf.getInt("replication.executor.workers", 1);
  // use a short 100ms sleep since this could be done inline with a RS startup
  // even if we fail, other region servers can take care of it
  this.executor = new ThreadPoolExecutor(nbWorkers, nbWorkers,
      100, TimeUnit.MILLISECONDS,
      new LinkedBlockingQueue<Runnable>());
  ThreadFactoryBuilder tfb = new ThreadFactoryBuilder();
  tfb.setNameFormat("ReplicationExecutor-%d");
  this.executor.setThreadFactory(tfb.build());
}
 

import org.apache.hadoop.hbase.replication.ReplicationZookeeper; //导入依赖的package包/类
/**
 * Called when a node has been deleted
 * @param path full path of the deleted node
 */
public void nodeDeleted(String path) {
  if (stopper.isStopped()) {
    return;
  }
  boolean cont = refreshRegionServersList(path);
  if (!cont) {
    return;
  }
  LOG.info(path + " znode expired, trying to lock it");
  transferQueues(ReplicationZookeeper.getZNodeName(path));
}
 

import org.apache.hadoop.hbase.replication.ReplicationZookeeper; //导入依赖的package包/类
/**
 * Creates a replication manager and sets the watch on all the other
 * registered region servers
 * @param zkHelper the zk helper for replication
 * @param conf the configuration to use
 * @param stopper the stopper object for this region server
 * @param fs the file system to use
 * @param replicating the status of the replication on this cluster
 * @param logDir the directory that contains all hlog directories of live RSs
 * @param oldLogDir the directory where old logs are archived
 */
public ReplicationSourceManager(final ReplicationZookeeper zkHelper,
                                final Configuration conf,
                                final Stoppable stopper,
                                final FileSystem fs,
                                final AtomicBoolean replicating,
                                final Path logDir,
                                final Path oldLogDir) {
  this.sources = new ArrayList<ReplicationSourceInterface>();
  this.replicating = replicating;
  this.zkHelper = zkHelper;
  this.stopper = stopper;
  this.hlogsById = new HashMap<String, SortedSet<String>>();
  this.oldsources = new ArrayList<ReplicationSourceInterface>();
  this.conf = conf;
  this.fs = fs;
  this.logDir = logDir;
  this.oldLogDir = oldLogDir;
  this.sleepBeforeFailover = conf.getLong("replication.sleep.before.failover", 2000);
  this.zkHelper.registerRegionServerListener(
      new OtherRegionServerWatcher(this.zkHelper.getZookeeperWatcher()));
  this.zkHelper.registerRegionServerListener(
      new PeersWatcher(this.zkHelper.getZookeeperWatcher()));
  this.zkHelper.listPeersIdsAndWatch();
  // It's preferable to failover 1 RS at a time, but with good zk servers
  // more could be processed at the same time.
  int nbWorkers = conf.getInt("replication.executor.workers", 1);
  // use a short 100ms sleep since this could be done inline with a RS startup
  // even if we fail, other region servers can take care of it
  this.executor = new ThreadPoolExecutor(nbWorkers, nbWorkers,
      100, TimeUnit.MILLISECONDS,
      new LinkedBlockingQueue<Runnable>());
  ThreadFactoryBuilder tfb = new ThreadFactoryBuilder();
  tfb.setNameFormat("ReplicationExecutor-%d");
  this.executor.setThreadFactory(tfb.build());
}
 

import org.apache.hadoop.hbase.replication.ReplicationZookeeper; //导入依赖的package包/类
public void initialize(final Server server, final FileSystem fs,
    final Path logDir, final Path oldLogDir) throws IOException {
  this.server = server;
  this.conf = this.server.getConfiguration();
  this.replication = isReplication(this.conf);
  this.scheduleThreadPool = Executors.newScheduledThreadPool(1,
    new ThreadFactoryBuilder()
      .setNameFormat(server.getServerName() + "Replication Statistics #%d")
      .setDaemon(true)
      .build());
  if (replication) {
    try {
      this.zkHelper = new ReplicationZookeeper(server, this.replicating);
    } catch (KeeperException ke) {
      throw new IOException("Failed replication handler create " +
         "(replicating=" + this.replicating, ke);
    }
    this.replicationManager = new ReplicationSourceManager(zkHelper, conf, this.server, fs,
        this.replicating, logDir, oldLogDir);
    this.statsThreadPeriod =
        this.conf.getInt("replication.stats.thread.period.seconds", 5 * 60);
    LOG.debug("ReplicationStatisticsThread " + this.statsThreadPeriod);
  } else {
    this.replicationManager = null;
    this.zkHelper = null;
  }
}