diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/KeyValueItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/KeyValueItemWriter.java
index 586a548a6..b1e1141b6 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/KeyValueItemWriter.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/KeyValueItemWriter.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  * 
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
  * the License. You may obtain a copy of the License at
@@ -45,6 +45,12 @@ public abstract class KeyValueItemWriter<K, V> implements ItemWriter<V>, Initial
 		}
 		flush();
 	}
+
+	/**
+	 * Flush items to the key/value store.
+	 * 
+	 * @throws Exception if unable to flush items
+	 */
 	protected void flush() throws Exception {}
 
 	/**
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/KafkaItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/KafkaItemWriter.java
index f9ad5125d..a96f5c6c6 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/KafkaItemWriter.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/KafkaItemWriter.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2020 the original author or authors.
+ * Copyright 2019-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -44,19 +44,20 @@ public class KafkaItemWriter<K, T> extends KeyValueItemWriter<K, T> {
 	@Override
 	protected void writeKeyValue(K key, T value) {
 		if (this.delete) {
-			listenableFutures.add(this.kafkaTemplate.sendDefault(key, null));
+			this.listenableFutures.add(this.kafkaTemplate.sendDefault(key, null));
 		}
 		else {
-			listenableFutures.add(this.kafkaTemplate.sendDefault(key, value));
+			this.listenableFutures.add(this.kafkaTemplate.sendDefault(key, value));
 		}
 	}
+
 	@Override
 	protected void flush() throws Exception{
-		kafkaTemplate.flush();
-		for(ListenableFuture<SendResult<K,T>> future: listenableFutures){
+		this.kafkaTemplate.flush();
+		for(ListenableFuture<SendResult<K,T>> future: this.listenableFutures){
 			future.get();
 		}
-		listenableFutures.clear();
+		this.listenableFutures.clear();
 	}
 
 	@Override
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/KafkaItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/KafkaItemWriterTests.java
index 19337af5b..0eac5ac2d 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/KafkaItemWriterTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/KafkaItemWriterTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2020 the original author or authors.
+ * Copyright 2019-2021 the original author or authors.
  *
  *  Licensed under the Apache License, Version 2.0 (the "License");
  *  you may not use this file except in compliance with the License.
@@ -31,7 +31,9 @@ import org.springframework.util.concurrent.ListenableFuture;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.times;
 
 public class KafkaItemWriterTests {
 
@@ -49,7 +51,7 @@ public class KafkaItemWriterTests {
 	public void setUp() throws Exception {
 		MockitoAnnotations.openMocks(this);
 		when(this.kafkaTemplate.getDefaultTopic()).thenReturn("defaultTopic");
-		when(this.kafkaTemplate.sendDefault(any(), any())).thenReturn(future);
+		when(this.kafkaTemplate.sendDefault(any(), any())).thenReturn(this.future);
 		this.itemKeyMapper = new KafkaItemKeyMapper();
 		this.writer = new KafkaItemWriter<>();
 		this.writer.setKafkaTemplate(this.kafkaTemplate);