public void start() throws IOException, InterruptedException {

    super.start();
    String source = "SystemProducerBench";

    int size = Integer.parseInt(cmd.getOptionValue(OPT_SHORT_MESSAGE_SIZE));
    RandomValueGenerator randGenerator = new RandomValueGenerator(System.currentTimeMillis());
    value = randGenerator.getNextString(size, size).getBytes();

    NoOpMetricsRegistry metricsRegistry = new NoOpMetricsRegistry();
    List<SystemStreamPartition> ssps = createSSPs(systemName, physicalStreamName, startPartition, endPartition);
    SystemProducer producer = factory.getProducer(systemName, config, metricsRegistry);
    producer.register(source);
    producer.start();

    System.out.println("starting production at " + Instant.now());
    Instant startTime = Instant.now();
    for (int index = 0; index < totalEvents; index++) {
      SystemStreamPartition ssp = ssps.get(index % ssps.size());
      OutgoingMessageEnvelope messageEnvelope = createMessageEnvelope(ssp, index);
      producer.send(source, messageEnvelope);
    }

    System.out.println("Ending production at " + Instant.now());
    System.out.println(String.format("Event Rate is %s Messages/Sec",
        totalEvents * 1000 / Duration.between(startTime, Instant.now()).toMillis()));

    producer.flush(source);

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