Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Polish 06556dc97e

Browse Source
This commit is contained in:
Mahmoud Ben Hassine
2019-05-07 17:14:12 +02:00
parent 06556dc97e
commit 00ac7c5833
6 changed files with 83 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<K, T> extends KeyValueItemWriter<K, T> {
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<K, T> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}

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

@@ -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<K, V> {
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<K, V> {
/**
* 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<K, V> {
}
/**
* 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<K, V> {
/**
* Validates and builds a {@link KafkaItemWriter}.
*
* @return a {@link KafkaItemWriter}
*/
public KafkaItemWriter<K, V> build() {

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

@@ -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;
import org.springframework.lang.NonNullApi;

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

@@ -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;
import org.springframework.lang.NonNullApi;

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

@@ -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<String> 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<String> 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<String, String> {
@@ -75,4 +110,4 @@ public class KafkaItemWriterTests {
return source;
}
}
}
}

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

@@ -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 {
}
}
}
}