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;
  }
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.
Atomically rolls the DataWindow of all aggregators associated with the AggregatorDataProvider .