@Override
  protected boolean executeOnce() throws Exception {
    if (isClosed.get()) {
      return false;
    }

    final Event event = lastEvent != null ? lastEvent : extractor.supply();
    // Record the last event for retry when exception occurs
    setLastEvent(event);

    try {
      if (event == null) {
        return false;
      }

      if (!(event instanceof ProgressReportEvent)) {
        outputPipeConnector.transfer(event);
        PipeConfigRegionConnectorMetrics.getInstance().markConfigEvent(taskID);
      }
      decreaseReferenceCountAndReleaseLastEvent(event, true);

    } catch (final PipeException e) {
      setLastExceptionEvent(event);
      if (!isClosed.get()) {
        throw e;
      } else {
        LOGGER.info(
            "{} in pipe transfer, ignored because pipe is dropped.",
            e.getClass().getSimpleName(),
            e);
        clearReferenceCountAndReleaseLastEvent(event);
      }
    } catch (final Exception e) {
      setLastExceptionEvent(event);
      if (!isClosed.get()) {
        throw new PipeException(
            String.format(
                "Exception in pipe transfer, subtask: %s, last event: %s", taskID, lastEvent),
            e);
      } else {
        LOGGER.info("Exception in pipe transfer, ignored because pipe is dropped.", e);
        clearReferenceCountAndReleaseLastEvent(event);
      }
    }

    return true;
  }
Try to consume an {@link Event} by the {@link IoTDBConfigRegionExtractor}.

@return {@code true} if the {@link Event} is consumed successfully, {@code false} if no more
    {@link Event} can be consumed
@throws Exception if any error occurs when consuming the {@link Event}
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtask.java