public void consumeTabletInsertionEventsWithRetry(
      final TabletInsertionEventConsumer consumer, final String callerName) throws PipeException {
    final Iterable<TabletInsertionEvent> iterable = toTabletInsertionEvents();
    final Iterator<TabletInsertionEvent> iterator = iterable.iterator();
    int tabletEventCount = 0;
    while (iterator.hasNext()) {
      final TabletInsertionEvent parsedEvent = iterator.next();
      tabletEventCount++;
      int retryCount = 0;
      while (true) {
        // If failed due do insufficient memory, retry until success to avoid race among multiple
        // processor threads
        try {
          consumer.consume((PipeRawTabletInsertionEvent) parsedEvent);
          break;
        } catch (final PipeRuntimeOutOfMemoryCriticalException e) {
          if (retryCount++ % 100 == 0) {
            LOGGER.warn(
                "{}: failed to allocate memory for parsing TsFile {}, tablet event no. {}, retry count is {}, will keep retrying.",
                callerName,
                getTsFile(),
                tabletEventCount,
                retryCount,
                e);
          } else if (LOGGER.isDebugEnabled()) {
            LOGGER.debug(
                "{}: failed to allocate memory for parsing TsFile {}, tablet event no. {}, retry count is {}, will keep retrying.",
                callerName,
                getTsFile(),
                tabletEventCount,
                retryCount,
                e);
          }
        }
      }
    }
  }
Get ProgressIndex without waiting for tsfile close. Can be used in getting progressIndex when
memTable becomes immutable.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java