public void start() throws IOException, InterruptedException {
    super.start();
    MessageConsumer consumeFn = new MessageConsumer();
    StreamApplication app = appDesc -> {
      String systemFactoryName = new SystemConfig(config).getSystemFactory(systemName).get();
      GenericSystemDescriptor sd = new GenericSystemDescriptor(systemName, systemFactoryName);
      GenericInputDescriptor<Object> isd = sd.getInputDescriptor(streamId, new NoOpSerde<>());
      MessageStream<Object> stream = appDesc.getInputStream(isd);
      stream.map(consumeFn);
    };
    ApplicationRunner runner = ApplicationRunners.getApplicationRunner(app, new MapConfig());

    runner.run();

    while (consumeFn.getEventsConsumed() < totalEvents) {
      Thread.sleep(10);
    }

    Instant endTime = Instant.now();

    runner.kill();

    System.out.println("\n*******************");
    System.out.println(String.format("Started at %s Ending at %s ", consumeFn.startTime, endTime));
    System.out.println(String.format("Event Rate is %s Messages/Sec ",
        consumeFn.getEventsConsumed() * 1000 / Duration.between(consumeFn.startTime, Instant.now()).toMillis()));

    System.out.println(
        "Event Rate is " + consumeFn.getEventsConsumed() * 1000 / Duration.between(consumeFn.startTime, endTime).toMillis());
    System.out.println("*******************\n");

    System.exit(0);
  }
Generic benchmark test for a test consumer but with samza framework
samza-tools/src/main/java/org/apache/samza/tools/benchmark/SystemConsumerWithSamzaBench.java