@Test
  public void testAgeOfOldestWal() throws Exception {
    try {
      ManualEnvironmentEdge manualEdge = new ManualEnvironmentEdge();
      EnvironmentEdgeManager.injectEdge(manualEdge);

      String peerId = "1";
      MetricsSource metrics = new MetricsSource(peerId);
      Configuration conf = new Configuration(TEST_UTIL.getConfiguration());
      conf.setInt("replication.source.maxretriesmultiplier", 1);
      ReplicationPeer mockPeer = Mockito.mock(ReplicationPeer.class);
      Mockito.when(mockPeer.getConfiguration()).thenReturn(conf);
      Mockito.when(mockPeer.getPeerBandwidth()).thenReturn(0L);
      ReplicationPeerConfig peerConfig = Mockito.mock(ReplicationPeerConfig.class);
      Mockito.when(peerConfig.getReplicationEndpointImpl())
        .thenReturn(DoNothingReplicationEndpoint.class.getName());
      Mockito.when(mockPeer.getPeerConfig()).thenReturn(peerConfig);
      ReplicationSourceManager manager = Mockito.mock(ReplicationSourceManager.class);
      Mockito.when(manager.getGlobalMetrics())
        .thenReturn(mock(MetricsReplicationGlobalSourceSource.class));
      RegionServerServices rss =
        TEST_UTIL.createMockRegionServerService(ServerName.parseServerName("a.b.c,1,1"));
      ReplicationQueueId queueId = new ReplicationQueueId(rss.getServerName(), peerId);
      ReplicationSource source = new ReplicationSource();
      source.init(conf, null, manager, null, mockPeer, rss,
        new ReplicationQueueData(queueId, ImmutableMap.of()), null, p -> OptionalLong.empty(),
        metrics);

      final Path log1 = new Path(logDir, "log-walgroup-a.8");
      manualEdge.setValue(10);
      // Diff of current time (10) and log-walgroup-a.8 timestamp will be 2.
      source.enqueueLog(log1);
      MetricsReplicationSourceSource metricsSource1 = getSourceMetrics(peerId);
      assertEquals(2, metricsSource1.getOldestWalAge());

      final Path log2 = new Path(logDir, "log-walgroup-b.4");
      // Diff of current time (10) and log-walgroup-b.4 will be 6 so oldestWalAge should be 6
      source.enqueueLog(log2);
      assertEquals(6, metricsSource1.getOldestWalAge());
      // Clear all metrics.
      metrics.clear();
    } finally {
      EnvironmentEdgeManager.reset();
    }
  }
Test ReplicationSource retries startup once an uncaught exception happens during initialization
and <b>replication.source.regionserver.abort</b> is set to true.
hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSource.java