public void start() throws IOException, InterruptedException {
    super.start();
    SystemAdmin systemAdmin = factory.getAdmin(systemName, config);
    SystemStreamMetadata ssm =
        systemAdmin.getSystemStreamMetadata(Collections.singleton(physicalStreamName)).get(physicalStreamName);

    NoOpMetricsRegistry metricsRegistry = new NoOpMetricsRegistry();
    Set<SystemStreamPartition> ssps = createSSPs(systemName, physicalStreamName, startPartition, endPartition);
    SystemConsumer consumer = factory.getConsumer(systemName, config, metricsRegistry);
    for (SystemStreamPartition ssp : ssps) {
      consumer.register(ssp, ssm.getSystemStreamPartitionMetadata().get(ssp.getPartition()).getOldestOffset());
    }

    consumer.start();

    System.out.println("starting consumption at " + Instant.now());
    Instant startTime = Instant.now();
    int numEvents = 0;
    while (numEvents < totalEvents) {
      Map<SystemStreamPartition, List<IncomingMessageEnvelope>> pollResult = consumer.poll(ssps, 2000);
      numEvents += pollResult.values().stream().mapToInt(List::size).sum();
    }

    System.out.println("Ending consumption at " + Instant.now());
    System.out.println(String.format("Event Rate is %s Messages/Sec ",
        numEvents * 1000 / Duration.between(startTime, Instant.now()).toMillis()));
    consumer.stop();
    System.exit(0);
  }
Generic benchmark test for {@link SystemConsumer}.
samza-tools/src/main/java/org/apache/samza/tools/benchmark/SystemConsumerBench.java