private void runTestAtomicity(Admin admin) throws Exception {
    createTableIfMissing(admin, useMob);
    TestContext ctx = new TestContext(conf);

    byte rows[][] = new byte[numUniqueRows][];
    for (int i = 0; i < numUniqueRows; i++) {
      rows[i] = Bytes.toBytes("test_row_" + i);
    }

    List<AtomicityWriter> writers = Lists.newArrayList();
    for (int i = 0; i < numWriters; i++) {
      AtomicityWriter writer = new AtomicityWriter(ctx, rows, FAMILIES, sharedPool);
      writers.add(writer);
      ctx.addThread(writer);
    }
    // Add a flusher
    ctx.addThread(new RepeatingTestThread(ctx) {
      @Override
      public void doAnAction() throws Exception {
        try {
          admin.flush(TABLE_NAME);
        } catch (IOException ioe) {
          LOG.warn("Ignoring exception while flushing: " + StringUtils.stringifyException(ioe));
        }
        // Flushing has been a source of ACID violations previously (see HBASE-2856), so ideally,
        // we would flush as often as possible. On a running cluster, this isn't practical:
        // (1) we will cause a lot of load due to all the flushing and compacting
        // (2) we cannot change the flushing/compacting related Configuration options to try to
        // alleviate this
        // (3) it is an unrealistic workload, since no one would actually flush that often.
        // Therefore, let's flush every minute to have more flushes than usual, but not overload
        // the running cluster.
        if (!crazyFlush) {
          Thread.sleep(60000);
        }
      }
    });

    List<AtomicGetReader> getters = Lists.newArrayList();
    for (int i = 0; i < numGetters; i++) {
      AtomicGetReader getter =
        new AtomicGetReader(ctx, rows[i % numUniqueRows], FAMILIES, sharedPool);
      getters.add(getter);
      ctx.addThread(getter);
    }

    List<AtomicScanReader> scanners = Lists.newArrayList();
    for (int i = 0; i < numScanners; i++) {
      AtomicScanReader scanner = new AtomicScanReader(ctx, FAMILIES, sharedPool);
      scanners.add(scanner);
      ctx.addThread(scanner);
    }

    ctx.startThreads();
    ctx.waitFor(millisToRun);
    ctx.stop();

    LOG.info("Finished test. Writers:");
    for (AtomicityWriter writer : writers) {
      LOG.info("  wrote " + writer.numWritten.get());
    }
    LOG.info("Readers:");
    for (AtomicGetReader reader : getters) {
      LOG.info("  read " + reader.numRead.get());
    }
    LOG.info("Scanners:");
    for (AtomicScanReader scanner : scanners) {
      LOG.info("  scanned " + scanner.numScans.get());
      LOG.info("  verified " + scanner.numRowsScanned.get() + " rows");
    }
  }
Thread that does full scans of the table looking for any partially completed rows.
hbase-server/src/test/java/org/apache/hadoop/hbase/AcidGuaranteesTestTool.java