@Override
  public synchronized void run() {
    try {
      if (betterPostponeFlushNow()) {
        log.debug("Flush short-circuit -> schedule next turn earlier");
        scheduledFuture = scheduler.schedule(this, quickReCheckingPeriodMs, MILLISECONDS);
        return;
      }

      final long startedAtNs = System.nanoTime();

      final IntRef contentionQuota = new IntRef(contentionQuotaPerTurn);
      final FlushResult flushResult = flushAsMuchAsPossibleWithinQuota(contentionQuota);

      final long finishedAtNs = System.nanoTime();

      //control loop: adjust period and contention quota for next turn
      // based on the quota spent and success (or lack of) reached
      final int unspentQuota = contentionQuota.get();
      final int previousQuotaPerTurn = contentionQuotaPerTurn;
      totalQuotaSpent.addAndGet(previousQuotaPerTurn - unspentQuota);

      switch (flushResult) {
        case FLUSHED_ALL -> {
          totalFlushingTimeUs.addAndGet(NANOSECONDS.toMicros(finishedAtNs - startedAtNs));
          totalFlushes.incrementAndGet();

          if (0 < unspentQuota && unspentQuota < previousQuotaPerTurn) {
            // if (unspentQuota == previousQuotaPerTurn) => most likely we just did nothing, so no
            // reason to change per turn quota
            contentionQuotaPerTurn = clamp(contentionQuotaPerTurn - 1, minContentionQuota, maxContentionQuota);
          }

          scheduledFuture = scheduler.schedule(this, regularCheckingPeriodMs, MILLISECONDS);
          if (log.isDebugEnabled()) {
            log.debug("Flushed everything: contention quota(" + previousQuotaPerTurn + " -> " + unspentQuota + ") -> " +
                      "next turn scheduled regularly, with quota: " + contentionQuotaPerTurn);
          }
        }
        case HAS_MORE_TO_FLUSH -> {
          totalFlushingTimeUs.addAndGet(NANOSECONDS.toMicros(finishedAtNs - startedAtNs));
          totalFlushes.incrementAndGet();
          totalFlushesRetried.incrementAndGet();

          if (unspentQuota < 0) {
            contentionQuotaPerTurn = clamp(contentionQuotaPerTurn * 2, minContentionQuota, maxContentionQuota);
          }

          scheduledFuture = scheduler.schedule(this, quickReCheckingPeriodMs, MILLISECONDS);
          if (log.isDebugEnabled()) {
            log.debug(
              "Flush something, but more remains: contention quota(" + previousQuotaPerTurn + " -> " + unspentQuota + ") -> " +
              "next turn scheduled early, with quota: " + contentionQuotaPerTurn);
          }
        }
        case NOTHING_TO_FLUSH_NOW -> {
          scheduledFuture = scheduler.schedule(this, regularCheckingPeriodMs, MILLISECONDS);
          if (log.isDebugEnabled()) {
            log.debug(
              "Nothing to flush now: contention quota(" + previousQuotaPerTurn + " -> " + unspentQuota + ") -> " +
              "next turn scheduled regularly, with quota: " + contentionQuotaPerTurn + " unchanged");
          }
        }
      }
    }
    catch (InterruptedException e) {
      log.error("Flushing thread interrupted -> exiting", e);
    }
    catch (AlreadyDisposedException | RejectedExecutionException e){
      log.warn("Stop flushing: pool is shutting down or whole application is closing", e);
    }
    catch (Throwable t) {
      log.warn("Unhandled exception during flush (reschedule regularly)", t);
      scheduledFuture = scheduler.schedule(this, regularCheckingPeriodMs, MILLISECONDS);
    }
  }
@Override
  public synchronized void run()