import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceManager; //导入依赖的package包/类
/**
 * @throws java.lang.Exception
 */
@BeforeClass
public static void setUpBeforeClass() throws Exception {
  TEST_UTIL.startMiniZKCluster();
  Configuration conf = TEST_UTIL.getConfiguration();
  conf.setBoolean(HConstants.REPLICATION_ENABLE_KEY, true);
  admin = new ReplicationAdmin(conf);
  Path oldLogDir = new Path(TEST_UTIL.getDataTestDir(),
      HConstants.HREGION_OLDLOGDIR_NAME);
  Path logDir = new Path(TEST_UTIL.getDataTestDir(),
      HConstants.HREGION_LOGDIR_NAME);
  manager = new ReplicationSourceManager(admin.getReplicationZk(), conf,
      // The following stopper never stops so that we can respond
      // to zk notification
      new Stoppable() {
        @Override
        public void stop(String why) {}
        @Override
        public boolean isStopped() {return false;}
      }, FileSystem.get(conf), replicating, logDir, oldLogDir);
}
 

import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceManager; //导入依赖的package包/类
@Override
public void init(Configuration conf, FileSystem fs, ReplicationSourceManager manager,
    ReplicationQueues rq, ReplicationPeers rp, Stoppable stopper, String peerClusterId,
    UUID clusterId, ReplicationEndpoint replicationEndpoint, MetricsSource metrics)
        throws IOException {

  this.manager = manager;
  this.peerClusterId = peerClusterId;
}
 

import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceManager; //导入依赖的package包/类
@Override
public void init(Configuration conf, FileSystem fs,
                 ReplicationSourceManager manager, Stoppable stopper,
                 AtomicBoolean replicating, String peerClusterId)
    throws IOException {
  this.manager = manager;
  this.peerClusterId = peerClusterId;
}
 

import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceManager; //导入依赖的package包/类
@Override
public void init(Configuration conf, FileSystem fs, ReplicationSourceManager manager,
    ReplicationQueues rq, ReplicationPeers rp, Stoppable stopper, String peerClusterId,
    UUID clusterId) throws IOException {

  this.manager = manager;
  this.peerClusterId = peerClusterId;
}
 

import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceManager; //导入依赖的package包/类
@Override
public void init(Configuration conf, FileSystem fs, ReplicationSourceManager manager,
    ReplicationQueueStorage rq, ReplicationPeer rp, Server server, String peerClusterId,
    UUID clusterId, WALFileLengthProvider walFileLengthProvider, MetricsSource metrics)
    throws IOException {
  this.manager = manager;
  this.peerClusterId = peerClusterId;
  this.metrics = metrics;
  this.walFileLengthProvider = walFileLengthProvider;
}
 

import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceManager; //导入依赖的package包/类
/**
 * Tests that {@link ReplicationSource#terminate(String)} will timeout properly
 */
@Test
public void testTerminateTimeout() throws Exception {
  ReplicationSource source = new ReplicationSource();
  ReplicationEndpoint replicationEndpoint = new HBaseInterClusterReplicationEndpoint() {
    @Override
    protected void doStart() {
      notifyStarted();
    }

    @Override
    protected void doStop() {
      // not calling notifyStopped() here causes the caller of stop() to get a Future that never
      // completes
    }
  };
  replicationEndpoint.start();
  ReplicationPeer mockPeer = Mockito.mock(ReplicationPeer.class);
  Mockito.when(mockPeer.getPeerBandwidth()).thenReturn(0L);
  Configuration testConf = HBaseConfiguration.create();
  testConf.setInt("replication.source.maxretriesmultiplier", 1);
  ReplicationSourceManager manager = Mockito.mock(ReplicationSourceManager.class);
  Mockito.when(manager.getTotalBufferUsed()).thenReturn(new AtomicLong());
  source.init(testConf, null, manager, null, mockPeer, null, "testPeer", null,
    p -> OptionalLong.empty(), null);
  ExecutorService executor = Executors.newSingleThreadExecutor();
  Future<?> future = executor.submit(new Runnable() {

    @Override
    public void run() {
      source.terminate("testing source termination");
    }
  });
  long sleepForRetries = testConf.getLong("replication.source.sleepforretries", 1000);
  Waiter.waitFor(testConf, sleepForRetries * 2, new Predicate<Exception>() {

    @Override
    public boolean evaluate() throws Exception {
      return future.isDone();
    }

  });

}
 

