basic-lib/src/main/java/com/streamsets/pipeline/stage/processor/aggregation/aggregator/Aggregators.java
@SuppressWarnings("unchecked")
  public <A extends SimpleAggregator> A createSimple(String name, Class<? extends Aggregator> klass) {
    Utils.checkState(!started, "Already started");
    try {
      A aggregator = (A) CONSTRUCTORS.get(klass).newInstance(name);
      dataProvider.addAggregator(aggregator);
      aggregator.setDataProvider(dataProvider);
      return aggregator;
    } catch (Exception ex) {
      throw new RuntimeException(ex);
    }
  }
@SuppressWarnings("unchecked")
  public <A extends SimpleAggregator> A createSimple(String name, Class<? extends Aggregator> klass) {
    Utils.checkState(!started, "Already started");
    try {
      A aggregator = (A) CONSTRUCTORS.get(klass).newInstance(name);
      dataProvider.addAggregator(aggregator);
      aggregator.setDataProvider(dataProvider);
      return aggregator;
    } catch (Exception ex) {
      throw new RuntimeException(ex);
    }
  }
@ SuppressWarnings ( "unchecked" ) public < A extends SimpleAggregator > A createSimple ( String name , Class < ? extends Aggregator > klass ) { Utils . checkState ( ! started , "Already started" ) ; try { A aggregator = ( A ) CONSTRUCTORS . get ( klass ) . newInstance ( name ) ; dataProvider . addAggregator ( aggregator ) ; aggregator . setDataProvider ( dataProvider ) ; return aggregator ; } catch ( Exception ex ) { throw new RuntimeException ( ex ) ; } }
Creates a simple Aggregator.

@param name name of the aggregator.
@param klass Aggregator class
@return a simple aggregator instance.
Creates a simple Aggregator .