@Override
  public void process(final Event event, final EventCollector eventCollector) throws Exception {
    final AtomicLong lastReceiveTime =
        pipeName2LastValueReceiveTimeMap.computeIfAbsent(
            pipeName, key -> new AtomicLong(System.currentTimeMillis()));

    final long previousTime = lastReceiveTime.get();

    if (System.currentTimeMillis() - previousTime > outputMaxDelayMilliseconds) {
      final AtomicReference<Exception> exception = new AtomicReference<>();

      pipeName2timeSeries2TimeSeriesRuntimeStateMap
          .get(pipeName)
          .keySet()
          .forEach(
              timeSeries -> {
                final AtomicReference<TimeSeriesRuntimeState> stateReference =
                    pipeName2timeSeries2TimeSeriesRuntimeStateMap.get(pipeName).get(timeSeries);
                synchronized (stateReference) {
                  final PipeRowCollector rowCollector =
                      new PipeRowCollector(pipeTaskMeta, null, dataBaseName, isTableModel);
                  try {
                    collectWindowOutputs(
                        stateReference.get().forceOutput(), timeSeries, rowCollector);
                  } catch (final IOException e) {
                    exception.set(e);
                  }
                  rowCollector
                      .convertToTabletInsertionEvents(false)
                      .forEach(
                          tabletEvent -> {
                            try {
                              eventCollector.collect(tabletEvent);
                            } catch (Exception e) {
                              exception.set(e);
                            }
                          });
                }
              });
      if (exception.get() != null) {
        // Retry at the fixed interval
        lastReceiveTime.set(System.currentTimeMillis());
        throw exception.get();
      }
      // Forbidding emitting results until next data comes
      // If the last receive time has changed, it means new data has come
      // thus the next output is needed
      lastReceiveTime.compareAndSet(previousTime, Long.MAX_VALUE);
    }

    eventCollector.collect(event);
  }
{@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