title: KafkaConsumer is not safe for multi-threaded access from SparkStreaming
body: <p>I have set up multiple streams reading from different kafka topics:</p>
  
  <pre><code>for(topic &lt;- topics) {
  
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent, 
      Subscribe[String, String](
        Array(topicConfig.srcTopic),
        kafkaParameters()
      )
    )
    stream.map(...).reduce(...)...
  }
  </code></pre>
  
  <p>kafkaParameters basically has the needed config:</p>
  
  <pre><code>  "bootstrap.servers" -&gt; bootstrapServers,
    "key.deserializer" -&gt; classOf[StringDeserializer],
    "value.deserializer" -&gt; classOf[StringDeserializer],
    "group.id" -&gt; groupId,
    "auto.offset.reset" -&gt; "latest",
    "enable.auto.commit" -&gt; (false: java.lang.Boolean)//,
  </code></pre>
  
  <p>I am using kafka 0.10 integration with spark streaming. The spark configuration is:</p>
  
  <pre><code>val conf = new SparkConf().setMaster("local[*]").setAppName("AppName").set("spark.streaming.concurrentJobs", "4")
  </code></pre>
  
  <p>And getting the exception:</p>
  
  <pre><code>Caused by: java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
  at org.apache.kafka.clients.consumer.KafkaConsumer.acquire(KafkaConsumer.java:1625)
  at org.apache.kafka.clients.consumer.KafkaConsumer.seek(KafkaConsumer.java:1198)
  at org.apache.spark.streaming.kafka010.CachedKafkaConsumer.seek(CachedKafkaConsumer.scala:95)
  at org.apache.spark.streaming.kafka010.CachedKafkaConsumer.get(CachedKafkaConsumer.scala:69)
  at org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.next(KafkaRDD.scala:228)
  at org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.next(KafkaRDD.scala:194)
  at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
  at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
  at scala.collection.Iterator$$anon$10.next(Iterator.scala:393)
  at scala.collection.Iterator$class.foreach(Iterator.scala:893)
  at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
  at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
  at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
  at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
  at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
  at scala.collection.AbstractIterator.to(Iterator.scala:1336)
  at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
  at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1336)
  at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
  at scala.collection.AbstractIterator.toArray(Iterator.scala:1336)
  at org.apache.spark.rdd.RDD$$anonfun$take$1$$anonfun$29.apply(RDD.scala:1353)
  at org.apache.spark.rdd.RDD$$anonfun$take$1$$anonfun$29.apply(RDD.scala:1353)
  at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1944)
  at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1944)
  at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
  at org.apache.spark.scheduler.Task.run(Task.scala:99)
  at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:282)
  ... 3 more
  </code></pre>