Introducing KafkaItemReader for single step jobs

resolves #679

Rebase update

rebase update pom cleanup

Cleaned up tests

Updated

Added updates based on code review

polish

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

@@ -13,6 +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>
 	</properties>
 
 	<dependencies>
@@ -31,7 +32,6 @@
 			<optional>true</optional>
 			<version>${spring-boot.version}</version>
 		</dependency>
-
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-test</artifactId>
@@ -82,6 +82,47 @@
 			<groupId>com.fasterxml.jackson.core</groupId>
 			<artifactId>jackson-annotations</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-binder-kafka</artifactId>
+			<version>3.1.0-SNAPSHOT</version>
+		</dependency>
+		<dependency>
+			<groupId>org.junit.jupiter</groupId>
+			<artifactId>junit-jupiter</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.junit.jupiter</groupId>
+			<artifactId>junit-jupiter-engine</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.junit.jupiter</groupId>
+			<artifactId>junit-jupiter-params</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<version>${spring-kafka-version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<version>${spring-kafka-version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.junit.jupiter</groupId>
+			<artifactId>junit-jupiter-api</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.junit.platform</groupId>
+			<artifactId>junit-platform-launcher</artifactId>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 
 </project>

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<Object, Map<Object, Object>> 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<Object, Map<Object, Object>>()
+				.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");
+		}
+	}
+
+}

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<Integer> 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<Integer> 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<Integer> 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;
+	}
+}

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

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<Map<Object, Object>> 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<Map<Object, Object>> writtenItems = itemWriter.getWrittenItems();
+		assertThat(writtenItems.size()).isEqualTo(4);
+		List<Object> 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<String, Object> configps = new HashMap<>(
+				KafkaTestUtils.producerProps(embeddedKafkaBroker));
+		Producer<String, Object> producer = new DefaultKafkaProducerFactory<>(configps,
+				new StringSerializer(), new JsonSerializer<>()).createProducer();
+		Map<Object, Object> 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<Map<Object, Object>> itemWriter() {
+			return new ListItemWriter<>();
+		}
+	}
+
+}