private RegionReplicationSink setUpSpiedRegionReplicationSink(
    final RegionReplicationSink regionReplicationSink, final HRegionForTest primaryRegion,
    final AtomicBoolean completedRef) {
    final AtomicInteger onCompleteCounter = new AtomicInteger(0);
    final AtomicInteger getStartFlushAllDescriptorCounter = new AtomicInteger(0);
    RegionReplicationSink spiedRegionReplicationSink = Mockito.spy(regionReplicationSink);

    Mockito.doAnswer((invocationOnMock) -> {
      if (!startTest) {
        invocationOnMock.callRealMethod();
        return null;
      }
      int count = onCompleteCounter.incrementAndGet();
      if (count == 1) {
        primaryRegion.cyclicBarrier.await();
        invocationOnMock.callRealMethod();
        completedRef.set(true);
        return null;
      }
      invocationOnMock.callRealMethod();
      return null;
    }).when(spiedRegionReplicationSink).onComplete(Mockito.anyList(), Mockito.anyMap());

    Mockito.doAnswer((invocationOnMock) -> {
      if (!startTest) {
        return invocationOnMock.callRealMethod();
      }
      if (
        primaryRegion.prepareFlush
          && Thread.currentThread().getName().equals(HRegionForTest.USER_THREAD_NAME)
      ) {
        int count = getStartFlushAllDescriptorCounter.incrementAndGet();
        if (count == 1) {
          // onComplete could execute
          primaryRegion.cyclicBarrier.await();
          return invocationOnMock.callRealMethod();
        }
      }
      return invocationOnMock.callRealMethod();
    }).when(spiedRegionReplicationSink).getStartFlushAllDescriptor(Mockito.any());

    primaryRegion.setRegionReplicationSink(spiedRegionReplicationSink);
    return spiedRegionReplicationSink;
  }
Write First cell,replicating to secondary replica is error.
hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/regionreplication/TestRegionReplicationSinkCallbackAndFlushConcurrently.java