diff --git a/build.gradle b/build.gradle index f3f50b6e4..25428e95c 100644 --- a/build.gradle +++ b/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") diff --git a/spring-batch-docs/asciidoc/appendix.adoc b/spring-batch-docs/asciidoc/appendix.adoc index 824b7b55c..23c565ff6 100644 --- a/spring-batch-docs/asciidoc/appendix.adoc +++ b/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 diff --git a/spring-batch-docs/asciidoc/readersAndWriters.adoc b/spring-batch-docs/asciidoc/readersAndWriters.adoc index 5ad293daf..292fd9326 100644 --- a/spring-batch-docs/asciidoc/readersAndWriters.adoc +++ b/spring-batch-docs/asciidoc/readersAndWriters.adoc @@ -3054,6 +3054,7 @@ Spring Batch offers the following readers and writers for commonly used messagin * <> * <> * <> +* <> [[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: 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 new file mode 100644 index 000000000..ea5e37996 --- /dev/null +++ b/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; + +/** + *

+ * An {@link ItemWriter} implementation for Apache Kafka using a + * {@link KafkaTemplate} with default topic configured. + *

+ * + * @author Mathieu Ouellet + * @since 4.2 + * + */ +public class KafkaItemWriter extends KeyValueItemWriter { + + private KafkaTemplate 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 kafkaTemplate) { + this.kafkaTemplate = kafkaTemplate; + } +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/builder/KafkaItemWriterBuilder.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/builder/KafkaItemWriterBuilder.java new file mode 100644 index 000000000..86dff9fc7 --- /dev/null +++ b/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 { + + private KafkaTemplate kafkaTemplate; + + private Converter 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 kafkaTemplate(KafkaTemplate 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 itemKeyMapper(Converter 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 delete(boolean delete) { + this.delete = delete; + return this; + } + + /** + * Validates and builds a {@link KafkaItemWriter}. + * + * @return a {@link KafkaItemWriter} + */ + public KafkaItemWriter build() { + Assert.notNull(this.kafkaTemplate, "kafkaTemplate is required."); + Assert.notNull(this.itemKeyMapper, "itemKeyMapper is required."); + + KafkaItemWriter writer = new KafkaItemWriter<>(); + writer.setKafkaTemplate(this.kafkaTemplate); + writer.setItemKeyMapper(this.itemKeyMapper); + writer.setDelete(this.delete); + return writer; + } +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/builder/package-info.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/builder/package-info.java new file mode 100644 index 000000000..09021d79b --- /dev/null +++ b/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; \ No newline at end of file diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/package-info.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/package-info.java new file mode 100644 index 000000000..820d538e4 --- /dev/null +++ b/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; \ No newline at end of file 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 new file mode 100644 index 000000000..b048497c5 --- /dev/null +++ b/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 kafkaTemplate; + + private KafkaItemKeyMapper itemKeyMapper; + + private KafkaItemWriter 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 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 { + + @Override + public String convert(String source) { + return source; + } + } +} \ No newline at end of file diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/builder/KafkaItemWriterBuilderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/builder/KafkaItemWriterBuilderTests.java new file mode 100644 index 000000000..22d21c022 --- /dev/null +++ b/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 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().itemKeyMapper(this.itemKeyMapper).build(); + } + + @Test + public void testNullItemKeyMapper() { + this.thrown.expect(IllegalArgumentException.class); + this.thrown.expectMessage("itemKeyMapper is required."); + + new KafkaItemWriterBuilder().kafkaTemplate(this.kafkaTemplate).build(); + } + + @Test + public void testKafkaItemWriterBuild() { + // given + boolean delete = true; + + // when + KafkaItemWriter writer = new KafkaItemWriterBuilder() + .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 { + + @Override + public String convert(String source) { + return source; + } + } + +} \ No newline at end of file