title: How to provide remotely located(like hdfs) keystore files to kafak client in a spark direct-stream application
body: <pre><code>public static void main(String[] args){
      Map&lt;String, Object&gt; kafkaParams = new HashMap&lt;String, Object&gt;();
      if(args.length &lt; 2){
          logger.error("Please provide ssl key location and which env to connect to");
      }
      else{
          ReadAndRelay.path = args[0];
          ReadAndRelay.env = args[1];
      }
  
      kafkaParams.put("security.protocol", "SSL");
      try{
          if(ReadAndRelay.env.equals("dev")){
              kafkaParams.put("group.id" , "group_id");
              kafkaParams.put("ssl.keystore.location", ReadAndRelay.path+"/keystore.jks");
              kafkaParams.put("ssl.truststore.location", ReadAndRelay.path+"/truststore.jks");
              kafkaParams.put("bootstrap.servers", "bootstrap_servers");
          }
      }catch(Exception e){
          e.printStackTrace();
      }
  
      kafkaParams.put("ssl.truststore.password", "truststore_password");
      kafkaParams.put("ssl.keystore.password", "keystore_password");
      kafkaParams.put("ssl.key.password", "key_password");
      kafkaParams.put("key.deserializer", StringDeserializer.class);
      kafkaParams.put("value.deserializer", StringDeserializer.class);
      kafkaParams.put("auto.offset.reset", "latest");
      kafkaParams.put("enable.auto.commit", false);
  
      Collection&lt;String&gt; topics = Arrays.asList("topic1","topic2");
  
      SparkConf sparkConf = new SparkConf().setAppName("kafka-stream").setMaster("local[4]");
      JavaStreamingContext streamingContext = new JavaStreamingContext(sparkConf, new Duration(2000));
      final JavaInputDStream&lt;ConsumerRecord&lt;String, String&gt;&gt; stream =
        KafkaUtils.createDirectStream(
          streamingContext,
          LocationStrategies.PreferConsistent(),
          ConsumerStrategies.&lt;String, String&gt;Subscribe(topics, kafkaParams)
        );
  }
  </code></pre>
  
  <p>Above code works with path of keystore files being local. How to have to keystore files in a common location(like HDFS) and use them in the spark application to create a direct-stream(which creates the kafka-consumer) or create a kafak-producer for each of the rdd(because these will be executed at worker nodes / executors)?</p>
  
  <p>When I try to use the hdfs file location as usually in kafka client properties, throws error saying file not found. What is the right way to provide files in hdfs to kafka client properties.</p>
  
  <pre><code>17/03/20 16:18:00 ERROR StreamingContext: Error starting the context, marking it as stopped
  org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
  at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:702)
  at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:557)
  at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:540)
  at org.apache.spark.streaming.kafka010.Subscribe.onStart(ConsumerStrategy.scala:83)
  at org.apache.spark.streaming.kafka010.DirectKafkaInputDStream.consumer(DirectKafkaInputDStream.scala:75)
  at org.apache.spark.streaming.kafka010.DirectKafkaInputDStream.start(DirectKafkaInputDStream.scala:243)
  at org.apache.spark.streaming.DStreamGraph$$anonfun$start$5.apply(DStreamGraph.scala:49)
  at org.apache.spark.streaming.DStreamGraph$$anonfun$start$5.apply(DStreamGraph.scala:49)
  at scala.collection.parallel.mutable.ParArray$ParArrayIterator.foreach_quick(ParArray.scala:143)
  at scala.collection.parallel.mutable.ParArray$ParArrayIterator.foreach(ParArray.scala:136)
  at scala.collection.parallel.ParIterableLike$Foreach.leaf(ParIterableLike.scala:972)
  at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply$mcV$sp(Tasks.scala:49)
  at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply(Tasks.scala:48)
  at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply(Tasks.scala:48)
  at scala.collection.parallel.Task$class.tryLeaf(Tasks.scala:51)
  at scala.collection.parallel.ParIterableLike$Foreach.tryLeaf(ParIterableLike.scala:969)
  at scala.collection.parallel.AdaptiveWorkStealingTasks$WrappedTask$class.compute(Tasks.scala:152)
  at scala.collection.parallel.AdaptiveWorkStealingForkJoinTasks$WrappedTask.compute(Tasks.scala:443)
  at scala.concurrent.forkjoin.RecursiveAction.exec(RecursiveAction.java:160)
  at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
  at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
  at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
  at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
  at ... run in separate thread using org.apache.spark.util.ThreadUtils ... ()
  at org.apache.spark.streaming.StreamingContext.liftedTree1$1(StreamingContext.scala:578)
  at org.apache.spark.streaming.StreamingContext.start(StreamingContext.scala:572)
  at org.apache.spark.streaming.api.java.JavaStreamingContext.start(JavaStreamingContext.scala:556)
  at it.gis.servicemanagement.dcap.dsvs.spark.kafka_stream.ReadAndRelay.main(ReadAndRelay.java:168)
  Caused by: org.apache.kafka.common.KafkaException:         org.apache.kafka.common.KafkaException: java.io.FileNotFoundException: hdfs:/namenode:9000/tmp/kafka_dev_certs/keystore.jks (No such file or directory)
  at org.apache.kafka.common.network.SslChannelBuilder.configure(SslChannelBuilder.java:44)
  at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:70)
  at org.apache.kafka.clients.ClientUtils.createChannelBuilder(ClientUtils.java:83)
  at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:623)
  at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:557)
  at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:540)
  at org.apache.spark.streaming.kafka010.Subscribe.onStart(ConsumerStrategy.scala:83)
  at org.apache.spark.streaming.kafka010.DirectKafkaInputDStream.consumer(DirectKafkaInputDStream.scala:75)
  at org.apache.spark.streaming.kafka010.DirectKafkaInputDStream.start(DirectKafkaInputDStream.scala:243)
  at org.apache.spark.streaming.DStreamGraph$$anonfun$start$5.apply(DStreamGraph.scala:49)
  at org.apache.spark.streaming.DStreamGraph$$anonfun$start$5.apply(DStreamGraph.scala:49)
  at scala.collection.parallel.mutable.ParArray$ParArrayIterator.foreach_quick(ParArray.scala:143)
  at scala.collection.parallel.mutable.ParArray$ParArrayIterator.foreach(ParArray.scala:136)
  at scala.collection.parallel.ParIterableLike$Foreach.leaf(ParIterableLike.scala:972)
  at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply$mcV$sp(Tasks.scala:49)
  at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply(Tasks.scala:48)
  at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply(Tasks.scala:48)
  at scala.collection.parallel.Task$class.tryLeaf(Tasks.scala:51)
  at scala.collection.parallel.ParIterableLike$Foreach.tryLeaf(ParIterableLike.scala:969)
  at scala.collection.parallel.AdaptiveWorkStealingTasks$WrappedTask$class.compute(Tasks.scala:152)
  at scala.collection.parallel.AdaptiveWorkStealingForkJoinTasks$WrappedTask.compute(Tasks.scala:443)
  at scala.concurrent.forkjoin.RecursiveAction.exec(RecursiveAction.java:160)
  at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
  at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
  at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
  at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
  Caused by: org.apache.kafka.common.KafkaException: java.io.FileNotFoundException: hdfs:/namenode:9000/tmp/kafka_dev_certs/keystore.jks (No such file or directory)
  at org.apache.kafka.common.security.ssl.SslFactory.configure(SslFactory.java:110)
  at org.apache.kafka.common.network.SslChannelBuilder.configure(SslChannelBuilder.java:41)
  ... 25 more
  Caused by: java.io.FileNotFoundException: hdfs://namenode:9000/tmp/kafka_dev_certs/keystore.jks (No such file or directory)
  at java.io.FileInputStream.open0(Native Method)
  at java.io.FileInputStream.open(FileInputStream.java:195)
  at java.io.FileInputStream.&lt;init&gt;(FileInputStream.java:138)
  at java.io.FileInputStream.&lt;init&gt;(FileInputStream.java:93)
  at org.apache.kafka.common.security.ssl.SslFactory$SecurityStore.load(SslFactory.java:205)
  at org.apache.kafka.common.security.ssl.SslFactory$SecurityStore.access$000(SslFactory.java:190)
  at org.apache.kafka.common.security.ssl.SslFactory.createSSLContext(SslFactory.java:126)
  at org.apache.kafka.common.security.ssl.SslFactory.configure(SslFactory.java:108)
  ... 26 more
  17/03/20 16:18:00 INFO ReceiverTracker: ReceiverTracker stopped
  17/03/20 16:18:00 INFO JobGenerator: Stopping JobGenerator immediately
  17/03/20 16:18:00 INFO RecurringTimer: Stopped timer for JobGenerator after time -1
  17/03/20 16:18:00 INFO JobGenerator: Stopped JobGenerator
  17/03/20 16:18:00 INFO JobScheduler: Stopped JobScheduler
  </code></pre>