title: Spark Java: java.util.ConcurrentModificationException while broadcasting object of type GenericObjectPool
body: <p>I am working on a spark-streaming project in <strong>java</strong>. I am trying to send some messages from spark to apache <strong>kafka</strong> using kafka-producer java api. Since creating instance of KafkaProducer for each element would be very expensive, I am trying to use a pool of producer using apache common pooling framework. As shown in the code snippet below, I am creating the GenericObjectPool instance and broadcasting it as shown below:- </p>
  
  <pre><code>GenericObjectPool&lt;KafkaProducer&lt;String, String&gt;&gt; producerPool = new GenericObjectPool&lt;KafkaProducer&lt;String, String&gt;&gt;(
                  new KafkaProducerFactory(prop));
  final Broadcast&lt;GenericObjectPool&lt;KafkaProducer&lt;String, String&gt;&gt;&gt; pool = ssc.sparkContext() .broadcast(producerPool);  //**Causing exception**
  </code></pre>
  
  <p><strong>KafkaProducerFactory class's code is pasted below:-</strong></p>
  
  <pre><code>import java.io.Serializable;
  import java.util.Map;
  
  import org.apache.commons.pool2.BasePooledObjectFactory;
  import org.apache.commons.pool2.PooledObject;
  import org.apache.commons.pool2.impl.DefaultPooledObject;
  import org.apache.kafka.clients.producer.KafkaProducer;
  
  public class KafkaProducerFactory&lt;K,V&gt; extends BasePooledObjectFactory&lt;KafkaProducer&lt;K, V&gt;&gt; 
  implements Serializable{
      private Map&lt;String,Object&gt; configs;
      public KafkaProducerFactory(Map&lt;String, Object&gt; configs) {
          this.configs=configs;
      }
  
      @Override
      public KafkaProducer&lt;K, V&gt; create() {
          return new KafkaProducer&lt;K, V&gt;(this.configs);
      }
  
      @Override
      public PooledObject&lt;KafkaProducer&lt;K,V&gt;&gt; wrap(KafkaProducer&lt;K,V&gt; producer) {
          return new DefaultPooledObject&lt;KafkaProducer&lt;K,V&gt;&gt;(producer);
      }
  
      @Override
      public void destroyObject(PooledObject&lt;KafkaProducer&lt;K,V&gt;&gt;obj){
          obj.getObject().close();
      }
  }
  </code></pre>
  
  <p>But above line is giving me the below pasted exception:- </p>
  
  <blockquote>
    <p>com.esotericsoftware.kryo.KryoException: java.util.ConcurrentModificationException</p>
  </blockquote>
  
  <p><strong>Full StackTrace is pasted below:-</strong></p>
  
  <pre><code>Exception in thread "main" com.esotericsoftware.kryo.KryoException: java.util.ConcurrentModificationException
  Serialization trace:
  classes (sun.misc.Launcher$AppClassLoader)
  classloader (java.security.ProtectionDomain)
  context (java.security.AccessControlContext)
  acc (org.apache.spark.util.MutableURLClassLoader)
  referent (java.lang.ref.WeakReference)
  factoryClassLoader (org.apache.commons.pool2.impl.GenericObjectPool)
      at com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.write(FieldSerializer.java:585)
      at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:213)
      at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:501)
      at com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.write(FieldSerializer.java:564)
      at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:213)
      at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:568)
      at com.esotericsoftware.kryo.serializers.DefaultArraySerializers$ObjectArraySerializer.write(DefaultArraySerializers.java:318)
      at com.esotericsoftware.kryo.serializers.DefaultArraySerializers$ObjectArraySerializer.write(DefaultArraySerializers.java:293)
      at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:501)
      at com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.write(FieldSerializer.java:564)
      at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:213)
      at com.esotericsoftware.kryo.Kryo.writeObjectOrNull(Kryo.java:549)
      at com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.write(FieldSerializer.java:570)
      at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:213)
      at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:501)
      at com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.write(FieldSerializer.java:564)
      at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:213)
      at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:501)
      at com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.write(FieldSerializer.java:564)
      at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:213)
      at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:568)
      at org.apache.spark.serializer.KryoSerializationStream.writeObject(KryoSerializer.scala:148)
      at org.apache.spark.broadcast.TorrentBroadcast$.blockifyObject(TorrentBroadcast.scala:203)
      at org.apache.spark.broadcast.TorrentBroadcast.writeBlocks(TorrentBroadcast.scala:102)
      at org.apache.spark.broadcast.TorrentBroadcast.&lt;init&gt;(TorrentBroadcast.scala:85)
      at org.apache.spark.broadcast.TorrentBroadcastFactory.newBroadcast(TorrentBroadcastFactory.scala:34)
      at org.apache.spark.broadcast.BroadcastManager.newBroadcast(BroadcastManager.scala:62)
      at org.apache.spark.SparkContext.broadcast(SparkContext.scala:1291)
      at org.apache.spark.api.java.JavaSparkContext.broadcast(JavaSparkContext.scala:648)
      at com.veda.txt.spark.Engine.start(Engine.java:63)
      at com.veda.txt.spark.Engine.main(Engine.java:126)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      at java.lang.reflect.Method.invoke(Method.java:622)
      at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:665)
      at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:170)
      at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:193)
      at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:112)
      at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
  Caused by: java.util.ConcurrentModificationException
      at java.util.Vector$Itr.checkForComodification(Vector.java:1127)
      at java.util.Vector$Itr.next(Vector.java:1104)
      at com.esotericsoftware.kryo.serializers.CollectionSerializer.write(CollectionSerializer.java:74)
      at com.esotericsoftware.kryo.serializers.CollectionSerializer.write(CollectionSerializer.java:18)
      at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:501)
      at com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.write(FieldSerializer.java:564)
      ... 39 more
  15/08/26 20:38:14 INFO SparkContext: Invoking stop() from shutdown hook
  </code></pre>
  
  <p>Please suggest me what went wrong.</p>
  
  <p>Thaks</p>