public void offer(IndexedConsensusRequest request) {
    // we don't need to serialize and offer request when replicaNum is 1.
    if (!threads.isEmpty()) {
      request.buildSerializedRequests();
      synchronized (this) {
        threads.forEach(
            thread -> {
              logger.debug(
                  "{}->{}: Push a log to the queue, where the queue length is {}",
                  impl.getThisNode().getGroupId(),
                  thread.getPeer().getEndpoint().getIp(),
                  thread.getPendingEntriesSize());
              if (!thread.offer(request)) {
                logger.debug(
                    "{}: Log queue of {} is full, ignore the log to this node, searchIndex: {}",
                    impl.getThisNode().getGroupId(),
                    thread.getPeer(),
                    request.getSearchIndex());
              }
            });
      }
    }
  }
Manage all asynchronous replication threads and corresponding async clients.
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java