@Override
  public int run(InputStream ins, PrintStream outs, PrintStream err, List<String> args) throws Exception {

    String[] argarry = args.toArray(new String[0]);
    Options opts = new Options();

    Option helpopt = OptionBuilder.hasArg(false).withDescription("print this message").create("help");

    Option inopt = OptionBuilder.hasArg().isRequired().withDescription("comma-separated input paths").create("in");

    Option outopt = OptionBuilder.hasArg().isRequired().withDescription("The output path.").create("out");

    Option pargs = OptionBuilder.hasArg().withDescription(
        "A string containing the command line arguments to pass to the tethered process. String should be enclosed in quotes")
        .create("exec_args");

    Option popt = OptionBuilder.hasArg().isRequired().withDescription("executable program, usually in HDFS")
        .create("program");

    Option outscopt = OptionBuilder.withType(File.class).hasArg().isRequired()
        .withDescription("schema file for output of reducer").create("outschema");

    Option outscmapopt = OptionBuilder.withType(File.class).hasArg()
        .withDescription("(optional) map output schema file,  if different from outschema").create("outschemamap");

    Option redopt = OptionBuilder.withType(Integer.class).hasArg().withDescription("(optional) number of reducers")
        .create("reduces");

    Option cacheopt = OptionBuilder.withType(Boolean.class).hasArg()
        .withDescription(
            "(optional) boolean indicating whether or not the executable should be distributed via distributed cache")
        .create("exec_cached");

    Option protoopt = OptionBuilder.hasArg()
        .withDescription("(optional) specifies the transport protocol 'http' or 'sasl'").create("protocol");

    opts.addOption(redopt);
    opts.addOption(outscopt);
    opts.addOption(popt);
    opts.addOption(pargs);
    opts.addOption(inopt);
    opts.addOption(outopt);
    opts.addOption(helpopt);
    opts.addOption(outscmapopt);
    opts.addOption(cacheopt);
    opts.addOption(protoopt);

    CommandLineParser parser = new GnuParser();

    CommandLine line = null;
    HelpFormatter formatter = new HelpFormatter();

    JobConf job = new JobConf();

    try {
      line = parser.parse(opts, argarry);

      if (line.hasOption("help")) {
        formatter.printHelp("tether", opts);
        return 0;
      }

      FileInputFormat.addInputPaths(job, line.getOptionValue("in"));
      FileOutputFormat.setOutputPath(job, new Path(line.getOptionValue("out")));

      List<String> exargs = null;
      Boolean cached = false;

      if (line.hasOption("exec_args")) {
        String[] splitargs = line.getOptionValue("exec_args").split(" ");
        exargs = new ArrayList<>(Arrays.asList(splitargs));
      }
      if (line.hasOption("exec_cached")) {
        cached = Boolean.parseBoolean(line.getOptionValue("exec_cached"));
      }
      TetherJob.setExecutable(job, new File(line.getOptionValue("program")), exargs, cached);

      File outschema = (File) line.getParsedOptionValue("outschema");
      job.set(AvroJob.OUTPUT_SCHEMA, Schema.parse(outschema).toString());
      if (line.hasOption("outschemamap")) {
        job.set(AvroJob.MAP_OUTPUT_SCHEMA,
            new Schema.Parser().parse((File) line.getParsedOptionValue("outschemamap")).toString());
      }
      if (line.hasOption("reduces")) {
        job.setNumReduceTasks((Integer) line.getParsedOptionValue("reduces"));
      }
      if (line.hasOption("protocol")) {
        TetherJob.setProtocol(job, line.getOptionValue("protocol"));
      }
    } catch (Exception exp) {
      System.out.println("Unexpected exception: " + exp.getMessage());
      formatter.printHelp("tether", opts);
      return -1;
    }

    TetherJob.runJob(job);
    return 0;
  }
@Test
  void test() throws Exception {

    // Create the schema files.
    Schema outscheme = new Pair<Utf8, Long>(new Utf8(""), 0L).getSchema();

    // we need to write the schemas to a file
    File midscfile = new File(INPUT_DIR.getPath(), "midschema.avpr");
    try (FileWriter hf = new FileWriter(midscfile)) {
      hf.write(outscheme.toString());
    }

    JobConf job = new JobConf();
    String inputPathStr = INPUT_DIR.getPath();
    String outputPathStr = OUTPUT_DIR.getPath();
    Path outputPath = new Path(outputPathStr);

    outputPath.getFileSystem(job).delete(outputPath, true);

    // create the input file
    WordCountUtil.writeLinesFile(inputPathStr + "/lines.avro");

    // create a string of the arguments
    String execargs = "-classpath " + System.getProperty("java.class.path");
    execargs += " org.apache.avro.mapred.tether.WordCountTask";

    // Create a list of the arguments to pass to the tull run method
    java.util.List<String> runargs = new java.util.ArrayList<>();

    runargs.addAll(java.util.Arrays.asList("--program", "java"));
    runargs.addAll(asList("--exec_args", '"' + execargs + '"'));
    runargs.addAll(asList("--exec_cached", "false"));
    runargs.addAll(asList("--in", inputPathStr));
    runargs.addAll(asList("--out", outputPath.toString()));
    runargs.addAll(asList("--outschema", midscfile.toString()));

    TetherTool tool = new TetherTool();

    tool.run(null, null, System.err, runargs);

    // TODO:: We should probably do some validation
    // validate the output
    int numWords = 0;
    DatumReader<Pair<Utf8, Long>> reader = new SpecificDatumReader<>();
    try (InputStream cin = new BufferedInputStream(new FileInputStream(outputPathStr + "/part-00000.avro"));
        DataFileStream<Pair<Utf8, Long>> counts = new DataFileStream<>(cin, reader)) {
      for (Pair<Utf8, Long> wc : counts) {
        Assertions.assertEquals(WordCountUtil.COUNTS.get(wc.key().toString()), wc.value(), wc.key().toString());
        numWords++;
      }
    }
    Assertions.assertEquals(WordCountUtil.COUNTS.size(), numWords);
  }