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 ea5e37996..d924dd510 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,11 +1,11 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2019 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 + * https://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, @@ -51,6 +51,10 @@ public class KafkaItemWriter extends KeyValueItemWriter { Assert.notNull(this.kafkaTemplate.getDefaultTopic(), "KafkaTemplate must have the default topic set."); } + /** + * Set the {@link KafkaTemplate} to use. + * @param kafkaTemplate to use + */ 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 index 86dff9fc7..30c83e315 100644 --- 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 @@ -1,11 +1,11 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2019 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 + * https://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, @@ -36,7 +36,7 @@ public class KafkaItemWriterBuilder { private boolean delete; /** - * Establishes the KafkaTemplate to be used by the KafkaItemWriter. + * Establish 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) @@ -48,7 +48,6 @@ public class KafkaItemWriterBuilder { /** * 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) @@ -59,11 +58,10 @@ public class KafkaItemWriterBuilder { } /** - * 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 + * Indicate if the items being passed to the writer are all to be sent 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 sent with provided value + * and key converter by the itemKeyMapper. If set to true, the items will be sent 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) @@ -75,7 +73,6 @@ public class KafkaItemWriterBuilder { /** * Validates and builds a {@link KafkaItemWriter}. - * * @return a {@link KafkaItemWriter} */ public KafkaItemWriter build() { 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 index 09021d79b..45fb00839 100644 --- 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 @@ -1,11 +1,11 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2019 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 + * https://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, @@ -22,4 +22,4 @@ @NonNullApi package org.springframework.batch.item.kafka.builder; -import org.springframework.lang.NonNullApi; \ No newline at end of file +import org.springframework.lang.NonNullApi; 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 index 820d538e4..c0e084822 100644 --- 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 @@ -1,11 +1,11 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2019 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 + * https://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, @@ -22,4 +22,4 @@ @NonNullApi package org.springframework.batch.item.kafka; -import org.springframework.lang.NonNullApi; \ No newline at end of file +import org.springframework.lang.NonNullApi; 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 b048497c5..5b97c1210 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,16 +1,33 @@ +/* + * Copyright 2019 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 + * + * https://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 java.util.Arrays; +import java.util.List; + 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.junit.Assert.assertEquals; +import static org.junit.Assert.fail; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -26,46 +43,64 @@ public class KafkaItemWriterTests { @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(); + when(this.kafkaTemplate.getDefaultTopic()).thenReturn("defaultTopic"); + this.itemKeyMapper = new KafkaItemKeyMapper(); + this.writer = new KafkaItemWriter<>(); + this.writer.setKafkaTemplate(this.kafkaTemplate); + this.writer.setItemKeyMapper(this.itemKeyMapper); + this.writer.setDelete(false); + this.writer.afterPropertiesSet(); } @Test public void testAfterPropertiesSet() throws Exception { - writer = new KafkaItemWriter<>(); + this.writer = new KafkaItemWriter<>(); try { - writer.afterPropertiesSet(); + this.writer.afterPropertiesSet(); fail("Expected exception was not thrown"); } - catch (IllegalArgumentException ignore) { + catch (IllegalArgumentException exception) { + assertEquals("itemKeyMapper requires a Converter type.", exception.getMessage()); } - writer.setKafkaTemplate(kafkaTemplate); + this.writer.setItemKeyMapper(this.itemKeyMapper); try { - writer.afterPropertiesSet(); + this.writer.afterPropertiesSet(); fail("Expected exception was not thrown"); } - catch (IllegalArgumentException ignore) { + catch (IllegalArgumentException exception) { + assertEquals("KafkaTemplate must not be null.", exception.getMessage()); } - writer.setItemKeyMapper(itemKeyMapper); - writer.afterPropertiesSet(); + this.writer.setKafkaTemplate(this.kafkaTemplate); + try { + this.writer.afterPropertiesSet(); + } + catch (Exception e) { + fail("Must not throw an exception when correctly configured"); + } } @Test public void testBasicWrite() throws Exception { List items = Arrays.asList("val1", "val2"); - writer.write(items); + this.writer.write(items); - verify(kafkaTemplate).sendDefault(items.get(0), items.get(0)); - verify(kafkaTemplate).sendDefault(items.get(1), items.get(1)); + verify(this.kafkaTemplate).sendDefault(items.get(0), items.get(0)); + verify(this.kafkaTemplate).sendDefault(items.get(1), items.get(1)); + } + + @Test + public void testBasicDelete() throws Exception { + List items = Arrays.asList("val1", "val2"); + this.writer.setDelete(true); + + this.writer.write(items); + + verify(this.kafkaTemplate).sendDefault(items.get(0), null); + verify(this.kafkaTemplate).sendDefault(items.get(1), null); } static class KafkaItemKeyMapper implements Converter { @@ -75,4 +110,4 @@ public class KafkaItemWriterTests { 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 index 22d21c022..1ebe70ee9 100644 --- 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 @@ -1,11 +1,11 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2019 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 + * https://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, @@ -16,20 +16,21 @@ 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; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + /** * @author Mathieu Ouellet */ @@ -88,4 +89,4 @@ public class KafkaItemWriterBuilderTests { } } -} \ No newline at end of file +}