title: Kafka consumer Not committing offset using reactor-kafka
body: <p>I am newbie to Kafka and my kafka consumer is not commiting offset after specified number of messages are consumed.I am trying to do manual commit after consuming every 5 messages using reactor-kafka project.I am verifying my offset commit using lenses consumers UI.</p>
  <p>The following are my configurations:</p>
  <pre><code>&lt;dependency&gt;
      &lt;groupId&gt;io.projectreactor.kafka&lt;/groupId&gt;
      &lt;artifactId&gt;reactor-kafka&lt;/artifactId&gt;
      &lt;version&gt;1.3.11&lt;/version&gt;
  &lt;/dependency&gt;
  
  
  
  
  
     @EventListener(ApplicationReadyEvent.class)
          public void consumerListener() {
              Properties props = new Properties();
      
              props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapServersUrl);
              props.put(ConsumerConfig.GROUP_ID_CONFIG,&quot;consumer-group-1&quot;);
              props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
              props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,&quot;latest&quot;);
              props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
              props.put(ApplicationConstants.SECURITY_PROTOCOL,&quot;SSL&quot;);
              props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,&quot;/etc/security/truststore.jks&quot;);
              props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,&quot;Password&quot;);
              props.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG,&quot;SSL&quot;);
              props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG,&quot;/etc/security/keystore.jks&quot;);
              props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG,&quot;Password&quot;);
              props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,300000);
              props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG,300000);
              props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG,15000);
              props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG,30000);
              props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,100);
              props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,&quot;org.apache.kafka.clients.consumer.CooperativeStickyAssignor&quot;);
          
  
   ReceiverOptions&lt;String,String&gt; receiverOptions = ReceiverOptions&lt;String,String&gt;.create(props)                                                             
                                 .subscription(Collections.singleton(&quot;kafkaprodTopic&quot;))                                                                          
                                 .commitBatchSize(5);
  
      Flux&lt;ReceiverRecord&lt;String,String&gt;&gt; prodFlux = KafkaReceiver.create(receiverOptions)
                                                         .receive();
  
  
                                         prodFlux.subscribe(r -&gt;{
                                              //do business logic
                                              r.receiverOffset().acknowledge();
                                         });
  }
  </code></pre>
  <p>My consumer is able to consume messages succesfully and I am not seeing any Kafka exceptions as well , only problem is commit offset is not getting updated.</p>