Using Nifi in a secured cluster

Question

I am trying to use spark streaming to read data from Nifi in a secured cluster. I use SSLContext to authenticate by adding SSLContext in the SiteToSiteClient, but SSLContext is not serializable.
My codes are shown below:

      def main(args: Array[String]) {
        val pKeyFile = new java.io.File("/path/to/file-cert.pfx")
        val pKeyPassword = "password"
        val keyStore = java.security.KeyStore.getInstance("JKS")

        val kmf = javax.net.ssl.KeyManagerFactory.getInstance(javax.net.ssl.KeyManagerFactory.getDefaultAlgorithm())
        val keyInput = new java.io.FileInputStream(pKeyFile)
        keyStore.load(keyInput, pKeyPassword.toCharArray())
        keyInput.close()
        kmf.init(keyStore, pKeyPassword.toCharArray())

        val sslContext = javax.net.ssl.SSLContext.getInstance("SSL")
        sslContext.init(kmf.getKeyManagers(), null, new java.security.SecureRandom())

        val conf = new SiteToSiteClient
          .Builder()
          .sslContext(sslContext)
          .url("https://urlOfNifi:9090/nifi/")
          .portName("Spark_Test")
          .buildConfig()


        val config = new SparkConf().setAppName("Nifi_Spark_Data")
        val sc = new SparkContext(config)   
        val ssc = new StreamingContext(sc, Seconds(10))

        val lines = ssc.receiverStream(new NiFiReceiver(conf, StorageLevel.MEMORY_ONLY))

        val text = lines.map(dataPacket => new String(dataPacket.getContent, StandardCharsets.UTF_8))

        text.print()
        ssc.start()
        ssc.awaitTermination()  
   }
}

What I am trying to do is to get stream data from nifi but when I start my spark streaming application I got error below:

 Exception during serialization: java.io.NotSerializableException: javax.net.ssl.SSLContext
Serialization stack:
        - object not serializable (class: javax.net.ssl.SSLContext, value: javax.net.ssl.SSLContext@2181e104)
        - field (class: org.apache.nifi.remote.client.SiteToSiteClient$StandardSiteToSiteClientConfig, name: sslContext, type: class javax.net.ssl.SSLContext)
        - object (class org.apache.nifi.remote.client.SiteToSiteClient$StandardSiteToSiteClientConfig, org.apache.nifi.remote.client.SiteToSiteClient$StandardSiteToSiteClientConfig@5a0d6057)
        - field (class: org.apache.nifi.spark.NiFiReceiver, name: clientConfig, type: interface org.apache.nifi.remote.client.SiteToSiteClientConfig)
        - object (class org.apache.nifi.spark.NiFiReceiver, org.apache.nifi.spark.NiFiReceiver@224fb09a)
        - element of array (index: 0)
        - array (class [Lorg.apache.spark.streaming.receiver.Receiver;, size 1)
        - field (class: scala.collection.mutable.WrappedArray$ofRef, name: array, type: class [Ljava.lang.Object;)
        - object (class scala.collection.mutable.WrappedArray$ofRef, WrappedArray(org.apache.nifi.spark.NiFiReceiver@224fb09a))
        - writeObject data (class: org.apache.spark.rdd.ParallelCollectionPartition)
        - object (class org.apache.spark.rdd.ParallelCollectionPartition, org.apache.spark.rdd.ParallelCollectionPartition@87d)
        - field (class: org.apache.spark.scheduler.ResultTask, name: partition, type: interface org.apache.spark.Partition)
        - object (class org.apache.spark.scheduler.ResultTask, ResultTask(12, 0))
        at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1283)
        at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1271)
        at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1270)
        at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
        at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
        at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1270)
        at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:697)
        at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:697)
        at scala.Option.foreach(Option.scala:236)
        at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:697)
        at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1496)
        at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1458)
        at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1447)

It seems that the SiteToSiteClientCofig can be serialized while SSLContext inside it is not. In sparkStreaming, objects that are going to be used in other nodes should be serializable but I cannot find a way to make SSLContext serializable. Is there any way to run the spark streaming to receive Nifi stream data in a secure cluster?

thank in advance.

Answer 1

Rather than creating the SSLContext ahead of time, you should be able to call the following methods on SiteToSiteClient.Builder:

keystoreFilename(...)
keystorePass(...)
keystoreType(...)
truststoreFilename(...)
truststorePass(...)
truststoreType(...)

By doing that the NiFiReceiver will create an SSLContext when it builds the SiteToSiteClient from the SiteToSiteClientConfig, which is after serialization.

Note, this would probably require that the keystore/truststore was put on all the nodes where your spark streaming job was running, and in the same location on each node.