protected void performMobFlush(MemStoreSnapshot snapshot, long cacheFlushId,
    InternalScanner scanner, StoreFileWriter writer, MonitoredTask status,
    ThroughputController throughputController, Consumer<Path> writerCreationTracker)
    throws IOException {
    StoreFileWriter mobFileWriter = null;
    int compactionKVMax =
      conf.getInt(HConstants.COMPACTION_KV_MAX, HConstants.COMPACTION_KV_MAX_DEFAULT);
    long mobCount = 0;
    long mobSize = 0;
    long time = snapshot.getTimeRangeTracker().getMax();
    mobFileWriter = mobStore.getStoreEngine().requireWritingToTmpDirFirst()
      ? mobStore.createWriterInTmp(new Date(time), snapshot.getCellsCount(),
        store.getColumnFamilyDescriptor().getCompressionType(), store.getRegionInfo().getStartKey(),
        false)
      : mobStore.createWriter(new Date(time), snapshot.getCellsCount(),
        store.getColumnFamilyDescriptor().getCompressionType(), store.getRegionInfo().getStartKey(),
        false, writerCreationTracker);
    // the target path is {tableName}/.mob/{cfName}/mobFiles
    // the relative path is mobFiles
    byte[] fileName = Bytes.toBytes(mobFileWriter.getPath().getName());
    ScannerContext scannerContext =
      ScannerContext.newBuilder().setBatchLimit(compactionKVMax).build();
    List<ExtendedCell> cells = new ArrayList<>();
    boolean hasMore;
    String flushName = ThroughputControlUtil.getNameForThrottling(store, "flush");
    boolean control =
      throughputController != null && !store.getRegionInfo().getTable().isSystemTable();
    if (control) {
      throughputController.start(flushName);
    }
    IOException ioe = null;
    // Clear all past MOB references
    mobRefSet.get().clear();
    try {
      do {
        hasMore = scanner.next((List) cells, scannerContext);
        if (!cells.isEmpty()) {
          for (ExtendedCell c : cells) {
            // If we know that this KV is going to be included always, then let us
            // set its memstoreTS to 0. This will help us save space when writing to
            // disk.
            if (
              c.getValueLength() <= mobCellValueSizeThreshold || MobUtils.isMobReferenceCell(c)
                || c.getTypeByte() != KeyValue.Type.Put.getCode()
            ) {
              writer.append(c);
            } else {
              // append the original keyValue in the mob file.
              mobFileWriter.append(c);
              mobSize += c.getValueLength();
              mobCount++;
              // append the tags to the KeyValue.
              // The key is same, the value is the filename of the mob file
              ExtendedCell reference =
                MobUtils.createMobRefCell(c, fileName, this.mobStore.getRefCellTags());
              writer.append(reference);
            }
            if (control) {
              throughputController.control(flushName, c.getSerializedSize());
            }
          }
          cells.clear();
        }
      } while (hasMore);
    } catch (InterruptedException e) {
      ioe =
        new InterruptedIOException("Interrupted while control throughput of flushing " + flushName);
      throw ioe;
    } catch (IOException e) {
      ioe = e;
      throw e;
    } finally {
      if (control) {
        throughputController.finish(flushName);
      }
      if (ioe != null) {
        mobFileWriter.close();
      }
    }

    if (mobCount > 0) {
      // commit the mob file from temp folder to target folder.
      // If the mob file is committed successfully but the store file is not,
      // the committed mob file will be handled by the sweep tool as an unused
      // file.
      status.setStatus("Flushing mob file " + store + ": appending metadata");
      mobFileWriter.appendMetadata(cacheFlushId, false, mobCount);
      status.setStatus("Flushing mob file " + store + ": closing flushed file");
      mobFileWriter.close();
      mobStore.commitFile(mobFileWriter.getPath(), targetPath);
      LOG.debug("Flush store file: {}, store: {}", writer.getPath(), getStoreInfo());
      mobStore.updateMobFlushCount();
      mobStore.updateMobFlushedCellsCount(mobCount);
      mobStore.updateMobFlushedCellsSize(mobSize);
      // Add mob reference to store file metadata
      mobRefSet.get().add(mobFileWriter.getPath().getName());
    } else {
      try {
        status.setStatus("Flushing mob file " + store + ": no mob cells, closing flushed file");
        mobFileWriter.close();
        // If the mob file is empty, delete it instead of committing.
        store.getFileSystem().delete(mobFileWriter.getPath(), true);
      } catch (IOException e) {
        LOG.error("Failed to delete the temp mob file", e);
      }
    }
  }
Flushes the cells in the mob store.
<ol>
In the mob store, the cells with PUT type might have or have no mob tags.
<li>If a cell does not have a mob tag, flushing the cell to different files depends on the
value length. If the length is larger than a threshold, it's flushed to a mob file and the mob
file is flushed to a store file in HBase. Otherwise, directly flush the cell to a store file in
HBase.</li>
<li>If a cell have a mob tag, its value is a mob file name, directly flush it to a store file
in HBase.</li>
</ol>
@param snapshot             Memstore snapshot.
@param cacheFlushId         Log cache flush sequence number.
@param scanner              The scanner of memstore snapshot.
@param writer               The store file writer.
@param status               Task that represents the flush operation and may be updated with
                            status.
@param throughputController A controller to avoid flush too fast.
hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreFlusher.java