@Test
  public void testReplayFlushStartMarkers() throws IOException {
    // load some data to primary and flush. 1 flush and some more unflushed data
    putDataWithFlushes(primaryRegion, 100, 100, 100);
    int numRows = 200;

    // now replay the edits and the flush marker
    reader = createWALReaderForPrimary();

    LOG.info("-- Replaying edits and flush events in secondary");

    FlushDescriptor startFlushDesc = null;

    int lastReplayed = 0;
    while (true) {
      WAL.Entry entry = reader.next();
      if (entry == null) {
        break;
      }
      FlushDescriptor flushDesc = WALEdit.getFlushDescriptor(entry.getEdit().getCells().get(0));
      if (flushDesc != null) {
        // first verify that everything is replayed and visible before flush event replay
        HStore store = secondaryRegion.getStore(Bytes.toBytes("cf1"));
        long storeMemstoreSize = store.getMemStoreSize().getHeapSize();
        long regionMemstoreSize = secondaryRegion.getMemStoreDataSize();
        MemStoreSize mss = store.getFlushableSize();

        if (flushDesc.getAction() == FlushAction.START_FLUSH) {
          startFlushDesc = flushDesc;
          LOG.info("-- Replaying flush start in secondary");
          PrepareFlushResult result = secondaryRegion.replayWALFlushStartMarker(startFlushDesc);
          assertNull(result.result);
          assertEquals(result.flushOpSeqId, startFlushDesc.getFlushSequenceNumber());
          assertTrue(regionMemstoreSize > 0);
          assertTrue(mss.getHeapSize() > 0);

          // assert that the store memstore is smaller now
          long newStoreMemstoreSize = store.getMemStoreSize().getHeapSize();
          LOG.info("Memstore size reduced by:"
            + Strings.humanReadableInt(newStoreMemstoreSize - storeMemstoreSize));
          assertTrue(storeMemstoreSize > newStoreMemstoreSize);
          verifyData(secondaryRegion, 0, lastReplayed + 1, cq, families);

        }
        // after replay verify that everything is still visible
        verifyData(secondaryRegion, 0, lastReplayed + 1, cq, families);
      } else {
        lastReplayed = replayEdit(secondaryRegion, entry);
      }
    }

    // at this point, there should be some data (rows 0-100) in memstore snapshot
    // and some more data in memstores (rows 100-200)

    verifyData(secondaryRegion, 0, numRows, cq, families);

    // Test case 1: replay the same flush start marker again
    LOG.info("-- Replaying same flush start in secondary again");
    PrepareFlushResult result = secondaryRegion.replayWALFlushStartMarker(startFlushDesc);
    assertNull(result); // this should return null. Ignoring the flush start marker
    // assert that we still have prepared flush with the previous setup.
    assertNotNull(secondaryRegion.getPrepareFlushResult());
    assertEquals(secondaryRegion.getPrepareFlushResult().flushOpSeqId,
      startFlushDesc.getFlushSequenceNumber());
    assertTrue(secondaryRegion.getMemStoreDataSize() > 0); // memstore is not empty
    verifyData(secondaryRegion, 0, numRows, cq, families);

    // Test case 2: replay a flush start marker with a smaller seqId
    FlushDescriptor startFlushDescSmallerSeqId =
      clone(startFlushDesc, startFlushDesc.getFlushSequenceNumber() - 50);
    LOG.info("-- Replaying same flush start in secondary again " + startFlushDescSmallerSeqId);
    result = secondaryRegion.replayWALFlushStartMarker(startFlushDescSmallerSeqId);
    assertNull(result); // this should return null. Ignoring the flush start marker
    // assert that we still have prepared flush with the previous setup.
    assertNotNull(secondaryRegion.getPrepareFlushResult());
    assertEquals(secondaryRegion.getPrepareFlushResult().flushOpSeqId,
      startFlushDesc.getFlushSequenceNumber());
    assertTrue(secondaryRegion.getMemStoreDataSize() > 0); // memstore is not empty
    verifyData(secondaryRegion, 0, numRows, cq, families);

    // Test case 3: replay a flush start marker with a larger seqId
    FlushDescriptor startFlushDescLargerSeqId =
      clone(startFlushDesc, startFlushDesc.getFlushSequenceNumber() + 50);
    LOG.info("-- Replaying same flush start in secondary again " + startFlushDescLargerSeqId);
    result = secondaryRegion.replayWALFlushStartMarker(startFlushDescLargerSeqId);
    assertNull(result); // this should return null. Ignoring the flush start marker
    // assert that we still have prepared flush with the previous setup.
    assertNotNull(secondaryRegion.getPrepareFlushResult());
    assertEquals(secondaryRegion.getPrepareFlushResult().flushOpSeqId,
      startFlushDesc.getFlushSequenceNumber());
    assertTrue(secondaryRegion.getMemStoreDataSize() > 0); // memstore is not empty
    verifyData(secondaryRegion, 0, numRows, cq, families);

    LOG.info("-- Verifying edits from secondary");
    verifyData(secondaryRegion, 0, numRows, cq, families);

    LOG.info("-- Verifying edits from primary.");
    verifyData(primaryRegion, 0, numRows, cq, families);
  }
Tests cases where we prepare a flush with some seqId and we receive other flush start markers
equal to, greater or less than the previous flush start marker.
hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionReplayEvents.java