@Override
  public void process(
      final TsFileInsertionEvent tsFileInsertionEvent, final EventCollector eventCollector)
      throws Exception {
    try {
      if (tsFileInsertionEvent instanceof PipeTsFileInsertionEvent) {
        final AtomicReference<Exception> ex = new AtomicReference<>();
        ((PipeTsFileInsertionEvent) tsFileInsertionEvent)
            .consumeTabletInsertionEventsWithRetry(
                event -> {
                  try {
                    process(event, eventCollector);
                  } catch (Exception e) {
                    ex.set(e);
                  }
                },
                "AggregateProcessor::process");
        if (ex.get() != null) {
          throw ex.get();
        }
      } else {
        for (final TabletInsertionEvent tabletInsertionEvent :
            tsFileInsertionEvent.toTabletInsertionEvents()) {
          process(tabletInsertionEvent, eventCollector);
        }
      }
    } finally {
      tsFileInsertionEvent.close();
    }
    // The timeProgressIndex shall only be reported by the output events
    // whose progressIndex is bounded with tablet events
    if (tsFileInsertionEvent instanceof PipeTsFileInsertionEvent) {
      ((PipeTsFileInsertionEvent) tsFileInsertionEvent).skipReportOnCommit();
    }
  }
{@link AggregateProcessor} is a {@link PipeProcessor} that can adopt different implementations of
{@link AbstractWindowingProcessor} as windowing strategy and use calculation methods from all the
{@link AbstractOperatorProcessor}s to calculate the given operators. Both the {@link
AbstractWindowingProcessor} and {@link AbstractOperatorProcessor} can be implemented by user and
loaded as a normal {@link PipeProcessor}
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java