title: What is reason for getting ProducerFencedException during producer.send?
body: <p>Trying to load around 50K messages into KAFKA topic. In the beginning of few runs getting below exception but not all the time.</p>
  
  <pre><code>org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state  
  at org.apache.kafka.clients.producer.internals.TransactionManager.maybeFailWithError(TransactionManager.java:784) ~[kafka-clients-2.0.0.jar:?]  
  at org.apache.kafka.clients.producer.internals.TransactionManager.beginAbort(TransactionManager.java:229) ~[kafka-clients-2.0.0.jar:?]  
  at  org.apache.kafka.clients.producer.KafkaProducer.abortTransaction(KafkaProducer.java:679) ~[kafka-clients-2.0.0.jar:?]  
  at myPackage.persistUpdatesPostAction(MyCode.java:??) ~[aKafka.jar:?]  
  ...  
  Caused by: org.apache.kafka.common.errors.ProducerFencedException: Producer
  attempted an operation with an old epoch. Either there is a newer producer with
  the same transactionalId, or the producer's transaction has been expired by the
  broker.  
  </code></pre>
  
  <p>Code Block is below:  </p>
  
  <pre><code>public void persistUpdatesPostAction(List&lt;Message&gt; messageList ) {
      if ((messageList == null) || (messageList.isEmpty())) {
          return;
      }
      logger.createDebug("Messages in batch(postAction) : "+ messageList.size());
      Producer&lt;String,String&gt; producer = KafkaUtils.getProducer(Thread.currentThread().getName());
      try {
          producer.beginTransaction();
          createKafkaBulkInsert1(producer, messageList, "Topic1");
          createKafkaBulkInsert2(producer, messageList, "Topic2");
          createKafkaBulkInsert3(producer, messageList, "Topic3");
          producer.commitTransaction();
      } catch (Exception e) {
          producer.abortTransaction();
          producer.close();
          KafkaUtils.removeProducer(Thread.currentThread().getName());
      }
  }
  
  -----------
  
  static Properties setPropertiesProducer() {
      Properties temp = new Properties();
      temp.put("bootstrap.servers", "localhost:9092");
      temp.put("acks", "all");
      temp.put("retries", 1);
      temp.put("batch.size", 16384);
      temp.put("linger.ms", 5);
      temp.put("buffer.memory", 33554432);
      temp.put("key.serializer",   "org.apache.kafka.common.serialization.StringSerializer");
      temp.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
      return temp;
  }
  
  public static Producer&lt;String, String&gt; getProducer(String aThreadId) {
      if ((producerMap.size() == 0) || (producerMap.get(aThreadId) == null)) {
          Properties temp = producerProps;
          temp.put("transactional.id", aThreadId);
          Producer&lt;String, String&gt; producer = new KafkaProducer&lt;String, String&gt;(temp);
          producerMap.put(aThreadId, producer);
          producer.initTransactions();
          return producer;
      }
      return producerMap.get(aThreadId);
  }
  
  public static void removeProducer(String aThreadId) {
      logger.createDebug("Removing Thread ID :" + aThreadId);
      if (producerMap.get(aThreadId) == null)
          return;
      producerMap.remove(aThreadId);
  }
  </code></pre>