class RollingSort extends StormBenchmark {

  @Override
  public StormTopology getTopology(Config config) {
    final int spoutNum = BenchmarkUtils.getInt(config, SPOUT_NUM, DEFAULT_SPOUT_NUM);
    final int boltNum =  BenchmarkUtils.getInt(config, SORT_BOLT_NUM, DEFAULT_SORT_BOLT_NUM);
    final int msgSize = BenchmarkUtils.getInt(config, RandomMessageSpout.MESSAGE_SIZE,
            RandomMessageSpout.DEFAULT_MESSAGE_SIZE);
    final int chunkSize = BenchmarkUtils.getInt(config, SortBolt.CHUNK_SIZE,
            SortBolt.DEFAULT_CHUNK_SIZE);
    final int emitFreq = BenchmarkUtils.getInt(config, SortBolt.EMIT_FREQ,
            SortBolt.DEFAULT_EMIT_FREQ);
    spout = new RandomMessageSpout(msgSize, BenchmarkUtils.ifAckEnabled(config));
    TopologyBuilder builder = new TopologyBuilder();
    builder.setSpout(SPOUT_ID, spout, spoutNum);
    builder.setBolt(SORT_BOLT_ID, new SortBolt(emitFreq, chunkSize), boltNum)
            .localOrShuffleGrouping(SPOUT_ID);
    return builder.createTopology();
  }

}