@Override
  public int run(InputStream in, PrintStream out, PrintStream err, List<String> args) throws Exception {
    OptionParser optParser = new OptionParser();
    OptionSpec<Long> offsetOpt = optParser.accepts("offset", "offset for reading input").withRequiredArg()
        .ofType(Long.class).defaultsTo(Long.valueOf(0));
    OptionSpec<Long> limitOpt = optParser.accepts("limit", "maximum number of records in the outputfile")
        .withRequiredArg().ofType(Long.class).defaultsTo(Long.MAX_VALUE);
    OptionSpec<Double> fracOpt = optParser.accepts("samplerate", "rate at which records will be collected")
        .withRequiredArg().ofType(Double.class).defaultsTo(Double.valueOf(1));

    OptionSet opts = optParser.parse(args.toArray(new String[0]));
    List<String> nargs = (List<String>) opts.nonOptionArguments();
    if (nargs.size() < 2) {
      printHelp(out);
      return 0;
    }

    inFiles = Util.getFiles(nargs.subList(0, nargs.size() - 1));

    System.out.println("List of input files:");
    for (Path p : inFiles) {
      System.out.println(p);
    }
    currentInput = -1;
    nextInput();

    OutputStream output = out;
    String lastArg = nargs.get(nargs.size() - 1);
    if (nargs.size() > 1 && !lastArg.equals("-")) {
      output = Util.createFromFS(lastArg);
    }
    writer = new DataFileWriter<>(new GenericDatumWriter<>());

    String codecName = reader.getMetaString(DataFileConstants.CODEC);
    CodecFactory codec = (codecName == null) ? CodecFactory.fromString(DataFileConstants.NULL_CODEC)
        : CodecFactory.fromString(codecName);
    writer.setCodec(codec);
    for (String key : reader.getMetaKeys()) {
      if (!DataFileWriter.isReservedMeta(key)) {
        writer.setMeta(key, reader.getMeta(key));
      }
    }
    writer.create(schema, output);

    long offset = opts.valueOf(offsetOpt);
    long limit = opts.valueOf(limitOpt);
    double samplerate = opts.valueOf(fracOpt);
    sampleCounter = 1;
    totalCopied = 0;
    reuse = null;

    if (limit < 0) {
      System.out.println("limit has to be non-negative");
      this.printHelp(out);
      return 1;
    }
    if (offset < 0) {
      System.out.println("offset has to be non-negative");
      this.printHelp(out);
      return 1;
    }
    if (samplerate < 0 || samplerate > 1) {
      System.out.println("samplerate has to be a number between 0 and 1");
      this.printHelp(out);
      return 1;
    }

    skip(offset);
    writeRecords(limit, samplerate);
    System.out.println(totalCopied + " records written.");

    writer.flush();
    writer.close();
    Util.close(out);
    return 0;
  }
@Test
  void helpfulMessageWhenNoArgsGiven() throws Exception {
    ByteArrayOutputStream buffer = new ByteArrayOutputStream(1024);
    int returnCode;
    try (PrintStream out = new PrintStream(buffer)) {
      returnCode = new CatTool().run(System.in, out, System.err, Collections.emptyList());
    }

    assertEquals(0, returnCode);
    assertTrue(buffer.toString().trim().length() > 200, "should have lots of help");
  }