void doAsyncFlush() {
      DiskAccessException fatalDae = null;
      if (logger.isDebugEnabled()) {
        logger.debug("Async writer thread started");
      }
      boolean doingFlush = false;
      final ThreadsMonitoring threadMonitoring = getThreadMonitoring();
      final AbstractExecutor threadMonitorExecutor =
          threadMonitoring.createAbstractExecutor(AsyncWriterExecutor);
      threadMonitorExecutor.suspendMonitoring();
      threadMonitoring.register(threadMonitorExecutor);

      try {
        while (waitUntilFlushIsReady()) {
          threadMonitorExecutor.resumeMonitoring();
          try {
            int drainCount = diskStore.fillDrainList();
            if (drainCount > 0) {
              Iterator<Object> it = diskStore.getDrainList().iterator();
              while (it.hasNext()) {
                threadMonitorExecutor.reportProgress();
                Object o = it.next();
                if (o instanceof FlushNotifier) {
                  flushChild();
                  if (LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER) {
                    if (!it.hasNext()) {
                      doingFlush = false;
                      CacheObserverHolder.getInstance().afterWritingBytes();
                    }
                  }
                  ((FlushNotifier) o).doFlush();
                } else {
                  try {
                    AsyncDiskEntry ade = (AsyncDiskEntry) o;
                    InternalRegion region = ade.region;
                    VersionTag tag = ade.tag;
                    if (ade.versionOnly) {
                      DiskEntry.Helper.doAsyncFlush(tag, region);
                    } else {
                      DiskEntry entry = ade.de;
                      // We check isPendingAsync
                      if (entry.getDiskId().isPendingAsync()) {
                        if (LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER) {
                          if (!doingFlush) {
                            doingFlush = true;
                            CacheObserverHolder.getInstance().goingToFlush();
                          }
                        }
                        DiskEntry.Helper.doAsyncFlush(entry, region, tag);
                      } else {
                        // If it is no longer pending someone called
                        // unscheduleAsyncWrite
                        // so we don't need to write the entry, but
                        // if we have a version tag we need to record the
                        // operation
                        // to update the RVV
                        if (tag != null) {
                          DiskEntry.Helper.doAsyncFlush(tag, region);
                        }
                      }
                    }
                  } catch (RegionDestroyedException ignore) {
                    // Normally we flush before closing or destroying a region
                    // but in some cases it is closed w/o flushing.
                    // So just ignore it; see bug 41305.
                  }
                }
              }
              flushChild();
              if (doingFlush) {
                doingFlush = false;
                if (LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER) {
                  CacheObserverHolder.getInstance().afterWritingBytes();
                }
              }
              diskStore.getStats().incQueueSize(-drainCount);
            }
          } finally {
            threadMonitorExecutor.suspendMonitoring();
          }
        }
      } catch (InterruptedException ie) {
        flushChild();
        Thread.currentThread().interrupt();
        diskStore.getCache().getCancelCriterion().checkCancelInProgress(ie);
        throw new IllegalStateException("Async writer thread stopping due to unexpected interrupt");
      } catch (DiskAccessException dae) {
        boolean okToIgnore = dae.getCause() instanceof ClosedByInterruptException;
        if (!okToIgnore || !diskStore.isStopFlusher()) {
          fatalDae = dae;
        }
      } catch (CancelException ignore) {
      } catch (Throwable t) {
        logger.fatal("Fatal error from asynchronous flusher thread",
            t);
        fatalDae = new DiskAccessException(
            "Fatal error from asynchronous flusher thread", t, diskStore);
      } finally {
        if (logger.isDebugEnabled()) {
          logger.debug("Async writer thread stopped. Pending opcount={}",
              diskStore.getAsyncQueue().size());
        }
        diskStore.flusherThreadTerminated = true;
        diskStore.stopFlusher = true; // set this before calling handleDiskAccessException
        threadMonitoring.unregister(threadMonitorExecutor);

        // or it will hang
        if (fatalDae != null) {
          diskStore.handleDiskAccessException(fatalDae);
        }
      }
    }
Return true if we have enough async items to do a flush
geode-core/src/main/java/org/apache/geode/internal/cache/DiskStoreImpl.java