title: Reactor Kafka: Exactly Once Processing Sample
body: <p>I've read many articles where there are many different configurations to achieve exactly once processing.</p>
  
  <p>Here is my producer config:</p>
  
  <pre><code>final Map&lt;String, Object&gt; props = Maps.newConcurrentMap();
  
  props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
  props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
  
  
  props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
  props.put(ProducerConfig.ACKS_CONFIG, "all"); 
  props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "producer-tx-1");
  </code></pre>
  
  <p>Here is my consumer config:</p>
  
  <pre><code>final Map&lt;String, Object&gt; props = Maps.newHashMap();
  
  props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
  props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
  props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
  
  props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
  props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
  </code></pre>
  
  <p>I read this [sample scenarios][1]</p>
  
  <p>And I try to follow but i'm getting some issues:</p>
  
  <p>Here is my producer code:</p>
  
  <pre><code>    @Override
  public Mono&lt;SenderResult&lt;Void&gt;&gt; buy(Message msg) {
      final ReactiveKafkaProducerTemplate kafkaProducerTemplate = kafkaConfig.getKafkaProducerTemplate();
      return kafkaProducerTemplate.transactionManager().begin().then(kafkaProducerTemplate.send(mytopic, msg));
  
  }
  </code></pre>
  
  <p>My Consumer code:</p>
  
  <pre><code>@Override
  public void run(ApplicationArguments arg0) throws Exception {
      final ReactiveKafkaProducerTemplate kafkaProducerTemplate = kafkaConfig.getKafkaProducerTemplate();
      final ReactiveKafkaConsumerTemplate kafkaConsumerTemplate = kafkaConfig.getKafkaConsumerTemplate(mytopic, Message.class);
  
      final Flux&lt;ConsumerRecord&lt;String, Message&gt;&gt; flux = kafkaConsumerTemplate.receiveExactlyOnce(kafkaProducerTemplate.transactionManager())
              .concatMap(receiverRecordFlux -&gt; receiverRecordFlux );
  
      flux.subscribe(record -&gt; {
          final Message message = record.value();
  
          System.out.printf("received message: timestamp=%s key=%d value=%s\n",
                  dateFormat.format(new Date(record.timestamp())),
                  record.key(),
                  message);
   transactionService.processAndSendToNextTopic(message)
                  .doOnSuccess(aVoid -&gt; kafkaProducerTemplate.transactionManager().commit())
                  .subscribe();
  
      });
  }
  </code></pre>
  
  <p>I'm always getting following error when trying to produce and consume message: </p>
  
  <pre><code>Invalid transition attempted from state IN_TRANSACTION to state IN_TRANSACTION
  </code></pre>