title: Kafkaconsumer is not safe for multi-threading access
body: <p>I am using below code to read from Kafka topic , and process the data.</p>
  
  <pre><code>JavaDStream&lt;Row&gt; transformedMessages = messages.flatMap(record -&gt; processData(record))
                  .transform(new Function&lt;JavaRDD&lt;Row&gt;, JavaRDD&lt;Row&gt;&gt;() {
                      //JavaRDD&lt;Row&gt; records = ss.emptyDataFrame().toJavaRDD();
                      StructType schema = DataTypes.createStructType(fields);
  
                      public JavaRDD&lt;Row&gt; call(JavaRDD&lt;Row&gt; rdd) throws Exception {
                          records = rdd.union(records);
                          return rdd;
                      }
          });
  
         transformedMessages.foreachRDD(record -&gt; {
              //System.out.println("Aman" +record.count());
              StructType schema = DataTypes.createStructType(fields);
  
              Dataset ds = ss.createDataFrame(records, schema);
              ds.createOrReplaceTempView("trades");
              System.out.println(ds.count());
              ds.show();
  
          });
  </code></pre>
  
  <p>While running the code, i am getting below 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:1624)
      at org.apache.kafka.clients.consumer.KafkaConsumer.seek(KafkaConsumer.java:1197)
      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$12.nextCur(Iterator.scala:434)
      at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:440)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
      at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.agg_doAggregateWithoutKey$(Unknown Source)
      at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
      at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
      at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:377)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
      at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:126)
      at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
      at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
      at org.apache.spark.scheduler.Task.run(Task.scala:99)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:322)
  </code></pre>
  
  <p>The fact that i only have one DStream, i am not sure why i am getting this exception.
  I am reading from 3 partitions in a Kafka topic. I assume that the "createDirectStream" will be creating 3 consumers to read the data.</p>
  
  <p>Below is the code for for KafkaConsumer, acquire method:</p>
  
  <pre><code> private void acquire() {
          this.ensureNotClosed();
          long threadId = Thread.currentThread().getId();
          if(threadId != this.currentThread.get() &amp;&amp; !this.currentThread.compareAndSet(-1L, threadId)) {
              throw new ConcurrentModificationException("KafkaConsumer is not safe for multi-threaded access");
          } else {
              this.refcount.incrementAndGet();
          }
      }
  </code></pre>