@Override
  public void addMoreSystemConfigs(Properties props) {
    props.put(ApplicationConfig.APP_RUNNER_CLASS, LocalApplicationRunner.class.getName());
    List<Integer> partitions = IntStream.range(startPartition, endPartition).boxed().collect(Collectors.toList());
    props.put(ApplicationConfig.APP_NAME, "SamzaBench");
    props.put(JobConfig.PROCESSOR_ID, "1");
    props.put(JobCoordinatorConfig.JOB_COORDINATOR_FACTORY, PassthroughJobCoordinatorFactory.class.getName());
    props.put(String.format(ConfigBasedSspGrouperFactory.CONFIG_STREAM_PARTITIONS, streamId),
        Joiner.on(",").join(partitions));
    props.put(TaskConfig.GROUPER_FACTORY, ConfigBasedSspGrouperFactory.class.getName());
  }
Generic benchmark test for a test consumer but with samza framework
samza-tools/src/main/java/org/apache/samza/tools/benchmark/SystemConsumerWithSamzaBench.java