Fixed backwards compatibility for ExecuteSparkJob processor
@@ -151,7 +151,6 @@ public class ExecuteSparkJob extends AbstractNiFiProcessor {
.description("The deploy mode for YARN master (client, cluster). Only applicable for yarn mode. "
+ "NOTE: Please ensure that you have not set this in your application.")
.required(false)
- .defaultValue("client")
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.expressionLanguageSupported(true)
.build();
@@ -581,6 +580,7 @@ public class ExecuteSparkJob extends AbstractNiFiProcessor {
.build());
}
+ if (StringUtils.isNotEmpty(sparkYarnDeployMode)) {
if ((!sparkMaster.contains("local")) && (!sparkMaster.equals("yarn")) && (!sparkMaster.contains("mesos")) && (!sparkMaster.contains("spark"))) {
results.add(new ValidationResult.Builder()
.subject(this.getClass().getSimpleName())
@@ -597,6 +597,7 @@ public class ExecuteSparkJob extends AbstractNiFiProcessor {
.explanation("yarn master requires a deploy mode to be specified as either 'client' or 'cluster'")
.build());
}
+ }
return results;
}
@@ -614,7 +615,7 @@ public class ExecuteSparkJob extends AbstractNiFiProcessor {
}
private OptionalSparkConfigurator setDeployMode(String sparkMaster, String sparkYarnDeployMode) {
- if (sparkMaster.equals("yarn")) {
+ if (sparkMaster.equals("yarn") && StringUtils.isNotEmpty(sparkYarnDeployMode)) {
launcher.setDeployMode(sparkYarnDeployMode);
getLog().info("YARN deploy mode set to: {}", new Object[]{sparkYarnDeployMode});
}
