@Test
  public void testRunDoubleMemStoreCompactors() throws IOException, InterruptedException {
    int flushSize = 500;
    Configuration conf = HBaseConfiguration.create();
    conf.set(HStore.MEMSTORE_CLASS_NAME, MyCompactingMemStoreWithCustomCompactor.class.getName());
    conf.setDouble(CompactingMemStore.IN_MEMORY_FLUSH_THRESHOLD_FACTOR_KEY, 0.25);
    MyCompactingMemStoreWithCustomCompactor.RUNNER_COUNT.set(0);
    conf.set(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, String.valueOf(flushSize));
    // Set the lower threshold to invoke the "MERGE" policy
    conf.set(MemStoreCompactionStrategy.COMPACTING_MEMSTORE_THRESHOLD_KEY, String.valueOf(0));
    init(name.getMethodName(), conf, ColumnFamilyDescriptorBuilder.newBuilder(family)
      .setInMemoryCompaction(MemoryCompactionPolicy.BASIC).build());
    byte[] value = Bytes.toBytes("thisisavarylargevalue");
    MemStoreSizing memStoreSizing = new NonThreadSafeMemStoreSizing();
    long ts = EnvironmentEdgeManager.currentTime();
    long seqId = 100;
    // older data whihc shouldn't be "seen" by client
    store.add(createCell(qf1, ts, seqId, value), memStoreSizing);
    store.add(createCell(qf2, ts, seqId, value), memStoreSizing);
    store.add(createCell(qf3, ts, seqId, value), memStoreSizing);
    assertEquals(1, MyCompactingMemStoreWithCustomCompactor.RUNNER_COUNT.get());
    StoreFlushContext storeFlushCtx = store.createFlushContext(id++, FlushLifeCycleTracker.DUMMY);
    storeFlushCtx.prepare();
    // This shouldn't invoke another in-memory flush because the first compactor thread
    // hasn't accomplished the in-memory compaction.
    store.add(createCell(qf1, ts + 1, seqId + 1, value), memStoreSizing);
    store.add(createCell(qf1, ts + 1, seqId + 1, value), memStoreSizing);
    store.add(createCell(qf1, ts + 1, seqId + 1, value), memStoreSizing);
    assertEquals(1, MyCompactingMemStoreWithCustomCompactor.RUNNER_COUNT.get());
    // okay. Let the compaction be completed
    MyMemStoreCompactor.START_COMPACTOR_LATCH.countDown();
    CompactingMemStore mem = (CompactingMemStore) ((HStore) store).memstore;
    while (mem.isMemStoreFlushingInMemory()) {
      TimeUnit.SECONDS.sleep(1);
    }
    // This should invoke another in-memory flush.
    store.add(createCell(qf1, ts + 2, seqId + 2, value), memStoreSizing);
    store.add(createCell(qf1, ts + 2, seqId + 2, value), memStoreSizing);
    store.add(createCell(qf1, ts + 2, seqId + 2, value), memStoreSizing);
    assertEquals(2, MyCompactingMemStoreWithCustomCompactor.RUNNER_COUNT.get());
    conf.set(HConstants.HREGION_MEMSTORE_FLUSH_SIZE,
      String.valueOf(TableDescriptorBuilder.DEFAULT_MEMSTORE_FLUSH_SIZE));
    storeFlushCtx.flushCache(Mockito.mock(MonitoredTask.class));
    storeFlushCtx.commit(Mockito.mock(MonitoredTask.class));
  }
If there are two running InMemoryFlushRunnable, the later InMemoryFlushRunnable may change the
versionedList. And the first InMemoryFlushRunnable will use the chagned versionedList to remove
the corresponding segments. In short, there will be some segements which isn't in merge are
removed.
hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHStore.java