SpringCloud/spring-cloud-task
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Adding KafkaItemWriter single step autoconfig

Browse Source 
resolves #696

Updated to allow user to set the itemKeyMapper

Set a very simple keymapper that returns the same result

Updated based on code reviews

Updated Kafka version
This commit is contained in:
Glenn Renfro
2020-08-03 16:27:33 -04:00
parent a841508604
commit 626af44828
5 changed files with 322 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-cloud-starter-single-step-batch-job/pom.xml
View File

@@ -13,7 +13,7 @@
<test.containers.version>1.15.0</test.containers.version>
<test.rabbit.containers.version>1.15.0</test.rabbit.containers.version>
<test.ducttape.version>1.0.8</test.ducttape.version>
<spring-kafka-version>2.5.3.RELEASE</spring-kafka-version>
<spring-kafka-version>2.6.3</spring-kafka-version>
</properties>
<dependencies>

97
spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterAutoConfiguration.java Normal file
View File

@@ -0,0 +1,97 @@
/*
* Copyright 2020-2020 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.cloud.task.batch.autoconfigure.kafka;
import java.util.HashMap;
import java.util.Map;
import org.springframework.batch.item.kafka.KafkaItemWriter;
import org.springframework.batch.item.kafka.builder.KafkaItemWriterBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.batch.BatchAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.convert.converter.Converter;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.util.Assert;
/**
*
* Autconfiguration for a {@code KafkaItemReader}.
*
* @author Glenn Renfro
* @since 2.3
*/
@Configuration
@EnableConfigurationProperties({ KafkaProperties.class, KafkaItemWriterProperties.class })
@AutoConfigureAfter(BatchAutoConfiguration.class)
public class KafkaItemWriterAutoConfiguration {
@Autowired
private KafkaProperties kafkaProperties;
@Bean
@ConditionalOnMissingBean
@ConditionalOnProperty(prefix = "spring.batch.job.kafkaitemwriter", name = "topic")
public KafkaItemWriter<Object, Map<Object, Object>> kafkaItemWriter(
KafkaItemWriterProperties kafkaItemWriterProperties,
ProducerFactory<Object, Map<Object, Object>> producerFactory,
@Qualifier("batchItemKeyMapper") Converter<Object, Object> itemKeyMapper) {
validateProperties(kafkaItemWriterProperties);
KafkaTemplate template = new KafkaTemplate(producerFactory);
template.setDefaultTopic(kafkaItemWriterProperties.getTopic());
return new KafkaItemWriterBuilder<Object, Map<Object, Object>>()
.delete(kafkaItemWriterProperties.isDelete()).kafkaTemplate(template)
.itemKeyMapper(itemKeyMapper).build();
}
@Bean
@ConditionalOnMissingBean(name = "batchItemKeyMapper")
public Converter<Object, Object> batchItemKeyMapper() {
return new Converter<Object, Object>() {
@Override
public Object convert(Object source) {
return source;
}
};
}
@Bean
@ConditionalOnMissingBean
ProducerFactory<Object, Map<Object, Object>> producerFactory() {
Map<String, Object> configs = new HashMap<>();
configs.putAll(this.kafkaProperties.getProducer().buildProperties());
return new DefaultKafkaProducerFactory<Object, Map<Object, Object>>(configs, null,
new JsonSerializer<>());
}
private void validateProperties(KafkaItemWriterProperties kafkaItemWriterProperties) {
Assert.hasText(kafkaItemWriterProperties.getTopic(),
"topic must not be empty or null");
}
}

74
spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterProperties.java Normal file
View File

@@ -0,0 +1,74 @@
/*
* Copyright 2020-2020 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.cloud.task.batch.autoconfigure.kafka;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* Properties to configure a {@code KafkaItemWriter}.
*
* @author Glenn Renfro
* @since 2.3
*/
@ConfigurationProperties(prefix = "spring.batch.job.kafkaitemwriter")
public class KafkaItemWriterProperties {
private String topic;
private boolean delete;
/**
* Returns the name of the topic from which messages will be written.
* @return the name of the topic.
*/
public String getTopic() {
return topic;
}
/**
* The topic name from which the messages will be read.
* @param topic name of the topic
*/
public void setTopic(String topic) {
this.topic = topic;
}
/**
* 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.
* @return removal indicator.
*/
public boolean isDelete() {
return delete;
}
/**
* 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.
*/
public void setDelete(boolean delete) {
this.delete = delete;
}
}

4
spring-cloud-starter-single-step-batch-job/src/main/resources/META-INF/spring.factories
View File

