title: How to commit the offsets when using KafkaItemReader in spring batch job, once all the messages are processed and written to the .dat file?
body: <p>I have developed a Spring Batch Job which read from Kafka topic using KafkaItemReader class. I want to commit the offset only when the messages read in defined chunk are Processed and written successfully to an Output .dat file.</p>
  
  <pre><code>@Bean
  public Job kafkaEventReformatjob(
          @Qualifier("MaintStep") Step MainStep,
          @Qualifier("moveFileToFolder") Step moveFileToFolder,
          @Qualifier("compressFile") Step compressFile,
          JobExecutionListener listener)
  {
      return jobBuilderFactory.get("kafkaEventReformatJob")
              .listener(listener)
              .incrementer(new RunIdIncrementer())
              .flow(MainStep)
              .next(moveFileToFolder)
              .next(compressFile)
              .end()
              .build();
  }
  
  @Bean
  Step MainStep(
          ItemProcessor&lt;IncomingRecord, List&lt;Record&gt;&gt; flatFileItemProcessor,
          ItemWriter&lt;List&lt;Record&gt;&gt; flatFileWriter)
  {
      return stepBuilderFactory.get("mainStep")
              .&lt;InputRecord, List&lt;Record&gt;&gt; chunk(5000)
              .reader(kafkaItemReader())
              .processor(flatFileItemProcessor)
              .writer(writer())
              .listener(basicStepListener)
              .build();
  }
  //Reader reads all the messages from akfka topic and sending back in form of IncomingRecord.
   @Bean
  KafkaItemReader&lt;String, IncomingRecord&gt; kafkaItemReader() {
      Properties props = new Properties();
      props.putAll(this.properties.buildConsumerProperties());
      List&lt;Integer&gt; partitions = new ArrayList&lt;&gt;();
      partitions.add(0);
      partitions.add(1);
      return new KafkaItemReaderBuilder&lt;String, IncomingRecord&gt;()
              .partitions(partitions)
              .consumerProperties(props)
              .name("records")
              .saveState(true)
              .topic(topic)
              .pollTimeout(Duration.ofSeconds(40L))
              .build();
  }
  
    @Bean
  public ItemWriter&lt;List&lt;Record&gt;&gt; writer() {
      ListUnpackingItemWriter&lt;Record&gt; listUnpackingItemWriter = new ListUnpackingItemWriter&lt;&gt;();
      listUnpackingItemWriter.setDelegate(flatWriter());
      return listUnpackingItemWriter;
  }
  
  public ItemWriter&lt;Record&gt; flatWriter() {
      FlatFileItemWriter&lt;Record&gt; fileWriter = new FlatFileItemWriter&lt;&gt;();
      String tempFileName = "abc";
      LOGGER.info("Output File name " + tempFileName + " is in working directory ");
      String workingDir = service.getWorkingDir().toAbsolutePath().toString();
      Path outputFile = Paths.get(workingDir, tempFileName);
      fileWriter.setName("fileWriter");
      fileWriter.setResource(new FileSystemResource(outputFile.toString()));
      fileWriter.setLineAggregator(lineAggregator());
      fileWriter.setForceSync(true);
      fileWriter.setFooterCallback(customFooterCallback());
      fileWriter.close();
      LOGGER.info("Successfully created the file writer");
      return fileWriter;
  
  }
  
  @StepScope
  @Bean
  public TransformProcessor processor() {
      return new TransformProcessor();
  }
  </code></pre>
  
  <p>==============================================================================</p>
  
  <h1>Writer Class</h1>
  
  <pre><code> @BeforeStep
  public void beforeStep(StepExecution stepExecution) {
      this.stepExecution = stepExecution;
  }
  
  @AfterStep
  public void afterStep(StepExecution stepExecution) {
      this.stepExecution.setWriteCount(count);
  }
  
  @Override
  public void write(final List&lt;? extends List&lt;Record&gt;&gt; lists) throws Exception {
  
      List&lt;Record&gt; consolidatedList = new ArrayList&lt;&gt;();
      for (List&lt;Record&gt; list : lists) {
          if (!list.isEmpty() &amp;&amp; null != list)
              consolidatedList.addAll(list);
      }
  
      delegate.write(consolidatedList);
      count += consolidatedList.size(); // to count Trailer record count
  }
  </code></pre>
  
  <p>===============================================================</p>
  
  <h1>Item Processor</h1>
  
  <p>@Override
      public List process(IncomingRecord record) {</p>
  
  <pre><code>    List&lt;Record&gt; recordList = new ArrayList&lt;&gt;();
  
      if (null != record.getEventName() and a few other conditions inside this section) {
          // setting values of Record Class by extracting from the IncomingRecord.
          recordList.add(the valid records which matching the condition);
          }else{
          return null;
          }
  </code></pre>