@Override
    public void run() {
      logger.info("{}: Dispatcher for {} starts", impl.getThisNode(), peer);
      try {
        Batch batch;
        while (!Thread.interrupted() && !stopped) {
          long startTime = System.nanoTime();
          while ((batch = getBatch()).isEmpty()) {
            // we may block here if there is no requests in the queue
            IndexedConsensusRequest request =
                pendingEntries.poll(PENDING_REQUEST_TAKING_TIME_OUT_IN_SEC, TimeUnit.SECONDS);
            if (request != null) {
              bufferedEntries.add(request);
              // If write pressure is low, we simply sleep a little to reduce the number of RPC
              if (pendingEntries.size() <= config.getReplication().getMaxLogEntriesNumPerBatch()
                  && bufferedEntries.isEmpty()) {
                Thread.sleep(config.getReplication().getMaxWaitingTimeForAccumulatingBatchInMs());
              }
            }
            // Immediately check for interrupts after poll and sleep
            if (Thread.interrupted() || stopped) {
              throw new InterruptedException("Interrupted after polling and sleeping");
            }
          }
          // Immediately check for interrupts after a time-consuming getBatch() operation
          if (Thread.interrupted() || stopped) {
            throw new InterruptedException("Interrupted after getting a batch");
          }
          logDispatcherThreadMetrics.recordConstructBatchTime(System.nanoTime() - startTime);
          // we may block here if the synchronization pipeline is full
          syncStatus.addNextBatch(batch);
          logEntriesFromWAL.addAndGet(batch.getLogEntriesNumFromWAL());
          logEntriesFromQueue.addAndGet(
              batch.getLogEntries().size() - batch.getLogEntriesNumFromWAL());
          // sends batch asynchronously and migrates the retry logic into the callback handler
          sendBatchAsync(batch, new DispatchLogHandler(this, logDispatcherThreadMetrics, batch));
        }
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      } catch (Exception e) {
        logger.error("Unexpected error in logDispatcher for peer {}", peer, e);
      }
      runFinished.countDown();
      logger.info("{}: Dispatcher for {} exits", impl.getThisNode(), peer);
    }
try to remove a request from queue with memory control.
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java