docs: public void receive(IncomingT element) {
      assert adminExecutor.inEventLoop();
      if (stopped) {
        return;
      }
      if (window.isZero() || maxEvents == 1) {
        LOG.debug(
            "Received {}, flushing immediately (window = {}, maxEvents = {})",
            element,
            window,
            maxEvents);
        onFlush.accept(coalescer.apply(ImmutableList.of(element)));
      } else {
        currentBatch.add(element);
        if (currentBatch.size() == maxEvents) {
          LOG.debug(
              "Received {}, flushing immediately (because {} accumulated events)",
              element,
              maxEvents);
          flushNow();
        } else {
          LOG.debug("Received {}, scheduling next flush in {}", element, window);
          scheduleFlush();
        }
      }
    }