Add Kafka ItemWriter

This commit adds an ItemWriter implementation for Apache Kafka using
a KafkaTemplate with default topic set. Extending KeyValueItemWriter,
the writer requires a Converter to map the Kafka ProducerRecord's key
from the item.

Resolves BATCH-2765

build.gradle

@@ -56,6 +56,7 @@ allprojects {
 		springDataMongodbVersion = '2.2.0.M3'
 		springDataNeo4jVersion = '5.2.0.M3'
 		springIntegrationVersion = '5.2.0.M1'
+		springKafkaVersion = '2.2.5.RELEASE'
 		springLdapVersion = '2.3.2.RELEASE'
 
 		activemqVersion = '5.15.9'
@@ -380,6 +381,7 @@ project('spring-batch-infrastructure') {
         optional "org.springframework.ldap:spring-ldap-core:$springLdapVersion"
         optional "org.springframework.ldap:spring-ldap-core-tiger:$springLdapVersion"
         optional "org.springframework.ldap:spring-ldap-ldif-core:$springLdapVersion"
+		optional "org.springframework.kafka:spring-kafka:$springKafkaVersion"
 		// JSR-305 only used for non-required meta-annotations
 		compileOnly("com.google.code.findbugs:jsr305:3.0.2")
 		testCompileOnly("com.google.code.findbugs:jsr305:3.0.2")

spring-batch-docs/asciidoc/appendix.adoc

@@ -102,6 +102,9 @@
             and handles some transaction-related work that a non-"JPA-aware"
             `ItemWriter` would not need to know about and
             then delegates to another writer to do the actual writing.
+|KafkaItemWriter|Using a `KafkaTemplate` object, items are written to the default topic through the
+            `KafkaTemplate#sendDefault(Object, Object)` method using a `Converter` to map the key from the item.
+            A delete flag can also be configured to send delete events to the topic.
 |MimeMessageItemWriter|Using Spring's `JavaMailSender`, items of type `MimeMessage`
             are sent as mail messages.
 |MongoItemWriter|Given a `MongoOperations` object, items are written

spring-batch-docs/asciidoc/readersAndWriters.adoc

@@ -3054,6 +3054,7 @@ Spring Batch offers the following readers and writers for commonly used messagin
 * <<amqpItemWriter>>
 * <<jmsItemReader>>
 * <<jmsItemWriter>>
+* <<kafkaItemWriter>>
 
 [[amqpItemReader]]
 ===== `AmqpItemReader`
@@ -3081,6 +3082,12 @@ The `JmsItemWriter` is an `ItemWriter` for JMS that uses a `JmsTemplate`. The te
 should have a default destination, which is used to send items in `write(List)`. Spring
 Batch provides a `JmsItemWriterBuilder` to construct an instance of the `JmsItemWriter`.
 
+[[kafkaItemWriter]]
+===== `KafkaItemWriter`
+The `KafkaItemWriter` is an `ItemWriter` for Apache Kafka that uses a `KafkaTemplate` to
+send events to a default topic. Spring Batch provides a `KafkaItemWriterBuilder` to
+construct an instance of the `KafkaItemWriter`.
+
 [[databaseReaders]]
 ==== Database Readers
 Spring Batch offers the following database readers:

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/KafkaItemWriter.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.batch.item.kafka;
+
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.batch.item.KeyValueItemWriter;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.util.Assert;
+
+/**
+ * <p>
+ * An {@link ItemWriter} implementation for Apache Kafka using a
+ * {@link KafkaTemplate} with default topic configured.
+ * </p>
+ *
+ * @author Mathieu Ouellet
+ * @since 4.2
+ *
+ */
+public class KafkaItemWriter<K, T> extends KeyValueItemWriter<K, T> {
+
+	private KafkaTemplate<K, T> kafkaTemplate;
+
+	@Override
+	protected void writeKeyValue(K key, T value) {
+		if (this.delete) {
+			this.kafkaTemplate.sendDefault(key, null);
+		}
+		else {
+			this.kafkaTemplate.sendDefault(key, value);
+		}
+	}
+
+	@Override
+	protected void init() {
+		Assert.notNull(this.kafkaTemplate, "KafkaTemplate must not be null.");
+		Assert.notNull(this.kafkaTemplate.getDefaultTopic(), "KafkaTemplate must have the default topic set.");
+	}
+
+	public void setKafkaTemplate(KafkaTemplate<K, T> kafkaTemplate) {
+		this.kafkaTemplate = kafkaTemplate;
+	}
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/builder/KafkaItemWriterBuilder.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2018 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
+ *
+ *          http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.springframework.batch.item.kafka.builder;
+
+import org.springframework.batch.item.kafka.KafkaItemWriter;
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.util.Assert;
+
+/**
+ * A builder implementation for the {@link KafkaItemWriter}
+ *
+ * @author Mathieu Ouellet
+ * @since 4.2
+ */
+public class KafkaItemWriterBuilder<K, V> {
+
+	private KafkaTemplate<K, V> kafkaTemplate;
+
+	private Converter<V, K> itemKeyMapper;
+
+	private boolean delete;
+
+	/**
+	 * Establishes the KafkaTemplate to be used by the KafkaItemWriter.
+	 * @param kafkaTemplate the template to be used
+	 * @return this instance for method chaining
+	 * @see KafkaItemWriter#setKafkaTemplate(KafkaTemplate)
+	 */
+	public KafkaItemWriterBuilder<K, V> kafkaTemplate(KafkaTemplate<K, V> kafkaTemplate) {
+		this.kafkaTemplate = kafkaTemplate;
+		return this;
+	}
+
+	/**
+	 * Set the {@link Converter} to use to derive the key from the item.
+	 *
+	 * @param itemKeyMapper the Converter to use.
+	 * @return The current instance of the builder.
+	 * @see KafkaItemWriter#setItemKeyMapper(Converter)
+	 */
+	public KafkaItemWriterBuilder<K, V> itemKeyMapper(Converter<V, K> itemKeyMapper) {
+		this.itemKeyMapper = itemKeyMapper;
+		return this;
+	}
+
+	/**
+	 * Indicates if the items being passed to the writer are all to be send as delete events to the topic. A delete
+	 * event is made of a key with a null value. If set to false (default), the items will be send with provided value
+	 * and key converter by the itemKeyMapper. If set to true, the items will be send with the key converter from the
+	 * value by the itemKeyMapper and a null value.
+	 *
+	 * @param delete removal indicator.
+	 * @return The current instance of the builder.
+	 * @see KafkaItemWriter#setDelete(boolean)
+	 */
+	public KafkaItemWriterBuilder<K, V> delete(boolean delete) {
+		this.delete = delete;
+		return this;
+	}
+
+	/**
+	 * Validates and builds a {@link KafkaItemWriter}.
+	 * 
+	 * @return a {@link KafkaItemWriter}
+	 */
+	public KafkaItemWriter<K, V> build() {
+		Assert.notNull(this.kafkaTemplate, "kafkaTemplate is required.");
+		Assert.notNull(this.itemKeyMapper, "itemKeyMapper is required.");
+
+		KafkaItemWriter<K, V> writer = new KafkaItemWriter<>();
+		writer.setKafkaTemplate(this.kafkaTemplate);
+		writer.setItemKeyMapper(this.itemKeyMapper);
+		writer.setDelete(this.delete);
+		return writer;
+	}
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/builder/package-info.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2018 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
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Builders for Apache Kafka item reader and writer.
+ *
+ * @author Mathieu Ouellet
+ */
+@NonNullApi
+package org.springframework.batch.item.kafka.builder;
+
+import org.springframework.lang.NonNullApi;

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/package-info.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2018 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
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Apache Kafka related readers and writers
+ *
+ * @author Mathieu Ouellet
+ */
+@NonNullApi
+package org.springframework.batch.item.kafka;
+
+import org.springframework.lang.NonNullApi;

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/KafkaItemWriterTests.java

@@ -0,0 +1,78 @@
+package org.springframework.batch.item.kafka;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.kafka.core.KafkaTemplate;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class KafkaItemWriterTests {
+
+	@Mock
+	private KafkaTemplate<String, String> kafkaTemplate;
+
+	private KafkaItemKeyMapper itemKeyMapper;
+
+	private KafkaItemWriter<String, String> writer;
+
+	@Before
+	public void setUp() throws Exception {
+		MockitoAnnotations.initMocks(this);
+		when(kafkaTemplate.getDefaultTopic()).thenReturn("defaultTopic");
+		itemKeyMapper = new KafkaItemKeyMapper();
+		writer = new KafkaItemWriter<>();
+		writer.setKafkaTemplate(kafkaTemplate);
+		writer.setItemKeyMapper(itemKeyMapper);
+		writer.setDelete(false);
+		writer.afterPropertiesSet();
+	}
+
+	@Test
+	public void testAfterPropertiesSet() throws Exception {
+		writer = new KafkaItemWriter<>();
+
+		try {
+			writer.afterPropertiesSet();
+			fail("Expected exception was not thrown");
+		}
+		catch (IllegalArgumentException ignore) {
+		}
+
+		writer.setKafkaTemplate(kafkaTemplate);
+		try {
+			writer.afterPropertiesSet();
+			fail("Expected exception was not thrown");
+		}
+		catch (IllegalArgumentException ignore) {
+		}
+
+		writer.setItemKeyMapper(itemKeyMapper);
+		writer.afterPropertiesSet();
+	}
+
+	@Test
+	public void testBasicWrite() throws Exception {
+		List<String> items = Arrays.asList("val1", "val2");
+
+		writer.write(items);
+
+		verify(kafkaTemplate).sendDefault(items.get(0), items.get(0));
+		verify(kafkaTemplate).sendDefault(items.get(1), items.get(1));
+	}
+
+	static class KafkaItemKeyMapper implements Converter<String, String> {
+
+		@Override
+		public String convert(String source) {
+			return source;
+		}
+	}
+}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/builder/KafkaItemWriterBuilderTests.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2018 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
+ *
+ *          http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.springframework.batch.item.kafka.builder;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.springframework.batch.item.kafka.KafkaItemWriter;
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.test.util.ReflectionTestUtils;
+
+/**
+ * @author Mathieu Ouellet
+ */
+public class KafkaItemWriterBuilderTests {
+
+	@Rule
+	public ExpectedException thrown = ExpectedException.none();
+
+	@Mock
+	private KafkaTemplate<String, String> kafkaTemplate;
+
+	private KafkaItemKeyMapper itemKeyMapper;
+
+	@Before
+	public void setUp() {
+		MockitoAnnotations.initMocks(this);
+		this.itemKeyMapper = new KafkaItemKeyMapper();
+	}
+
+	@Test
+	public void testNullKafkaTemplate() {
+		this.thrown.expect(IllegalArgumentException.class);
+		this.thrown.expectMessage("kafkaTemplate is required.");
+
+		new KafkaItemWriterBuilder<String, String>().itemKeyMapper(this.itemKeyMapper).build();
+	}
+
+	@Test
+	public void testNullItemKeyMapper() {
+		this.thrown.expect(IllegalArgumentException.class);
+		this.thrown.expectMessage("itemKeyMapper is required.");
+
+		new KafkaItemWriterBuilder<String, String>().kafkaTemplate(this.kafkaTemplate).build();
+	}
+
+	@Test
+	public void testKafkaItemWriterBuild() {
+		// given
+		boolean delete = true;
+
+		// when
+		KafkaItemWriter<String, String> writer = new KafkaItemWriterBuilder<String, String>()
+				.kafkaTemplate(this.kafkaTemplate).itemKeyMapper(this.itemKeyMapper).delete(delete).build();
+
+		// then
+		assertTrue((Boolean) ReflectionTestUtils.getField(writer, "delete"));
+		assertEquals(this.itemKeyMapper, ReflectionTestUtils.getField(writer, "itemKeyMapper"));
+		assertEquals(this.kafkaTemplate, ReflectionTestUtils.getField(writer, "kafkaTemplate"));
+	}
+
+	static class KafkaItemKeyMapper implements Converter<String, String> {
+
+		@Override
+		public String convert(String source) {
+			return source;
+		}
+	}
+
+}