Reformatted to Kylo's code style
@@ -406,7 +406,6 @@ public class ExecuteSparkJob extends AbstractNiFiProcessor {
String datasourceIds = context.getProperty(DATASOURCES).evaluateAttributeExpressions(flowFile).getValue();
MetadataProviderService metadataService = context.getProperty(METADATA_SERVICE).asControllerService(MetadataProviderService.class);
-
final List<String> extraJarPaths = getExtraJarPaths(extraJars);
// If all 3 fields are filled out then assume kerberos is enabled, and user should be authenticated
@@ -443,12 +442,13 @@ public class ExecuteSparkJob extends AbstractNiFiProcessor {
return;
}
-
String sparkHome = context.getProperty(SPARK_HOME).evaluateAttributeExpressions(flowFile).getValue();
// Build environment
final Map<String, String> env = getDatasources(session, flowFile, PROVENANCE_JOB_STATUS_KEY, datasourceIds, metadataService, extraJarPaths);
- if (env == null) return;
+ if (env == null) {
+ return;
+ }
/* Launch the spark job as a child process */
SparkLauncher launcher = new SparkLauncher(env)
@@ -515,7 +515,8 @@ public class ExecuteSparkJob extends AbstractNiFiProcessor {
}
}
- private Map<String, String> getDatasources(ProcessSession session, FlowFile flowFile, String PROVENANCE_JOB_STATUS_KEY, String datasourceIds, MetadataProviderService metadataService, List<String> extraJarPaths) throws JsonProcessingException {
+ private Map<String, String> getDatasources(ProcessSession session, FlowFile flowFile, String PROVENANCE_JOB_STATUS_KEY, String datasourceIds, MetadataProviderService metadataService,
+ List<String> extraJarPaths) throws JsonProcessingException {
final Map<String, String> env = new HashMap<>();
if (StringUtils.isNotBlank(datasourceIds)) {
@@ -601,7 +602,9 @@ public class ExecuteSparkJob extends AbstractNiFiProcessor {
}
private class OptionalSparkConfigurator {
+
private SparkLauncher launcher;
+
public OptionalSparkConfigurator(SparkLauncher launcher) {
this.launcher = launcher;
}
@@ -668,6 +671,5 @@ public class ExecuteSparkJob extends AbstractNiFiProcessor {
}
-
}
}
