@Component
@StepScope
@AllArgsConstructor
public class KafkaWriter implements ItemWriter<JsonNode> {
 private final KafkaTemplate<String, JsonNode> kafkaTemplate;
 @Override
 public void write(Chunk<? extends JsonNode> chunk) throws Exception {
 Chunk<? extends JsonNode>.ChunkIterator iterator = chunk.iterator();
 while (iterator.hasNext()) {
 JsonNode jsonItem = iterator.next();
 try {
 ProducerRecord<String, JsonNode> producerRecord = createProducerRecord(jsonItem);
 send(producerRecord);
 } catch (Exception e) {
 iterator.remove(e);
 }
 }
 }
 private void send(ProducerRecord<String, JsonNode> producerRecord) throws ExecutionException, InterruptedException {
 CompletableFuture<SendResult<String, JsonNode>> future = kafkaTemplate.send(producerRecord);
 future.get();
 }
 private ProducerRecord<String, JsonNode> createProducerRecord(JsonNode jsonNode) {
 List<Header> headers = List.of();
 return new ProducerRecord<>("foo", null, "bar", jsonNode, headers);
 }
}

@Slf4j
@AllArgsConstructor
public class FooListener implements StepExecutionListener {
 @Override
 public void beforeStep(StepExecution stepExecution) {
 }
 @Override
 public ExitStatus afterStep(StepExecution stepExecution) {
 long readCount = stepExecution.getReadCount();
 long skipCount = stepExecution.getSkipCount();
 long writeCount = stepExecution.getWriteCount();
 log.info("ReadCount:[{}], SkipCount:[{}], WriteCount:[{}]", readCount, skipCount, writeCount);
 return stepExecution.getExitStatus();
 }
}

@Bean
public Step fooStep(
 SynchronizedItemStreamReader reader,
 AsyncItemProcessor processor,
 AsyncItemWriter writer) {
 return new StepBuilder("Bar STEP", jobRepository)
 .listener(new FooListener())
 .<Object, JsonNode>chunk(10, transactionManager)
 .reader(reader)
 .processor(processor)
 .writer(writer)
 .faultTolerant()
 .skip(Exception.class)
 .skipPolicy(new AlwaysSkipItemSkipPolicy())
 .noRetry(Exception.class)
 .processorNonTransactional()
 .build();
}