title: can we achieve exactly one processing of message using spring kafka batch listener?
body: <p>I am trying to achieve exactly once processing of each message on kafka topic. Here is my configuration:</p>
  
  <pre><code>config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");    
  config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
  config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
  config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  config.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 120000);
  config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 25);
  config.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 4096000);
  config.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 120000);
  config.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 600000);  
  config.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 8192000);
  </code></pre>
  
  <p>I have set <code>Acknolwedgement Mode</code> as MANUAL &amp; concurrency to 2. </p>
  
  <p>Yet it consumes messages more than once. Has anyone faced this issue.
  Also, with above configuration, consumer always receives only one message in one batch. I have tried increasing <code>fetch.min.bytes</code> and <code>fetch.max.wait.ms</code>, but it doesn't have any impact.</p>
  
  <p>The problem with batch configuration is resolved after I made change to ConcurrentKafkaListenerContainerFactory as followed:</p>
  
  <pre><code>ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory =
      new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
  factory.setConsumerFactory(consumerFactory());
  
  factory.setBatchListener(true);
  factory.getContainerProperties().setPollTimeout(3600000);
  </code></pre>
  
  <p>factory.getContainerProperties().setAckMode(org.springframework.kafka.listner.ContainerProperties.AckMode.MANUAL);
      factory.setMessageConverter(new BatchMessagingMessageConverter(stringJsonMessageConverter()));</p>