@Test
  public void testTerminateClearsBuffer() throws Exception {
    ReplicationSource source = new ReplicationSource();
    ReplicationSourceManager mockManager = new ReplicationSourceManager(null, null, conf, null,
      null, null, null, null, null, null, mock(MetricsReplicationGlobalSourceSource.class));
    ReplicationPeer mockPeer = mock(ReplicationPeer.class);
    Mockito.when(mockPeer.getPeerBandwidth()).thenReturn(0L);
    Configuration testConf = HBaseConfiguration.create();
    ReplicationQueueId queueId =
      new ReplicationQueueId(ServerName.valueOf("test,123,123"), "testPeer");
    source.init(testConf, null, mockManager, null, mockPeer, Mockito.mock(Server.class),
      new ReplicationQueueData(queueId, ImmutableMap.of()), null, p -> OptionalLong.empty(),
      mock(MetricsSource.class));
    ReplicationSourceWALReader reader =
      new ReplicationSourceWALReader(null, conf, null, 0, null, source, null);
    ReplicationSourceShipper shipper = new ReplicationSourceShipper(conf, null, source, reader);
    source.workerThreads.put("testPeer", shipper);
    WALEntryBatch batch = new WALEntryBatch(10, logDir);
    WAL.Entry mockEntry = mock(WAL.Entry.class);
    WALEdit mockEdit = mock(WALEdit.class);
    WALKeyImpl mockKey = mock(WALKeyImpl.class);
    when(mockEntry.getEdit()).thenReturn(mockEdit);
    when(mockEdit.isEmpty()).thenReturn(false);
    when(mockEntry.getKey()).thenReturn(mockKey);
    when(mockKey.estimatedSerializedSizeOf()).thenReturn(1000L);
    when(mockEdit.heapSize()).thenReturn(10000L);
    when(mockEdit.size()).thenReturn(0);
    ArrayList<Cell> cells = new ArrayList<>();
    KeyValue kv = new KeyValue(Bytes.toBytes("0001"), Bytes.toBytes("f"), Bytes.toBytes("1"),
      Bytes.toBytes("v1"));
    cells.add(kv);
    when(mockEdit.getCells()).thenReturn(cells);
    reader.addEntryToBatch(batch, mockEntry);
    reader.entryBatchQueue.put(batch);
    source.terminate("test");
    assertEquals(0, source.getSourceManager().getTotalBufferUsed());
  }
Tests that {@link ReplicationSource#terminate(String)} will timeout properly Moved here from
TestReplicationSource because doesn't need cluster.
hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSource.java