title: Kafka transactional Producer and consumer
body: <p>I'm working on kafka use case where I need to have transactional semantics on producer &amp; consumer side..I'm able to publish transactional messages using kafka transaction API 0.11 to the kafka cluster but on consumer side I'm facing the issue...I have set <code>isolation.level=read_committed</code> in properties file But I'm not able to consume it..I could see messages being consumed with <code>isolation.level=read_uncommitted</code> but this is not desired..</p>
  
  <p>producer code</p>
  
  <pre class="lang-java prettyprint-override"><code>package com.org.kafkaPro;
  
  import java.io.File;
  import java.io.FileNotFoundException;
  import java.io.FileReader;
  import java.io.IOException;
  import java.net.URL;
  import java.text.ParseException;
  import java.text.SimpleDateFormat;
  import java.util.Date;
  import java.util.Properties;
  
  import org.apache.kafka.clients.producer.KafkaProducer;
  import org.apache.kafka.clients.producer.ProducerRecord;
  import org.apache.kafka.common.errors.OutOfOrderSequenceException;
  import org.apache.kafka.common.errors.ProducerFencedException;
  
  import kafka.common.AuthorizationException;
  import kafka.common.KafkaException;
  
      public class ProducerWithTx
      {
  
  
          public static void main(String args[]) throws FileNotFoundException {
              URL in = ProducerWithTx.class.getResource("producertx.properties");
  
              Properties props = new Properties();
  
              try {
                  props.load(new FileReader(new File(in.getFile())));
              } catch (IOException e1) {
                  // TODO Auto-generated catch block
                  e1.printStackTrace();
              }
  
  
              Paymnt pay1= new Paymnt();
              pay1.setAccid(1);
              pay1.setAccountstate("y");
              pay1.setAccountzipcode(111);
              pay1.setBankid(12);
              pay1.setCreditcardtype(15);
              pay1.setCustomerid("2");
              SimpleDateFormat ft = new SimpleDateFormat ("yyyy-MM-dd");
              Date t = null;
              try {
                  t = ft.parse("2017-11-10");
              } catch (ParseException e) {
                  // TODO Auto-generated catch block
                  e.printStackTrace();
              }
              pay1.setPeriodid(t);
  
              String timeStamp = new SimpleDateFormat("yyyy.MM.dd:HH:mm:ss").format(new Date());
              props.put("transactional.id", "accid=" + pay1.getAccid() +  " custid=" +pay1.getCustomerid()+  " timestmp=" +timeStamp);
              KafkaProducer&lt;String, Paymnt&gt; producer = new KafkaProducer(props);
              producer.initTransactions();  
              try{
                  producer.beginTransaction();
  
  
  
                  //RecordMetadata metadata=producer.send((ProducerRecord&lt;String, Paymnt&gt;) new ProducerRecord&lt;String, Paymnt&gt;("test",pay1)).get();
  
                  producer.send((ProducerRecord&lt;String, Paymnt&gt;) new ProducerRecord&lt;String, Paymnt&gt;("test",pay1));
                  producer.commitTransaction();
  
                  //System.out.println("written to" +metadata.partition());
  
              }
              catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException  e){
  
                  // We can't recover from these exceptions, so our only option is to close the producer and exit.
                  producer.close();
              }
              catch(KafkaException e) {
                  // For all other exceptions, just abort the transaction and try again.
                  producer.abortTransaction();
              }
              producer.close();
          }
  
      }
  </code></pre>
  
  <p>producertx.properties</p>
  
  <pre><code>metadata.broker.list=localhost:9092
  bootstrap.servers=localhost:9092
  acks=all
  retries=1
  batch.size=16384
  linger.ms=1
  buffer.memory=33554432
  key.serializer=org.apache.kafka.common.serialization.StringSerializer
  value.serializer=com.org.kafkaPro.PaySerializer
  #transactional.id=1
  enable.idempotence=true
  num.partitions=3
  </code></pre>
  
  <p>Consumer </p>
  
  <pre class="lang-java prettyprint-override"><code>package com.org.kafkaPro;
  
  import java.io.File;
  import java.io.FileReader;
  import java.io.IOException;
  import java.net.URL;
  import java.util.ArrayList;
  import java.util.List;
  import java.util.Properties;
  
  public class Consumer {
  
      private static List&lt;ConsumerMultiThreaded&gt; consumersGroup;
  
  
      public static void main(String args[]) throws IOException {
  
  
          URL in = ProducerWithTx.class.getResource("consumer.properties");
  
          Properties props = new Properties();
  
          try {
              props.load(new FileReader(new File(in.getFile())));
          } catch (IOException e1) {
              // TODO Auto-generated catch block
              e1.printStackTrace();
          }
  
          consumersGroup=new ArrayList&lt;ConsumerMultiThreaded&gt;();
          ConsumerMultiThreaded con1= new ConsumerMultiThreaded(props);
          ConsumerMultiThreaded con2=new ConsumerMultiThreaded(props);
          ConsumerMultiThreaded con3=new ConsumerMultiThreaded(props);
          ConsumerMultiThreaded con4=new ConsumerMultiThreaded(props);
  
          consumersGroup.add(con1);
          consumersGroup.add(con2);
          consumersGroup.add(con3);
          consumersGroup.add(con4);
  
          for (ConsumerMultiThreaded consumer : consumersGroup) {
  
              Thread t=new Thread(consumer);
              t.start();
  
          }
  
          while(true){
              try {
                  Thread.sleep(100000);
              } catch (InterruptedException ie) {
  
              }
  
  
          }
      }
  }   
  </code></pre>
  
  <p>Consumer Runnable </p>
  
  <pre class="lang-java prettyprint-override"><code>public class ConsumerMultiThreaded implements Runnable {
  
  private final AtomicBoolean closed = new AtomicBoolean(false);
      private final KafkaConsumer&lt;String, Paymnt&gt; consumer;
      private final int minBatchSize =3;
      private final List&lt;ConsumerRecord&lt;String, Paymnt&gt;&gt; buffer;
  
  
      public ConsumerMultiThreaded(Properties props){
          this.consumer= new KafkaConsumer&lt;String, Paymnt&gt;(props);
          buffer = new ArrayList(minBatchSize);
      }
  
      @Override
      public void run() {
          try {
              consumer.subscribe(Arrays.asList("test"));
              while (!closed.get()) {
                  ConsumerRecords&lt;String,Paymnt&gt; records = consumer.poll(10000);
  
                  for (ConsumerRecord&lt;String, Paymnt&gt; record : records) {
                      buffer.add(record);
                  }
  
                  /*for (ConsumerRecord&lt;String, Paymnt&gt; record : records)
                  {
                      System.out.println("record consumed by Thread " +Thread.currentThread().getId() +" value is " +record.value().toString());
                  }*/
                  if(buffer.size()&gt;=minBatchSize){
                      for (TopicPartition partition : records.partitions()) {
                          List&lt;ConsumerRecord&lt;String, Paymnt&gt;&gt; partitionRecords = records.records(partition);
                          for (ConsumerRecord&lt;String, Paymnt&gt; record : partitionRecords) {
                              System.out.println("record consumed by Thread " +Thread.currentThread().getId() +"from partition" +record.partition() +"offset" +record.offset() + "value: " + record.value().toString());
                          }
                          long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
                          consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
                          buffer.clear();
                      }
                  }
              }
  
          } catch (WakeupException e) {
              // Ignore exception if closing
              if (!closed.get()) throw e;
          } 
          finally {
              consumer.close();
          }
  
      }
  
      public void shutdown() {
          closed.set(true);
          consumer.wakeup();
      }
  
  }
  </code></pre>
  
  <p>consumer.properties</p>
  
  <pre><code>bootstrap.servers=localhost:9092
  session.timeout.ms=30000
  key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
  value.deserializer=com.org.kafkaPro.PayDeserializer
  enable.auto.commit=false
  auto.offset.reset=earliest
  group.id=test
  isolation.level=read_committed
  </code></pre>
  
  <p>Appreciate your help..Thank you</p>