import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceManager; //导入依赖的package包/类
/**
 * Tests that recovered queues are preserved on a regionserver shutdown.
 * See HBASE-18192
 * @throws Exception
 */
@Test
public void testServerShutdownRecoveredQueue() throws Exception {
  try {
    // Ensure single-threaded WAL
    conf.set("hbase.wal.provider", "defaultProvider");
    conf.setInt("replication.sleep.before.failover", 2000);
    // Introduces a delay in regionserver shutdown to give the race condition a chance to kick in.
    conf.set(HConstants.REGION_SERVER_IMPL, ShutdownDelayRegionServer.class.getName());
    MiniHBaseCluster cluster = TEST_UTIL.startMiniCluster(2);
    TEST_UTIL_PEER.startMiniCluster(1);

    HRegionServer serverA = cluster.getRegionServer(0);
    final ReplicationSourceManager managerA =
        ((Replication) serverA.getReplicationSourceService()).getReplicationManager();
    HRegionServer serverB = cluster.getRegionServer(1);
    final ReplicationSourceManager managerB =
        ((Replication) serverB.getReplicationSourceService()).getReplicationManager();
    final Admin admin = TEST_UTIL.getAdmin();

    final String peerId = "TestPeer";
    admin.addReplicationPeer(peerId,
        new ReplicationPeerConfig().setClusterKey(TEST_UTIL_PEER.getClusterKey()));
    // Wait for replication sources to come up
    Waiter.waitFor(conf, 20000, new Waiter.Predicate<Exception>() {
      @Override public boolean evaluate() throws Exception {
        return !(managerA.getSources().isEmpty() || managerB.getSources().isEmpty());
      }
    });
    // Disabling peer makes sure there is at least one log to claim when the server dies
    // The recovered queue will also stay there until the peer is disabled even if the
    // WALs it contains have no data.
    admin.disableReplicationPeer(peerId);

    // Stopping serverA
    // It's queues should be claimed by the only other alive server i.e. serverB
    cluster.stopRegionServer(serverA.getServerName());
    Waiter.waitFor(conf, 20000, new Waiter.Predicate<Exception>() {
      @Override public boolean evaluate() throws Exception {
        return managerB.getOldSources().size() == 1;
      }
    });

    final HRegionServer serverC = cluster.startRegionServer().getRegionServer();
    serverC.waitForServerOnline();
    Waiter.waitFor(conf, 20000, new Waiter.Predicate<Exception>() {
      @Override public boolean evaluate() throws Exception {
        return serverC.getReplicationSourceService() != null;
      }
    });
    final ReplicationSourceManager managerC =
        ((Replication) serverC.getReplicationSourceService()).getReplicationManager();
    // Sanity check
    assertEquals(0, managerC.getOldSources().size());

    // Stopping serverB
    // Now serverC should have two recovered queues:
    // 1. The serverB's normal queue
    // 2. serverA's recovered queue on serverB
    cluster.stopRegionServer(serverB.getServerName());
    Waiter.waitFor(conf, 20000, new Waiter.Predicate<Exception>() {
      @Override public boolean evaluate() throws Exception {
        return managerC.getOldSources().size() == 2;
      }
    });
    admin.enableReplicationPeer(peerId);
    Waiter.waitFor(conf, 20000, new Waiter.Predicate<Exception>() {
      @Override public boolean evaluate() throws Exception {
        return managerC.getOldSources().size() == 0;
      }
    });
  } finally {
    conf.set(HConstants.REGION_SERVER_IMPL, HRegionServer.class.getName());
  }
}
 

import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceManager; //导入依赖的package包/类
@Override
public ReplicationSourceManager getSourceManager() {
  return manager;
}