From feb46fd3f74716db42e2d34639d4a6952e7267e5 Mon Sep 17 00:00:00 2001
From: Aditya Sathe <satheaditya2014@gmail.com>
Date: Sat, 21 Nov 2020 19:55:31 +0530
Subject: [PATCH] Make kafkaTemplate protected in KafkaItemWriter

Issue #3802
---
 .../batch/item/kafka/KafkaItemWriter.java           |  2 +-
 .../batch/item/kafka/KafkaItemWriterTests.java      | 13 +++++++++++++
 2 files changed, 14 insertions(+), 1 deletion(-)

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 d924dd510..cf12328c2 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
@@ -33,7 +33,7 @@ import org.springframework.util.Assert;
  */
 public class KafkaItemWriter<K, T> extends KeyValueItemWriter<K, T> {
 
-	private KafkaTemplate<K, T> kafkaTemplate;
+	protected KafkaTemplate<K, T> kafkaTemplate;
 
 	@Override
 	protected void writeKeyValue(K key, T value) {
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 5b97c1210..ad24c3fd2 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
@@ -103,6 +103,19 @@ public class KafkaItemWriterTests {
 		verify(this.kafkaTemplate).sendDefault(items.get(1), null);
 	}
 
+	@Test
+	public void testKafkaTemplateCanBeReferencedFromSubclass() {
+		KafkaItemWriter<String, String> kafkaItemWriter = new KafkaItemWriter<String, String>() {
+			@Override
+			protected void writeKeyValue(String key, String value) {
+				this.kafkaTemplate.sendDefault(key, value);
+			}
+		};
+		kafkaItemWriter.setKafkaTemplate(kafkaTemplate);
+		kafkaItemWriter.writeKeyValue("k", "v");
+		verify(kafkaTemplate).sendDefault("k", "v");
+	}
+
 	static class KafkaItemKeyMapper implements Converter<String, String> {
 
 		@Override