basic-lib/src/main/java/com/streamsets/pipeline/stage/processor/aggregation/aggregator/AggregatorDataProvider.java
public Map<Aggregator, AggregatorData> roll(long newDataWindowEndTimeMillis) {
    Utils.checkState(started, "Not started");
    Utils.checkState(!stopped, "Already stopped");

    Map<Aggregator, AggregatorData> result = data;
    Map<Aggregator, AggregatorData> newData = new ConcurrentHashMap<>();
    for (Aggregator aggregator : aggregators) {
      newData.put(aggregator, aggregator.createAggregatorData(newDataWindowEndTimeMillis));
    }
    data = newData;

    Map<Aggregator, AggregatorData> oldData = result;
    // In case of sliding window, aggregate the data windows to get the result
    result = aggregateDataWindows(result);

    if (currentDataWindow != null) {
      currentDataWindow.setDataAndClose(oldData);
    }
    DataWindow newDataWindow = createDataWindow(newDataWindowEndTimeMillis);
    synchronized (dataWindowQueue) {
      dataWindowQueue.add(newDataWindow);
      dataWindowList = new ArrayList<>(dataWindowQueue);
    }
    currentDataWindow = newDataWindow;
    return result;
  }
public Map<Aggregator, AggregatorData> roll(long newDataWindowEndTimeMillis) {
    Utils.checkState(started, "Not started");
    Utils.checkState(!stopped, "Already stopped");

    Map<Aggregator, AggregatorData> result = data;
    Map<Aggregator, AggregatorData> newData = new ConcurrentHashMap<>();
    for (Aggregator aggregator : aggregators) {
      newData.put(aggregator, aggregator.createAggregatorData(newDataWindowEndTimeMillis));
    }
    data = newData;

    Map<Aggregator, AggregatorData> oldData = result;
    // In case of sliding window, aggregate the data windows to get the result
    result = aggregateDataWindows(result);

    if (currentDataWindow != null) {
      currentDataWindow.setDataAndClose(oldData);
    }
    DataWindow newDataWindow = createDataWindow(newDataWindowEndTimeMillis);
    synchronized (dataWindowQueue) {
      dataWindowQueue.add(newDataWindow);
      dataWindowList = new ArrayList<>(dataWindowQueue);
    }
    currentDataWindow = newDataWindow;
    return result;
  }
Atomically rolls the DataWindow of all aggregators associated with the AggregatorDataProvider.

@param newDataWindowEndTimeMillis ending time of the new DataWindow.
@return a Map with all the Aggregators data of the DataWindow that closed.