diff --git a/spring-cloud-starter-single-step-batch-job/pom.xml b/spring-cloud-starter-single-step-batch-job/pom.xml index 201a66b2..7547e60e 100644 --- a/spring-cloud-starter-single-step-batch-job/pom.xml +++ b/spring-cloud-starter-single-step-batch-job/pom.xml @@ -13,7 +13,7 @@ 1.15.0 1.15.0 1.0.8 - 2.5.3.RELEASE + 2.6.3 diff --git a/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterAutoConfiguration.java b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterAutoConfiguration.java new file mode 100644 index 00000000..d7e16251 --- /dev/null +++ b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterAutoConfiguration.java @@ -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> kafkaItemWriter( + KafkaItemWriterProperties kafkaItemWriterProperties, + ProducerFactory> producerFactory, + @Qualifier("batchItemKeyMapper") Converter itemKeyMapper) { + + validateProperties(kafkaItemWriterProperties); + KafkaTemplate template = new KafkaTemplate(producerFactory); + template.setDefaultTopic(kafkaItemWriterProperties.getTopic()); + return new KafkaItemWriterBuilder>() + .delete(kafkaItemWriterProperties.isDelete()).kafkaTemplate(template) + .itemKeyMapper(itemKeyMapper).build(); + } + + @Bean + @ConditionalOnMissingBean(name = "batchItemKeyMapper") + public Converter batchItemKeyMapper() { + return new Converter() { + @Override + public Object convert(Object source) { + return source; + } + }; + } + + @Bean + @ConditionalOnMissingBean + ProducerFactory> producerFactory() { + Map configs = new HashMap<>(); + configs.putAll(this.kafkaProperties.getProducer().buildProperties()); + return new DefaultKafkaProducerFactory>(configs, null, + new JsonSerializer<>()); + } + + private void validateProperties(KafkaItemWriterProperties kafkaItemWriterProperties) { + Assert.hasText(kafkaItemWriterProperties.getTopic(), + "topic must not be empty or null"); + } + +} diff --git a/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterProperties.java b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterProperties.java new file mode 100644 index 00000000..9b75be30 --- /dev/null +++ b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterProperties.java @@ -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; + } + +} diff --git a/spring-cloud-starter-single-step-batch-job/src/main/resources/META-INF/spring.factories b/spring-cloud-starter-single-step-batch-job/src/main/resources/META-INF/spring.factories index 52cc0f44..bd226b89 100644 --- a/spring-cloud-starter-single-step-batch-job/src/main/resources/META-INF/spring.factories +++ b/spring-cloud-starter-single-step-batch-job/src/main/resources/META-INF/spring.factories @@ -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 + diff --git a/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterTests.java b/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterTests.java new file mode 100644 index 00000000..6f218868 --- /dev/null +++ b/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemWriterTests.java @@ -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 configs = new HashMap<>( + KafkaTestUtils.consumerProps("1", "false", embeddedKafkaBroker)); + Consumer consumer = new DefaultKafkaConsumerFactory<>(configs, + new StringDeserializer(), new JsonDeserializer<>()).createConsumer(); + consumer.subscribe(singleton(topicName)); + + ConsumerRecords consumerRecords = KafkaTestUtils + .getRecords(consumer); + assertThat(consumerRecords.count()).isEqualTo(5); + List> result = new ArrayList<>(); + consumerRecords.forEach(cs -> { + result.add((Map) cs.value()); + }); + List 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> itemWriter() { + List> 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> itemReaderList, + String value) { + Map prepMap = new HashMap<>(); + prepMap.put("first_name", value); + itemReaderList.add(prepMap); + } + + } + +}