static void run(
      final SystemExit systemExit,
      final String... args
  ) throws Throwable {
    final Arguments arguments = new Arguments.Builder()
        .parseArgs(args)
        .build();

    if (arguments.help) {
      usage();
      return;
    }

    final Properties props = getProperties(arguments);
    final DataGenProducer dataProducer = ProducerFactory
        .getProducer(arguments.keyFormat, arguments.valueFormat, arguments.valueDelimiter, props);
    final Optional<RateLimiter> rateLimiter = arguments.msgRate != -1
        ? Optional.of(RateLimiter.create(arguments.msgRate)) : Optional.empty();

    final Executor executor = Executors.newFixedThreadPool(
        arguments.numThreads,
        r -> {
          final Thread thread = new Thread(r);
          thread.setDaemon(true);
          return thread;
        }
    );
    final CompletionService<Void> service = new ExecutorCompletionService<>(executor);

    for (int i = 0; i < arguments.numThreads; i++) {
      service.submit(getProducerTask(arguments, dataProducer, props, rateLimiter));
    }
    for (int i = 0; i < arguments.numThreads; i++) {
      try {
        service.take().get();
      } catch (final InterruptedException e) {
        System.err.println("Interrupted waiting for threads to exit.");
        systemExit.exit(1);
      } catch (final ExecutionException e) {
        throw e.getCause();
      }
    }
  }
@Test(expected = DataGen.Arguments.ArgumentParseException.class)
  public void valueDelimiterCanOnlyBeSingleCharacter() throws Throwable {
    DataGen.run(
        mockSystem,
        "schema=./src/main/resources/purchase.avro",
        "key=id",
        "format=delimited",
        "value_delimiter=@@",
        "topic=foo"
    );
  }