@Override
protected void _run(String[] arguments) throws Exception {
JobAppOptions options=extractOptions(arguments);
Cluster cluster=options.clusterId.isEmpty() ? defaultCluster : applicationContext.getBean(options.clusterId,Cluster.class);
MavenManagedJar jar=options.jarId.isEmpty() ? defaultJar : applicationContext.getBean(options.jarId,MavenManagedJar.class);
Map<String,String> env=System.getenv();
if(options.runningCluster.isEmpty() && env.containsKey("RUNNING_CLUSTER")) {
options.runningCluster=env.get("RUNNING_CLUSTER");
logger.info("Getting AWS Job Flow ID ["+options.runningCluster+"] from RUNNING_CLUSTER environment variable");
}
if (!(options.runningCluster.isEmpty())) {
cluster=new PersistentCluster(options.runningCluster);
applicationContext.getAutowireCapableBeanFactory().autowireBean(cluster);
}
if(jar.getFirstArgumentIsNotAPath()) {
if(options.remainingArguments.isEmpty())
usage();
String firstArgument=options.remainingArguments.get(0);
if (firstArgument.contains(":") || firstArgument.contains("/"))
usage();
};
List<String> jarArgs=jar.getHeadArguments();
jarArgs.addAll(options.remainingArguments);
cluster.runJob(jar,jarArgs);
}