You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by jo...@apache.org on 2017/12/14 21:24:22 UTC
[2/3] nifi git commit: NIFI-4683: Add ability to execute Spark jobs
via Livy
http://git-wip-us.apache.org/repos/asf/nifi/blob/2192138b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/main/java/org/apache/nifi/processors/livy/ExecuteSparkInteractive.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/main/java/org/apache/nifi/processors/livy/ExecuteSparkInteractive.java b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/main/java/org/apache/nifi/processors/livy/ExecuteSparkInteractive.java
new file mode 100644
index 0000000..d388148
--- /dev/null
+++ b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/main/java/org/apache/nifi/processors/livy/ExecuteSparkInteractive.java
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.livy;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang.StringEscapeUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+
+import org.apache.nifi.processor.util.StandardValidators;
+import org.codehaus.jettison.json.JSONException;
+import org.codehaus.jettison.json.JSONObject;
+
+import org.apache.nifi.controller.api.livy.LivySessionService;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"spark", "livy", "http", "execute"})
+@CapabilityDescription("Execute Spark Code over a Livy-managed HTTP session to a live Spark context. Supports cached RDD sharing.")
+public class ExecuteSparkInteractive extends AbstractProcessor {
+
+ public static final PropertyDescriptor LIVY_CONTROLLER_SERVICE = new PropertyDescriptor.Builder()
+ .name("exec-spark-iactive-livy-controller-service")
+ .displayName("Livy Controller Service")
+ .description("The controller service to use for Livy-managed session(s).")
+ .required(true)
+ .identifiesControllerService(LivySessionService.class)
+ .build();
+
+ public static final PropertyDescriptor CODE = new PropertyDescriptor.Builder()
+ .name("exec-spark-iactive-code")
+ .displayName("Code")
+ .description("The code to execute in the session. This property can be empty, a constant value, or built from attributes "
+ + "using Expression Language. If this property is specified, it will be used regardless of the content of "
+ + "incoming flowfiles. If this property is empty, the content of the incoming flow file is expected "
+ + "to contain valid code to be issued by the processor to the session. Note that Expression "
+ + "Language is not evaluated for flow file contents.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+
+ /**
+ * Points to the charset name corresponding to the incoming flow file's encoding.
+ */
+ public static final PropertyDescriptor CHARSET = new PropertyDescriptor.Builder()
+ .name("exec-spark-iactive-charset")
+ .displayName("Character Set")
+ .description("The character set encoding for the incoming flow file.")
+ .required(true)
+ .defaultValue("UTF-8")
+ .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+
+ public static final PropertyDescriptor STATUS_CHECK_INTERVAL = new PropertyDescriptor.Builder()
+ .name("exec-spark-iactive-status-check-interval")
+ .displayName("Status Check Interval")
+ .description("The amount of time to wait between checking the status of an operation.")
+ .required(true)
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .defaultValue("1 sec")
+ .build();
+
+ public static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("FlowFiles that are successfully processed are sent to this relationship")
+ .build();
+
+ public static final Relationship REL_WAIT = new Relationship.Builder()
+ .name("wait")
+ .description("FlowFiles that are waiting on an available Spark session will be sent to this relationship")
+ .build();
+
+ public static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("FlowFiles are routed to this relationship when they cannot be parsed")
+ .build();
+
+ private volatile List<PropertyDescriptor> properties;
+ private volatile Set<Relationship> relationships;
+
+
+ public void init(final ProcessorInitializationContext context) {
+ List<PropertyDescriptor> properties = new ArrayList<>();
+ properties.add(LIVY_CONTROLLER_SERVICE);
+ properties.add(CODE);
+ properties.add(CHARSET);
+ properties.add(STATUS_CHECK_INTERVAL);
+ this.properties = Collections.unmodifiableList(properties);
+
+ Set<Relationship> relationships = new HashSet<>();
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_WAIT);
+ relationships.add(REL_FAILURE);
+ this.relationships = Collections.unmodifiableSet(relationships);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return properties;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, final ProcessSession session) throws ProcessException {
+
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final ComponentLog log = getLogger();
+ final LivySessionService livySessionService = context.getProperty(LIVY_CONTROLLER_SERVICE).asControllerService(LivySessionService.class);
+ final Map<String, String> livyController = livySessionService.getSession();
+ if (livyController == null || livyController.isEmpty()) {
+ log.debug("No Spark session available (yet), routing flowfile to wait");
+ session.transfer(flowFile, REL_WAIT);
+ return;
+ }
+ final long statusCheckInterval = context.getProperty(STATUS_CHECK_INTERVAL).evaluateAttributeExpressions(flowFile).asTimePeriod(TimeUnit.MILLISECONDS);
+ Charset charset;
+ try {
+ charset = Charset.forName(context.getProperty(CHARSET).evaluateAttributeExpressions(flowFile).getValue());
+ } catch (Exception e) {
+ log.warn("Illegal character set name specified, defaulting to UTF-8");
+ charset = StandardCharsets.UTF_8;
+ }
+
+ String sessionId = livyController.get("sessionId");
+ String livyUrl = livyController.get("livyUrl");
+ String code = context.getProperty(CODE).evaluateAttributeExpressions(flowFile).getValue();
+ if (StringUtils.isEmpty(code)) {
+ try (InputStream inputStream = session.read(flowFile)) {
+ // If no code was provided, assume it is in the content of the incoming flow file
+ code = IOUtils.toString(inputStream, charset);
+ } catch (IOException ioe) {
+ log.error("Error reading input flowfile, penalizing and routing to failure", new Object[]{flowFile, ioe.getMessage()}, ioe);
+ flowFile = session.penalize(flowFile);
+ session.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+ }
+
+ code = StringEscapeUtils.escapeJavaScript(code);
+ String payload = "{\"code\":\"" + code + "\"}";
+ try {
+ final JSONObject result = submitAndHandleJob(livyUrl, livySessionService, sessionId, payload, statusCheckInterval);
+ log.debug("********** ExecuteSparkInteractive Result of Job Submit: " + result);
+ if (result == null) {
+ session.transfer(flowFile, REL_FAILURE);
+ } else {
+ try {
+ final JSONObject output = result.getJSONObject("data");
+ flowFile = session.write(flowFile, out -> out.write(output.toString().getBytes()));
+ flowFile = session.putAttribute(flowFile, CoreAttributes.MIME_TYPE.key(), LivySessionService.APPLICATION_JSON);
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (JSONException je) {
+ // The result doesn't contain the data, just send the output object as the flow file content to failure (after penalizing)
+ log.error("Spark Session returned an error, sending the output JSON object as the flow file content to failure (after penalizing)");
+ flowFile = session.write(flowFile, out -> out.write(result.toString().getBytes()));
+ flowFile = session.putAttribute(flowFile, CoreAttributes.MIME_TYPE.key(), LivySessionService.APPLICATION_JSON);
+ flowFile = session.penalize(flowFile);
+ session.transfer(flowFile, REL_FAILURE);
+ }
+ }
+ } catch (IOException ioe) {
+ log.error("Failure processing flowfile {} due to {}, penalizing and routing to failure", new Object[]{flowFile, ioe.getMessage()}, ioe);
+ flowFile = session.penalize(flowFile);
+ session.transfer(flowFile, REL_FAILURE);
+ }
+ }
+
+ private JSONObject submitAndHandleJob(String livyUrl, LivySessionService livySessionService, String sessionId, String payload, long statusCheckInterval) throws IOException {
+ ComponentLog log = getLogger();
+ String statementUrl = livyUrl + "/sessions/" + sessionId + "/statements";
+ JSONObject output = null;
+ Map<String, String> headers = new HashMap<>();
+ headers.put("Content-Type", LivySessionService.APPLICATION_JSON);
+ headers.put("X-Requested-By", LivySessionService.USER);
+ headers.put("Accept", "application/json");
+
+ log.debug("********** submitAndHandleJob() Submitting Job to Spark via: " + statementUrl);
+ try {
+ JSONObject jobInfo = readJSONObjectFromUrlPOST(statementUrl, livySessionService, headers, payload);
+ log.debug("********** submitAndHandleJob() Job Info: " + jobInfo);
+ String statementId = String.valueOf(jobInfo.getInt("id"));
+ statementUrl = statementUrl + "/" + statementId;
+ jobInfo = readJSONObjectFromUrl(statementUrl, livySessionService, headers);
+ String jobState = jobInfo.getString("state");
+
+ log.debug("********** submitAndHandleJob() New Job Info: " + jobInfo);
+ Thread.sleep(statusCheckInterval);
+ if (jobState.equalsIgnoreCase("available")) {
+ log.debug("********** submitAndHandleJob() Job status is: " + jobState + ". returning output...");
+ output = jobInfo.getJSONObject("output");
+ } else if (jobState.equalsIgnoreCase("running") || jobState.equalsIgnoreCase("waiting")) {
+ while (!jobState.equalsIgnoreCase("available")) {
+ log.debug("********** submitAndHandleJob() Job status is: " + jobState + ". Waiting for job to complete...");
+ Thread.sleep(statusCheckInterval);
+ jobInfo = readJSONObjectFromUrl(statementUrl, livySessionService, headers);
+ jobState = jobInfo.getString("state");
+ }
+ output = jobInfo.getJSONObject("output");
+ } else if (jobState.equalsIgnoreCase("error")
+ || jobState.equalsIgnoreCase("cancelled")
+ || jobState.equalsIgnoreCase("cancelling")) {
+ log.debug("********** Job status is: " + jobState + ". Job did not complete due to error or has been cancelled. Check SparkUI for details.");
+ throw new IOException(jobState);
+ }
+ } catch (JSONException | InterruptedException e) {
+ throw new IOException(e);
+ }
+ return output;
+ }
+
+ private JSONObject readJSONObjectFromUrlPOST(String urlString, LivySessionService livySessionService, Map<String, String> headers, String payload)
+ throws IOException, JSONException {
+
+ HttpURLConnection connection = livySessionService.getConnection(urlString);
+ connection.setRequestMethod("POST");
+ connection.setDoOutput(true);
+
+ for (Map.Entry<String, String> entry : headers.entrySet()) {
+ connection.setRequestProperty(entry.getKey(), entry.getValue());
+ }
+
+ OutputStream os = connection.getOutputStream();
+ os.write(payload.getBytes());
+ os.flush();
+
+ if (connection.getResponseCode() != HttpURLConnection.HTTP_OK && connection.getResponseCode() != HttpURLConnection.HTTP_CREATED) {
+ throw new RuntimeException("Failed : HTTP error code : " + connection.getResponseCode() + " : " + connection.getResponseMessage());
+ }
+
+ InputStream content = connection.getInputStream();
+ BufferedReader rd = new BufferedReader(new InputStreamReader(content, StandardCharsets.UTF_8));
+ String jsonText = IOUtils.toString(rd);
+ return new JSONObject(jsonText);
+ }
+
+ private JSONObject readJSONObjectFromUrl(final String urlString, LivySessionService livySessionService, final Map<String, String> headers)
+ throws IOException, JSONException {
+
+ HttpURLConnection connection = livySessionService.getConnection(urlString);
+ for (Map.Entry<String, String> entry : headers.entrySet()) {
+ connection.setRequestProperty(entry.getKey(), entry.getValue());
+ }
+ connection.setRequestMethod("GET");
+ connection.setDoOutput(true);
+ InputStream content = connection.getInputStream();
+ BufferedReader rd = new BufferedReader(new InputStreamReader(content, StandardCharsets.UTF_8));
+ String jsonText = IOUtils.toString(rd);
+ return new JSONObject(jsonText);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi/blob/2192138b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
new file mode 100644
index 0000000..8de81e4
--- /dev/null
+++ b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+org.apache.nifi.processors.livy.ExecuteSparkInteractive
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi/blob/2192138b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/java/org/apache/nifi/processors/livy/ExecuteSparkInteractiveTestBase.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/java/org/apache/nifi/processors/livy/ExecuteSparkInteractiveTestBase.java b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/java/org/apache/nifi/processors/livy/ExecuteSparkInteractiveTestBase.java
new file mode 100644
index 0000000..3a2c67a
--- /dev/null
+++ b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/java/org/apache/nifi/processors/livy/ExecuteSparkInteractiveTestBase.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.livy;
+
+import org.eclipse.jetty.server.Request;
+import org.eclipse.jetty.server.handler.AbstractHandler;
+
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.io.IOException;
+import java.io.PrintWriter;
+
+public class ExecuteSparkInteractiveTestBase {
+
+ public static class LivyAPIHandler extends AbstractHandler {
+
+ int session1Requests = 0;
+
+ @Override
+ public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException {
+ baseRequest.setHandled(true);
+
+ response.setStatus(200);
+
+ if ("GET".equalsIgnoreCase(request.getMethod())) {
+
+ String responseBody = "{}";
+ response.setContentType("application/json");
+
+ if ("/sessions".equalsIgnoreCase(target)) {
+ responseBody = "{\"sessions\": [{\"id\": 1, \"kind\": \"spark\", \"state\": \"idle\"}]}";
+ } else if (target.startsWith("/sessions/") && !target.contains("statement")) {
+ responseBody = "{\"id\": 1, \"kind\": \"spark\", \"state\": \"idle\"}";
+
+ } else if ("/sessions/1/statements/7".equalsIgnoreCase(target)) {
+ switch (session1Requests) {
+ case 0:
+ responseBody = "{\"state\": \"waiting\"}";
+ break;
+ case 1:
+ responseBody = "{\"state\": \"running\"}";
+ break;
+ case 2:
+ responseBody = "{\"state\": \"available\", \"output\": {\"data\": {\"text/plain\": \"Hello world\"}}}";
+ break;
+ default:
+ responseBody = "{\"state\": \"error\"}";
+ break;
+ }
+ session1Requests++;
+ }
+
+ response.setContentLength(responseBody.length());
+
+ try (PrintWriter writer = response.getWriter()) {
+ writer.print(responseBody);
+ writer.flush();
+ }
+
+ } else if ("POST".equalsIgnoreCase(request.getMethod())) {
+
+ String responseBody = "{}";
+ response.setContentType("application/json");
+
+ if ("/sessions".equalsIgnoreCase(target)) {
+ responseBody = "{\"id\": 1, \"kind\": \"spark\", \"state\": \"idle\"}";
+ } else if ("/sessions/1/statements".equalsIgnoreCase(target)) {
+ responseBody = "{\"id\": 7}";
+ }
+
+ response.setContentLength(responseBody.length());
+
+ try (PrintWriter writer = response.getWriter()) {
+ writer.print(responseBody);
+ writer.flush();
+ }
+
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/2192138b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/java/org/apache/nifi/processors/livy/TestExecuteSparkInteractive.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/java/org/apache/nifi/processors/livy/TestExecuteSparkInteractive.java b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/java/org/apache/nifi/processors/livy/TestExecuteSparkInteractive.java
new file mode 100644
index 0000000..4219783
--- /dev/null
+++ b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/java/org/apache/nifi/processors/livy/TestExecuteSparkInteractive.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.livy;
+
+import org.apache.nifi.controller.livy.LivySessionController;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.eclipse.jetty.server.Handler;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.List;
+
+public class TestExecuteSparkInteractive extends ExecuteSparkInteractiveTestBase {
+
+ public static TestServer server;
+ public static String url;
+
+ public TestRunner runner;
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ // useful for verbose logging output
+ // don't commit this with this property enabled, or any 'mvn test' will be really verbose
+ // System.setProperty("org.slf4j.simpleLogger.log.nifi.processors.standard", "debug");
+
+ // create a Jetty server on a random port
+ server = createServer();
+ server.startServer();
+
+ // this is the base url with the random port
+ url = server.getUrl();
+ }
+
+ public void addHandler(Handler handler) {
+ server.addHandler(handler);
+ }
+
+ @AfterClass
+ public static void afterClass() throws Exception {
+ server.shutdownServer();
+ }
+
+ @Before
+ public void before() throws Exception {
+ runner = TestRunners.newTestRunner(ExecuteSparkInteractive.class);
+ LivySessionController livyControllerService = new LivySessionController();
+ runner.addControllerService("livyCS", livyControllerService);
+ runner.setProperty(livyControllerService, LivySessionController.LIVY_HOST, url.substring(url.indexOf("://") + 3, url.lastIndexOf(":")));
+ runner.setProperty(livyControllerService, LivySessionController.LIVY_PORT, url.substring(url.lastIndexOf(":") + 1));
+ runner.enableControllerService(livyControllerService);
+ runner.setProperty(ExecuteSparkInteractive.LIVY_CONTROLLER_SERVICE, "livyCS");
+
+ server.clearHandlers();
+ }
+
+ @After
+ public void after() {
+ runner.shutdown();
+ }
+
+ private static TestServer createServer() throws IOException {
+ return new TestServer();
+ }
+
+ @Test
+ public void testSparkSession() throws Exception {
+
+ addHandler(new LivyAPIHandler());
+
+ runner.enqueue("print \"hello world\"");
+ runner.run();
+ List<MockFlowFile> waitingFlowfiles = runner.getFlowFilesForRelationship(ExecuteSparkInteractive.REL_WAIT);
+ while (!waitingFlowfiles.isEmpty()) {
+ Thread.sleep(1000);
+ runner.clearTransferState();
+ runner.enqueue("print \"hello world\"");
+ runner.run();
+ waitingFlowfiles = runner.getFlowFilesForRelationship(ExecuteSparkInteractive.REL_WAIT);
+ }
+ runner.assertTransferCount(ExecuteSparkInteractive.REL_SUCCESS, 1);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/2192138b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/java/org/apache/nifi/processors/livy/TestExecuteSparkInteractiveSSL.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/java/org/apache/nifi/processors/livy/TestExecuteSparkInteractiveSSL.java b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/java/org/apache/nifi/processors/livy/TestExecuteSparkInteractiveSSL.java
new file mode 100644
index 0000000..e732617
--- /dev/null
+++ b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/java/org/apache/nifi/processors/livy/TestExecuteSparkInteractiveSSL.java
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.livy;
+
+import org.apache.nifi.controller.livy.LivySessionController;
+import org.apache.nifi.ssl.StandardSSLContextService;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.eclipse.jetty.server.Handler;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class TestExecuteSparkInteractiveSSL extends ExecuteSparkInteractiveTestBase {
+
+ private static Map<String, String> sslProperties;
+
+ public static TestServer server;
+ public static String url;
+
+ public TestRunner runner;
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ // useful for verbose logging output
+ // don't commit this with this property enabled, or any 'mvn test' will be really verbose
+ // System.setProperty("org.slf4j.simpleLogger.log.nifi.processors.standard", "debug");
+
+ // create the SSL properties, which basically store keystore / trustore information
+ // this is used by the StandardSSLContextService and the Jetty Server
+ sslProperties = createSslProperties();
+
+ // create a Jetty server on a random port
+ server = createServer();
+ server.startServer();
+
+ // Allow time for the server to start
+ Thread.sleep(1000);
+
+ // this is the base url with the random port
+ url = server.getSecureUrl();
+ }
+
+ public void addHandler(Handler handler) {
+ server.addHandler(handler);
+ }
+
+ @AfterClass
+ public static void afterClass() throws Exception {
+ server.shutdownServer();
+ }
+
+ @Before
+ public void before() throws Exception {
+ runner = TestRunners.newTestRunner(ExecuteSparkInteractive.class);
+
+ final StandardSSLContextService sslService = new StandardSSLContextService();
+ runner.addControllerService("ssl-context", sslService, sslProperties);
+ runner.enableControllerService(sslService);
+
+ // Allow time for the controller service to fully initialize
+ Thread.sleep(500);
+
+ LivySessionController livyControllerService = new LivySessionController();
+ runner.addControllerService("livyCS", livyControllerService);
+ runner.setProperty(livyControllerService, LivySessionController.LIVY_HOST, url.substring(url.indexOf("://") + 3, url.lastIndexOf(":")));
+ runner.setProperty(livyControllerService, LivySessionController.LIVY_PORT, url.substring(url.lastIndexOf(":") + 1));
+ runner.setProperty(livyControllerService, LivySessionController.SSL_CONTEXT_SERVICE, "ssl-context");
+ runner.enableControllerService(livyControllerService);
+
+ runner.setProperty(ExecuteSparkInteractive.LIVY_CONTROLLER_SERVICE, "livyCS");
+
+ server.clearHandlers();
+ }
+
+ @After
+ public void after() {
+ runner.shutdown();
+ }
+
+ private static TestServer createServer() throws IOException {
+ return new TestServer(sslProperties);
+ }
+
+ @Test
+ public void testSslSparkSession() throws Exception {
+ addHandler(new LivyAPIHandler());
+
+ runner.enqueue("print \"hello world\"");
+ runner.run();
+ List<MockFlowFile> waitingFlowfiles = runner.getFlowFilesForRelationship(ExecuteSparkInteractive.REL_WAIT);
+ while (!waitingFlowfiles.isEmpty()) {
+ Thread.sleep(1000);
+ runner.clearTransferState();
+ runner.enqueue("print \"hello world\"");
+ runner.run();
+ waitingFlowfiles = runner.getFlowFilesForRelationship(ExecuteSparkInteractive.REL_WAIT);
+ }
+ runner.assertTransferCount(ExecuteSparkInteractive.REL_SUCCESS, 1);
+ }
+
+ private static Map<String, String> createSslProperties() {
+ final Map<String, String> map = new HashMap<>();
+ map.put(StandardSSLContextService.KEYSTORE.getName(), "src/test/resources/localhost-ks.jks");
+ map.put(StandardSSLContextService.KEYSTORE_PASSWORD.getName(), "localtest");
+ map.put(StandardSSLContextService.KEYSTORE_TYPE.getName(), "JKS");
+ map.put(StandardSSLContextService.TRUSTSTORE.getName(), "src/test/resources/localhost-ts.jks");
+ map.put(StandardSSLContextService.TRUSTSTORE_PASSWORD.getName(), "localtest");
+ map.put(StandardSSLContextService.TRUSTSTORE_TYPE.getName(), "JKS");
+ return map;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/2192138b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/java/org/apache/nifi/processors/livy/TestServer.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/java/org/apache/nifi/processors/livy/TestServer.java b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/java/org/apache/nifi/processors/livy/TestServer.java
new file mode 100644
index 0000000..396df07
--- /dev/null
+++ b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/java/org/apache/nifi/processors/livy/TestServer.java
@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.livy;
+
+import org.apache.nifi.ssl.StandardSSLContextService;
+import org.eclipse.jetty.server.Handler;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.ServerConnector;
+import org.eclipse.jetty.server.handler.HandlerCollection;
+import org.eclipse.jetty.util.ssl.SslContextFactory;
+
+import java.util.Map;
+
+/**
+ * Test server to assist with unit tests that requires a server to be stood up.
+ */
+/**
+ * Test server to assist with unit tests that requires a server to be stood up.
+ */
+public class TestServer {
+
+ public static final String NEED_CLIENT_AUTH = "clientAuth";
+
+ private Server jetty;
+ private boolean secure = false;
+
+ /**
+ * Creates the test server.
+ */
+ public TestServer() {
+ createServer(null);
+ }
+
+ /**
+ * Creates the test server.
+ *
+ * @param sslProperties SSLProps to be used in the secure connection. The keys should should use the StandardSSLContextService properties.
+ */
+ public TestServer(final Map<String, String> sslProperties) {
+ createServer(sslProperties);
+ }
+
+ private void createServer(final Map<String, String> sslProperties) {
+ jetty = new Server();
+
+ // create the unsecure connector
+ createConnector();
+
+ // create the secure connector if sslProperties are specified
+ if (sslProperties != null) {
+ createSecureConnector(sslProperties);
+ }
+
+ jetty.setHandler(new HandlerCollection(true));
+ }
+
+ /**
+ * Creates the http connection
+ */
+ private void createConnector() {
+ final ServerConnector http = new ServerConnector(jetty);
+ http.setPort(0);
+ // Severely taxed environments may have significant delays when executing.
+ http.setIdleTimeout(30000L);
+ jetty.addConnector(http);
+ }
+
+ private void createSecureConnector(final Map<String, String> sslProperties) {
+ SslContextFactory ssl = new SslContextFactory();
+
+ if (sslProperties.get(StandardSSLContextService.KEYSTORE.getName()) != null) {
+ ssl.setKeyStorePath(sslProperties.get(StandardSSLContextService.KEYSTORE.getName()));
+ ssl.setKeyStorePassword(sslProperties.get(StandardSSLContextService.KEYSTORE_PASSWORD.getName()));
+ ssl.setKeyStoreType(sslProperties.get(StandardSSLContextService.KEYSTORE_TYPE.getName()));
+ }
+
+ if (sslProperties.get(StandardSSLContextService.TRUSTSTORE.getName()) != null) {
+ ssl.setTrustStorePath(sslProperties.get(StandardSSLContextService.TRUSTSTORE.getName()));
+ ssl.setTrustStorePassword(sslProperties.get(StandardSSLContextService.TRUSTSTORE_PASSWORD.getName()));
+ ssl.setTrustStoreType(sslProperties.get(StandardSSLContextService.TRUSTSTORE_TYPE.getName()));
+ }
+
+ final String clientAuth = sslProperties.get(NEED_CLIENT_AUTH);
+ if (clientAuth == null) {
+ ssl.setNeedClientAuth(true);
+ } else {
+ ssl.setNeedClientAuth(Boolean.parseBoolean(clientAuth));
+ }
+
+ // build the connector
+ final ServerConnector https = new ServerConnector(jetty, ssl);
+
+ // set host and port
+ https.setPort(0);
+ // Severely taxed environments may have significant delays when executing.
+ https.setIdleTimeout(30000L);
+
+ // add the connector
+ jetty.addConnector(https);
+
+ // mark secure as enabled
+ secure = true;
+ }
+
+ public void clearHandlers() {
+ HandlerCollection hc = (HandlerCollection) jetty.getHandler();
+ Handler[] ha = hc.getHandlers();
+ if (ha != null) {
+ for (Handler h : ha) {
+ hc.removeHandler(h);
+ }
+ }
+ }
+
+ public void addHandler(Handler handler) {
+ ((HandlerCollection) jetty.getHandler()).addHandler(handler);
+ }
+
+ public void startServer() throws Exception {
+ jetty.start();
+ }
+
+ public void shutdownServer() throws Exception {
+ jetty.stop();
+ jetty.destroy();
+ }
+
+ private int getPort() {
+ if (!jetty.isStarted()) {
+ throw new IllegalStateException("Jetty server not started");
+ }
+ return ((ServerConnector) jetty.getConnectors()[0]).getLocalPort();
+ }
+
+ private int getSecurePort() {
+ if (!jetty.isStarted()) {
+ throw new IllegalStateException("Jetty server not started");
+ }
+ return ((ServerConnector) jetty.getConnectors()[1]).getLocalPort();
+ }
+
+ public String getUrl() {
+ return "http://localhost:" + getPort();
+ }
+
+ public String getSecureUrl() {
+ String url = null;
+ if (secure) {
+ url = "https://localhost:" + getSecurePort();
+ }
+ return url;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/2192138b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/resources/localhost-ks.jks
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/resources/localhost-ks.jks b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/resources/localhost-ks.jks
new file mode 100755
index 0000000..df36197
Binary files /dev/null and b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/resources/localhost-ks.jks differ
http://git-wip-us.apache.org/repos/asf/nifi/blob/2192138b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/resources/localhost-ts.jks
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/resources/localhost-ts.jks b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/resources/localhost-ts.jks
new file mode 100755
index 0000000..7824378
Binary files /dev/null and b/nifi-nar-bundles/nifi-spark-bundle/nifi-livy-processors/src/test/resources/localhost-ts.jks differ
http://git-wip-us.apache.org/repos/asf/nifi/blob/2192138b/nifi-nar-bundles/nifi-spark-bundle/pom.xml
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-spark-bundle/pom.xml b/nifi-nar-bundles/nifi-spark-bundle/pom.xml
new file mode 100644
index 0000000..91b0ce8
--- /dev/null
+++ b/nifi-nar-bundles/nifi-spark-bundle/pom.xml
@@ -0,0 +1,85 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+ http://www.apache.org/licenses/LICENSE-2.0
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-nar-bundles</artifactId>
+ <version>1.5.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-spark-bundle</artifactId>
+ <packaging>pom</packaging>
+
+ <modules>
+ <module>nifi-livy-nar</module>
+ <module>nifi-livy-controller-service-api-nar</module>
+ <module>nifi-livy-controller-service-api</module>
+ <module>nifi-livy-controller-service</module>
+ <module>nifi-livy-processors</module>
+ </modules>
+
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>org.codehaus.jackson</groupId>
+ <artifactId>jackson-core-asl</artifactId>
+ <version>1.9.13</version>
+ </dependency>
+ <dependency>
+ <groupId>org.codehaus.jackson</groupId>
+ <artifactId>jackson-mapper-asl</artifactId>
+ <version>1.9.13</version>
+ </dependency>
+ <dependency>
+ <groupId>org.json</groupId>
+ <artifactId>json</artifactId>
+ <version>20160810</version>
+ </dependency>
+ <dependency>
+ <groupId>org.codehaus.jettison</groupId>
+ <artifactId>jettison</artifactId>
+ <version>1.3.8</version>
+ </dependency>
+ <dependency>
+ <groupId>commons-configuration</groupId>
+ <artifactId>commons-configuration</artifactId>
+ <version>1.10</version>
+ </dependency>
+ <dependency>
+ <groupId>com.sun.jersey</groupId>
+ <artifactId>jersey-client</artifactId>
+ <version>1.19.1</version>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-utils</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-processor-utils</artifactId>
+ </dependency>
+
+ </dependencies>
+</project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi/blob/2192138b/nifi-nar-bundles/pom.xml
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/pom.xml b/nifi-nar-bundles/pom.xml
index 0462e72..f462883 100755
--- a/nifi-nar-bundles/pom.xml
+++ b/nifi-nar-bundles/pom.xml
@@ -89,6 +89,7 @@
<module>nifi-extension-utils</module>
<module>nifi-redis-bundle</module>
<module>nifi-metrics-reporting-bundle</module>
+ <module>nifi-spark-bundle</module>
</modules>
<build>
http://git-wip-us.apache.org/repos/asf/nifi/blob/2192138b/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 11caa40..3f573b6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1462,7 +1462,19 @@
<version>1.5.0-SNAPSHOT</version>
<type>nar</type>
</dependency>
- <dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-livy-controller-service-api-nar</artifactId>
+ <version>1.5.0-SNAPSHOT</version>
+ <type>nar</type>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-livy-nar</artifactId>
+ <version>1.5.0-SNAPSHOT</version>
+ <type>nar</type>
+ </dependency>
+ <dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-properties</artifactId>
<version>1.5.0-SNAPSHOT</version>