diff --git a/spring-cloud-starter-single-step-batch-job/pom.xml b/spring-cloud-starter-single-step-batch-job/pom.xml index bcb1ddfd..201a66b2 100644 --- a/spring-cloud-starter-single-step-batch-job/pom.xml +++ b/spring-cloud-starter-single-step-batch-job/pom.xml @@ -13,6 +13,7 @@ 1.15.0 1.15.0 1.0.8 + 2.5.3.RELEASE @@ -31,7 +32,6 @@ true ${spring-boot.version} - org.springframework.boot spring-boot-starter-test @@ -82,6 +82,47 @@ com.fasterxml.jackson.core jackson-annotations + + org.springframework.cloud + spring-cloud-stream-binder-kafka + 3.1.0-SNAPSHOT + + + org.junit.jupiter + junit-jupiter + test + + + org.junit.jupiter + junit-jupiter-engine + test + + + org.junit.jupiter + junit-jupiter-params + test + + + org.springframework.kafka + spring-kafka-test + ${spring-kafka-version} + test + + + org.springframework.kafka + spring-kafka + ${spring-kafka-version} + + + org.junit.jupiter + junit-jupiter-api + test + + + org.junit.platform + junit-platform-launcher + test + diff --git a/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderAutoConfiguration.java b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderAutoConfiguration.java new file mode 100644 index 00000000..c91b7473 --- /dev/null +++ b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderAutoConfiguration.java @@ -0,0 +1,90 @@ +/* + * 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.time.Duration; +import java.util.ArrayList; +import java.util.Map; +import java.util.Properties; + +import org.springframework.batch.item.kafka.KafkaItemReader; +import org.springframework.batch.item.kafka.builder.KafkaItemReaderBuilder; +import org.springframework.beans.factory.annotation.Autowired; +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.util.StringUtils; + +/** + * + * AutoConfiguration for a {@code KafkaItemReader}. + * + * @author Glenn Renfro + * @since 2.3 + */ +@Configuration +@EnableConfigurationProperties({ KafkaProperties.class, KafkaItemReaderProperties.class }) +@AutoConfigureAfter(BatchAutoConfiguration.class) +public class KafkaItemReaderAutoConfiguration { + + @Autowired + private KafkaProperties kafkaProperties; + + @Bean + @ConditionalOnMissingBean + @ConditionalOnProperty(prefix = "spring.batch.job.kafkaitemreader", name = "name") + public KafkaItemReader> kafkaItemReader( + KafkaItemReaderProperties kafkaItemReaderProperties) { + Properties consumerProperties = new Properties(); + consumerProperties.putAll(this.kafkaProperties.getConsumer().buildProperties()); + validateProperties(kafkaItemReaderProperties); + if (kafkaItemReaderProperties.getPartitions() == null + || kafkaItemReaderProperties.getPartitions().size() == 0) { + kafkaItemReaderProperties.setPartitions(new ArrayList<>(1)); + kafkaItemReaderProperties.getPartitions().add(0); + } + return new KafkaItemReaderBuilder>() + .partitions(kafkaItemReaderProperties.getPartitions()) + .consumerProperties(consumerProperties) + .name(kafkaItemReaderProperties.getName()) + .pollTimeout(Duration + .ofSeconds(kafkaItemReaderProperties.getPollTimeOutInSeconds())) + .saveState(kafkaItemReaderProperties.isSaveState()).topic(kafkaItemReaderProperties.getTopic()).build(); + } + + private void validateProperties(KafkaItemReaderProperties kafkaItemReaderProperties) { + if (!StringUtils.hasText(kafkaItemReaderProperties.getName())) { + throw new IllegalArgumentException("Name must not be empty or null"); + } + if (!StringUtils.hasText(kafkaItemReaderProperties.getTopic())) { + throw new IllegalArgumentException("Topic must not be empty or null"); + } + if (!StringUtils.hasText(this.kafkaProperties.getConsumer().getGroupId())) { + throw new IllegalArgumentException("GroupId must not be empty or null"); + } + if (this.kafkaProperties.getBootstrapServers() == null + || this.kafkaProperties.getBootstrapServers().size() == 0) { + throw new IllegalArgumentException("Bootstrap Servers must be configured"); + } + } + +} diff --git a/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderProperties.java b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderProperties.java new file mode 100644 index 00000000..1f08150c --- /dev/null +++ b/spring-cloud-starter-single-step-batch-job/src/main/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderProperties.java @@ -0,0 +1,130 @@ +/* + * 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.List; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +/** + * Properties to configure a {@code KafkaItemReader}. + * + * @author Glenn Renfro + * @since 2.3 + */ +@ConfigurationProperties(prefix = "spring.batch.job.kafkaitemreader") +public class KafkaItemReaderProperties { + + private String name; + + private String topic; + + private List partitions = new ArrayList<>(); + + private long pollTimeOutInSeconds = 30L; + + private boolean saveState = true; + + /** + * Returns the configured value of the name used to calculate {@code ExecutionContext} + * keys. + * @return the name + */ + public String getName() { + return name; + } + + /** + * The name used to calculate the key within the + * {@link org.springframework.batch.item.ExecutionContext}. + * @param name name of the writer instance + * @see org.springframework.batch.item.ItemStreamSupport#setName(String) + */ + public void setName(String name) { + this.name = name; + } + + /** + * Returns the name of the topic from which messages will be read. + * @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; + } + + /** + * A list of partitions to manually assign to the consumer. Defaults to a single entry + * value of 1. + * @return the list of partitions. + */ + public List getPartitions() { + return partitions; + } + + /** + * A list of partitions to manually assign to the consumer. Defaults to a single entry + * value of 1. + * @param partitions list of partitions + */ + public void setPartitions(List partitions) { + this.partitions = partitions; + } + + /** + * Get the pollTimeout for the poll() operations. Defaults to 30 seconds. + * @return long containing the poll timeout. + */ + public long getPollTimeOutInSeconds() { + return pollTimeOutInSeconds; + } + + /** + * Set the pollTimeout for the poll() operations. Defaults to 30 seconds. + * @param pollTimeOutInSeconds the number of seconds to wait before timing out. + */ + public void setPollTimeOutInSeconds(long pollTimeOutInSeconds) { + this.pollTimeOutInSeconds = pollTimeOutInSeconds; + } + /** + * Configure if the state of the {@link org.springframework.batch.item.ItemStreamSupport} + * should be persisted within the {@link org.springframework.batch.item.ExecutionContext} + * for restart purposes. Defaults to true. + * @return current status of the saveState flag. + */ + public boolean isSaveState() { + return saveState; + } + + /** + * Configure if the state of the {@link org.springframework.batch.item.ItemStreamSupport} + * should be persisted within the {@link org.springframework.batch.item.ExecutionContext} + * for restart purposes. + * @param saveState true if state should be persisted. Defaults to true. + */ + public void setSaveState(boolean saveState) { + this.saveState = saveState; + } +} 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 6938add2..52cc0f44 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 @@ -5,4 +5,5 @@ org.springframework.boot.autoconfigure.EnableAutoConfiguration=org.springframewo org.springframework.cloud.task.batch.autoconfigure.jdbc.JdbcItemWriterAutoConfiguration, \ 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.rabbit.AmqpItemWriterAutoConfiguration, \ + org.springframework.cloud.task.batch.autoconfigure.kafka.KafkaItemReaderAutoConfiguration diff --git a/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderAutoConfigurationTests.java b/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderAutoConfigurationTests.java new file mode 100644 index 00000000..4c90230a --- /dev/null +++ b/spring-cloud-starter-single-step-batch-job/src/test/java/org/springframework/cloud/task/batch/autoconfigure/kafka/KafkaItemReaderAutoConfigurationTests.java @@ -0,0 +1,240 @@ +/* + * 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.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.StringSerializer; +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.ListItemWriter; +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.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.support.serializer.JsonDeserializer; +import org.springframework.kafka.support.serializer.JsonSerializer; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.kafka.test.utils.KafkaTestUtils; + +import static org.assertj.core.api.Assertions.assertThat; + +@EmbeddedKafka(partitions = 1, topics = { "test" }) +public class KafkaItemReaderAutoConfigurationTests { + + private static EmbeddedKafkaBroker embeddedKafkaBroker; + + @BeforeAll + public static void setupTest(EmbeddedKafkaBroker embeddedKafka) { + embeddedKafkaBroker = embeddedKafka; + embeddedKafka.addTopics(new NewTopic("topic1", 1, (short) 1), + new NewTopic("topic2", 2, (short) 1), + new NewTopic("topic3", 1, (short) 1)); + } + + @Test + public void testBaseKafkaItemReader() { + final String topicName = "topic1"; + populateSingleTopic(topicName); + ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner() + .withUserConfiguration(CustomMappingConfiguration.class) + .withConfiguration( + AutoConfigurations.of(PropertyPlaceholderAutoConfiguration.class, + BatchAutoConfiguration.class, + SingleStepJobAutoConfiguration.class, + KafkaItemReaderAutoConfiguration.class)) + .withPropertyValues("spring.batch.job.jobName=job", + "spring.batch.job.stepName=step1", "spring.batch.job.chunkSize=5", + "spring.kafka.consumer.bootstrap-servers=" + + embeddedKafkaBroker.getBrokersAsString(), + "spring.kafka.consumer.group-id=1", + "spring.batch.job.kafkaitemreader.name=kafkaItemReader", + "spring.batch.job.kafkaitemreader.poll-time-out-in-seconds=2", + "spring.batch.job.kafkaitemreader.topic=" + topicName, + "spring.kafka.consumer.value-deserializer=" + + JsonDeserializer.class.getName()); + + applicationContextRunner.run((context) -> { + JobLauncher jobLauncher = context.getBean(JobLauncher.class); + + Job job = context.getBean(Job.class); + + ListItemWriter itemWriter = context.getBean(ListItemWriter.class); + + JobExecution jobExecution = jobLauncher.run(job, new JobParameters()); + + JobExplorer jobExplorer = context.getBean(JobExplorer.class); + + while (jobExplorer.getJobExecution(jobExecution.getJobId()).isRunning()) { + Thread.sleep(1000); + } + + List> writtenItems = itemWriter.getWrittenItems(); + + assertThat(writtenItems.size()).isEqualTo(4); + assertThat(writtenItems.get(0).get("first_name")).isEqualTo("jane"); + assertThat(writtenItems.get(1).get("first_name")).isEqualTo("john"); + assertThat(writtenItems.get(2).get("first_name")).isEqualTo("susan"); + assertThat(writtenItems.get(3).get("first_name")).isEqualTo("jim"); + }); + } + + @Test + public void testBaseKafkaItemReaderMultiplePartitions() { + final String topicName = "topic2"; + populateSingleTopic(topicName); + ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner() + .withUserConfiguration(CustomMappingConfiguration.class) + .withConfiguration( + AutoConfigurations.of(PropertyPlaceholderAutoConfiguration.class, + BatchAutoConfiguration.class, + SingleStepJobAutoConfiguration.class, + KafkaItemReaderAutoConfiguration.class)) + .withPropertyValues("spring.batch.job.jobName=job", + "spring.batch.job.stepName=step1", "spring.batch.job.chunkSize=5", + "spring.kafka.consumer.bootstrap-servers=" + + embeddedKafkaBroker.getBrokersAsString(), + "spring.kafka.consumer.group-id=1", + "spring.batch.job.kafkaitemreader.name=kafkaItemReader", + "spring.batch.job.kafkaitemreader.partitions=0,1", + "spring.batch.job.kafkaitemreader.poll-time-out-in-seconds=2", + "spring.batch.job.kafkaitemreader.topic=" + topicName, + "spring.kafka.consumer.value-deserializer=" + + JsonDeserializer.class.getName()); + + applicationContextRunner.run((context) -> { + JobLauncher jobLauncher = context.getBean(JobLauncher.class); + + Job job = context.getBean(Job.class); + + ListItemWriter itemWriter = context.getBean(ListItemWriter.class); + + JobExecution jobExecution = jobLauncher.run(job, new JobParameters()); + + JobExplorer jobExplorer = context.getBean(JobExplorer.class); + + while (jobExplorer.getJobExecution(jobExecution.getJobId()).isRunning()) { + Thread.sleep(1000); + } + + basicValidation(itemWriter); + }); + } + + @Test + public void testBaseKafkaItemReaderPollTimeoutDefault() { + final String topicName = "topic3"; + populateSingleTopic(topicName); + ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner() + .withUserConfiguration(CustomMappingConfiguration.class) + .withConfiguration( + AutoConfigurations.of(PropertyPlaceholderAutoConfiguration.class, + BatchAutoConfiguration.class, + SingleStepJobAutoConfiguration.class, + KafkaItemReaderAutoConfiguration.class)) + .withPropertyValues("spring.batch.job.jobName=job", + "spring.batch.job.stepName=step1", "spring.batch.job.chunkSize=5", + "spring.kafka.consumer.bootstrap-servers=" + + embeddedKafkaBroker.getBrokersAsString(), + "spring.kafka.consumer.group-id=1", + "spring.batch.job.kafkaitemreader.name=kafkaItemReader", + "spring.batch.job.kafkaitemreader.topic=" + topicName, + "spring.kafka.consumer.value-deserializer=" + + JsonDeserializer.class.getName()); + Date startTime = new Date(); + applicationContextRunner.run((context) -> { + JobLauncher jobLauncher = context.getBean(JobLauncher.class); + + Job job = context.getBean(Job.class); + + ListItemWriter itemWriter = context.getBean(ListItemWriter.class); + + JobExecution jobExecution = jobLauncher.run(job, new JobParameters()); + + JobExplorer jobExplorer = context.getBean(JobExplorer.class); + + while (jobExplorer.getJobExecution(jobExecution.getJobId()).isRunning()) { + Thread.sleep(1000); + } + Date endTime = new Date(); + long seconds = (endTime.getTime() - startTime.getTime()) / 1000; + assertThat(seconds).isGreaterThanOrEqualTo(30); + basicValidation(itemWriter); + }); + } + + private void basicValidation(ListItemWriter itemWriter) { + List> writtenItems = itemWriter.getWrittenItems(); + assertThat(writtenItems.size()).isEqualTo(4); + List results = new ArrayList<>(); + for (int i = 0; i < 4; i++) { + results.add(writtenItems.get(i).get("first_name")); + } + + assertThat(results).contains("jane", "john", "susan", "jim"); + } + + private void populateSingleTopic(String topic) { + Map configps = new HashMap<>( + KafkaTestUtils.producerProps(embeddedKafkaBroker)); + Producer producer = new DefaultKafkaProducerFactory<>(configps, + new StringSerializer(), new JsonSerializer<>()).createProducer(); + Map testMap = new HashMap<>(); + testMap.put("first_name", "jane"); + producer.send(new ProducerRecord<>(topic, "my-aggregate-id", testMap)); + testMap = new HashMap<>(); + testMap.put("first_name", "john"); + producer.send(new ProducerRecord<>(topic, "my-aggregate-id", testMap)); + testMap = new HashMap<>(); + testMap.put("first_name", "susan"); + producer.send(new ProducerRecord<>(topic, "my-aggregate-id", testMap)); + testMap = new HashMap<>(); + testMap.put("first_name", "jim"); + producer.send(new ProducerRecord<>(topic, "my-aggregate-id", testMap)); + producer.flush(); + producer.close(); + } + + @EnableBatchProcessing + @Configuration + public static class CustomMappingConfiguration { + @Bean + public ListItemWriter> itemWriter() { + return new ListItemWriter<>(); + } + } + +}