@Test
public void testCLI_insufficientArg() {
  UnitTestHelper.setLog4jLevel(Parser.class, Level.FATAL);
    assertThrows(ParseException.class, () ->
            new CLIBuilder().with(ParserTopologyCLI.ParserOptions.BROKER_URL, "mybroker")
            .with(ParserTopologyCLI.ParserOptions.ZK_QUORUM, "myzk")
            .build(true));
  UnitTestHelper.setLog4jLevel(Parser.class, Level.ERROR);
}
 

Parser createParser() {
  Parser result = new BasicParser();
  return result;
}
 

Parser createParser() {
  Parser result = new BasicParser();
  return result;
}
 

Parser createParser() {
  Parser result = new BasicParser();
  return result;
}
 

Parser createParser() {
	Parser result = new BasicParser();
	return result;
}
 

@Override
public int run(String[] args) throws Exception {
	CommandLineParser cli = new CommandLineParser();
	if (args.length == 0) {
		cli.printUsage();
		return 1;
	}
	
	cli.addOption("input", false, "input path to the maps", "path");
	cli.addOption("output", false, "output path from the reduces", "path");
	
	cli.addOption("cpubin", false, "URI to application cpu executable", "class");
	cli.addOption("gpubin", false, "URI to application gpu executable", "class");
	
	Parser parser = cli.createParser();
	try {
		GenericOptionsParser genericParser = new GenericOptionsParser(getConf(), args);
		CommandLine results = 
			parser.parse(cli.options, genericParser.getRemainingArgs());
		JobConf job = new JobConf(getConf());
		
		if (results.hasOption("input")) {
			FileInputFormat.setInputPaths(job, (String) results.getOptionValue("input"));
		}
        if (results.hasOption("output")) {
        	FileOutputFormat.setOutputPath(job, new Path((String) results.getOptionValue("output")));
        }
        if (results.hasOption("cpubin")) {
        	setCPUExecutable(job, (String) results.getOptionValue("cpubin"));
        }
        if (results.hasOption("gpubin")) {
        	setGPUExecutable(job, (String) results.getOptionValue("gpubin"));
        }
        // if they gave us a jar file, include it into the class path
        String jarFile = job.getJar();
        if (jarFile != null) {
        	final URL[] urls = new URL[] { FileSystem.getLocal(job).
        			pathToFile(new Path(jarFile)).toURL() };
            //FindBugs complains that creating a URLClassLoader should be
            //in a doPrivileged() block. 
            ClassLoader loader =
                AccessController.doPrivileged(
                    new PrivilegedAction<ClassLoader>() {
                      public ClassLoader run() {
                        return new URLClassLoader(urls);
                      }
                    }
                  );
              job.setClassLoader(loader);
        }
        runJob(job);
        return 0;
	} catch (ParseException pe) {
		LOG.info("Error :" + pe);
		cli.printUsage();
		return 1;
	}
}
 

Parser createParser() {
  Parser result = new BasicParser();
  return result;
}
 

/**
 * Get parser.
 * @return Parser
 */
protected abstract Parser parser();