private void asyncTransferQueuedEventsIfNecessary() {
    long retryStartTime = System.currentTimeMillis();
    while (!retryEventQueue.isEmpty()) {
      synchronized (this) {
        if (isClosed.get() || retryEventQueue.isEmpty()) {
          return;
        }
        if (System.currentTimeMillis() - retryStartTime > TimeUnit.SECONDS.toMillis(20)) {
          // just in case that some events are polled and re-added into queue again and again,
          // causing this loop to run forever.
          LOGGER.warn(
              "PipeConsensus-ConsensusGroup-{}: retryEventQueue is not empty after 20 seconds. retryQueue size: {}",
              consensusGroupId,
              retryEventQueue.size());
          return;
        }

        // remove this event from queue. If retry fail as well, event will be re-added into
        // retryQueue.
        final EnrichedEvent peekedEvent = retryEventQueue.poll();
        // retry with interval when necessarily
        long retryInterval =
            peekedEvent.getRetryInterval() > EnrichedEvent.INITIAL_RETRY_INTERVAL_FOR_IOTV2
                ? peekedEvent.getRetryInterval()
                : 0L;
        LOGGER.info(
            "PipeConsensus-ConsensusGroup-{}: retry with interval {} for {}",
            consensusGroupId,
            retryInterval,
            peekedEvent);
        // need to retry in background service, otherwise the retryInterval will block the sender
        // procedure.
        backgroundTaskService.schedule(
            () -> {
              // do transfer
              if (peekedEvent instanceof PipeInsertNodeTabletInsertionEvent) {
                retryTransfer((PipeInsertNodeTabletInsertionEvent) peekedEvent);
              } else if (peekedEvent instanceof PipeTsFileInsertionEvent) {
                retryTransfer((PipeTsFileInsertionEvent) peekedEvent);
              } else if (peekedEvent instanceof PipeDeleteDataNodeEvent) {
                retryTransfer((PipeDeleteDataNodeEvent) peekedEvent);
              } else {
                if (LOGGER.isWarnEnabled()) {
                  LOGGER.warn(
                      "PipeConsensusAsyncConnector does not support transfer generic event: {}.",
                      peekedEvent);
                }
              }
            },
            retryInterval,
            TimeUnit.MILLISECONDS);
      }
    }
  }
Transfer queued {@link Event}s which are waiting for retry.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusAsyncConnector.java