@@ -6,4 +6,6 @@ org.springframework.boot.autoconfigure.EnableAutoConfiguration=org.springframewo
org.springframework.cloud.task.batch.autoconfigure.jdbc.JdbcCursorItemReaderAutoConfiguration, \
org.springframework.cloud.task.batch.autoconfigure.rabbit.AmqpItemReaderAutoConfiguration, \
org.springframework.cloud.task.batch.autoconfigure.rabbit.AmqpItemWriterAutoConfiguration, \
org.springframework.cloud.task.batch.autoconfigure.kafka.KafkaItemReaderAutoConfiguration
org.springframework.cloud.task.batch.autoconfigure.kafka.KafkaItemReaderAutoConfiguration, \
org.springframework.cloud.task.batch.autoconfigure.kafka.KafkaItemWriterAutoConfiguration

147
spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterTests.java Normal file
View File

@@ -0,0 +1,147 @@
/*
* Copyright 2020-2020 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.cloud.task.batch.autoconfigure.kafka;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.autoconfigure.batch.BatchAutoConfiguration;
import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.cloud.task.batch.autoconfigure.SingleStepJobAutoConfiguration;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import static java.util.Collections.singleton;
import static org.assertj.core.api.Assertions.assertThat;
@EmbeddedKafka(partitions = 1, topics = { "topic1" })
public class KafkaItemWriterTests {
private static EmbeddedKafkaBroker embeddedKafkaBroker;
@BeforeAll
public static void setupTest(EmbeddedKafkaBroker embeddedKafka) {
embeddedKafkaBroker = embeddedKafka;
embeddedKafka.addTopics("topic2");
}
@Test
public void testBaseKafkaItemWriter() {
final String topicName = "topic1";
ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner()
.withUserConfiguration(CustomMappingConfiguration.class)
.withConfiguration(
AutoConfigurations.of(PropertyPlaceholderAutoConfiguration.class,
BatchAutoConfiguration.class,
SingleStepJobAutoConfiguration.class,
KafkaItemWriterAutoConfiguration.class))
.withPropertyValues("spring.batch.job.jobName=job",
"spring.batch.job.stepName=step1", "spring.batch.job.chunkSize=5",
"spring.kafka.producer.bootstrap-servers="
+ embeddedKafkaBroker.getBrokersAsString(),
"spring.kafka.producer.keySerializer=org.springframework.kafka.support.serializer.JsonSerializer",
"spring.batch.job.kafkaitemwriter.topic=" + topicName);
applicationContextRunner.run((context) -> {
waitForTopicPopulation(context);
validateResults(topicName);
});
}
private void validateResults(String topicName) {
Map<String, Object> configs = new HashMap<>(
KafkaTestUtils.consumerProps("1", "false", embeddedKafkaBroker));
Consumer<String, Object> consumer = new DefaultKafkaConsumerFactory<>(configs,
new StringDeserializer(), new JsonDeserializer<>()).createConsumer();
consumer.subscribe(singleton(topicName));
ConsumerRecords<String, Object> consumerRecords = KafkaTestUtils
.getRecords(consumer);
assertThat(consumerRecords.count()).isEqualTo(5);
List<Map<Object, Object>> result = new ArrayList<>();
consumerRecords.forEach(cs -> {
result.add((Map<Object, Object>) cs.value());
});
List<String> firstNames = new ArrayList<>();
result.forEach(s -> firstNames.add((String) s.get("first_name")));
assertThat(firstNames.size()).isEqualTo(5);
assertThat(firstNames).contains("Jane");
assertThat(firstNames).contains("John");
assertThat(firstNames).contains("Liz");
assertThat(firstNames).contains("Cameron");
assertThat(firstNames).contains("Judy");
}
private void waitForTopicPopulation(ApplicationContext context) throws Exception {
JobLauncher jobLauncher = context.getBean(JobLauncher.class);
Job job = context.getBean(Job.class);
JobExecution jobExecution = jobLauncher.run(job, new JobParameters());
JobExplorer jobExplorer = context.getBean(JobExplorer.class);
while (jobExplorer.getJobExecution(jobExecution.getJobId()).isRunning()) {
Thread.sleep(1000);
}
}
@EnableBatchProcessing
@Configuration
public static class CustomMappingConfiguration {
@Bean
public ListItemReader<Map<Object, Object>> itemWriter() {
List<Map<Object, Object>> list = new ArrayList<>(5);
addNameToReaderList(list, "Jane");
addNameToReaderList(list, "John");
addNameToReaderList(list, "Liz");
addNameToReaderList(list, "Cameron");
addNameToReaderList(list, "Judy");
return new ListItemReader<>(list);
}
private void addNameToReaderList(List<Map<Object, Object>> itemReaderList,
String value) {
Map<Object, Object> prepMap = new HashMap<>();
prepMap.put("first_name", value);
itemReaderList.add(prepMap);
}
}
}