private void transferQueuedEventsIfNecessary(final boolean forced) {
    if (retryEventQueue.isEmpty()
        || (!forced
            && retryEventQueueEventCounter.getTabletInsertionEventCount()
                < PipeConfig.getInstance()
                    .getPipeAsyncConnectorForcedRetryTabletEventQueueSizeThreshold()
            && retryEventQueueEventCounter.getTsFileInsertionEventCount()
                < PipeConfig.getInstance()
                    .getPipeAsyncConnectorForcedRetryTsFileEventQueueSizeThreshold()
            && retryEventQueue.size()
                < PipeConfig.getInstance()
                    .getPipeAsyncConnectorForcedRetryTotalEventQueueSizeThreshold())) {
      return;
    }

    final long retryStartTime = System.currentTimeMillis();
    final int remainingEvents = retryEventQueue.size();
    while (!retryEventQueue.isEmpty()) {
      synchronized (this) {
        if (isClosed.get()) {
          return;
        }
        if (retryEventQueue.isEmpty()) {
          break;
        }

        final Event peekedEvent = retryEventQueue.peek();

        if (peekedEvent instanceof PipeInsertNodeTabletInsertionEvent) {
          retryTransfer((PipeInsertNodeTabletInsertionEvent) peekedEvent);
        } else if (peekedEvent instanceof PipeRawTabletInsertionEvent) {
          retryTransfer((PipeRawTabletInsertionEvent) peekedEvent);
        } else if (peekedEvent instanceof PipeTsFileInsertionEvent) {
          retryTransfer((PipeTsFileInsertionEvent) peekedEvent);
        } else {
          LOGGER.warn(
              "IoTDBThriftAsyncConnector does not support transfer generic event: {}.",
              peekedEvent);
        }

        final Event polledEvent = retryEventQueue.poll();
        retryEventQueueEventCounter.decreaseEventCount(polledEvent);
        if (polledEvent != peekedEvent) {
          LOGGER.error(
              "The event polled from the queue is not the same as the event peeked from the queue. "
                  + "Peeked event: {}, polled event: {}.",
              peekedEvent,
              polledEvent);
        }
        if (polledEvent != null && LOGGER.isDebugEnabled()) {
          LOGGER.debug("Polled event {} from retry queue.", polledEvent);
        }
      }

      // Stop retrying if the execution time exceeds the threshold for better realtime performance
      if (System.currentTimeMillis() - retryStartTime
          > PipeConfig.getInstance().getPipeAsyncConnectorMaxRetryExecutionTimeMsPerCall()) {
        if (retryEventQueueEventCounter.getTabletInsertionEventCount()
                < PipeConfig.getInstance()
                    .getPipeAsyncConnectorForcedRetryTabletEventQueueSizeThreshold()
            && retryEventQueueEventCounter.getTsFileInsertionEventCount()
                < PipeConfig.getInstance()
                    .getPipeAsyncConnectorForcedRetryTsFileEventQueueSizeThreshold()
            && retryEventQueue.size()
                < PipeConfig.getInstance()
                    .getPipeAsyncConnectorForcedRetryTotalEventQueueSizeThreshold()) {
          return;
        }

        if (remainingEvents <= retryEventQueue.size()) {
          throw new PipeException(
              "Failed to retry transferring events in the retry queue. Remaining events: "
                  + retryEventQueue.size()
                  + " (tablet events: "
                  + retryEventQueueEventCounter.getTabletInsertionEventCount()
                  + ", tsfile events: "
                  + retryEventQueueEventCounter.getTsFileInsertionEventCount()
                  + ").");
        }
      }
    }
  }
Transfer queued {@link Event}s which are waiting for retry.

@see PipeConnector#transfer(Event) for more details.
@see PipeConnector#transfer(TabletInsertionEvent) for more details.
@see PipeConnector#transfer(TsFileInsertionEvent) for more details.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBDataRegionAsyncConnector.java