Contribute to Spring Batch open source

Issue: StepBuilder - issue when setting the taskExecutor before faultTolerant() PR: Add AbstractTaskletStepBuilder copy constructor

Introduction

I participated in contributing to ``spring-batch``` through the open source contribution study. The goal of issue selection was to participate in open source for the first time at the time, so Let's not be greedy and contribute to easy yet important open source started with a simple goal. By participating in the GDG open source study, I was able to do PR within the deadline by observing the PR participation of the study operator and other study participants. (The merge has not been reflected yet, but has been confirmed by the maintainer.)

Brief description of Spring batch

Everyone knows Spring because it is a very famous open source, but it seems that there are many people who have not used it unless it is a project that uses spring-batch's batch job. I knew about it, but it was my first time seeing internal open source as open source. Spring batch is a lightweight, comprehensive batch framework used for processing large amounts of data. First, step includes components such as ItemReader, ItemProcessor, and ItemWriter, which are responsible for performing specific tasks. It allows processing of data chunk by chunk and constitutes a job with one or more step. job is considered as ``JobInstance as the execution unit of batch processing, and is distinguished by ````JobParameter. JobExecution contains records of execution.

Issue details

Issue was simple and the test code was well written, making it easy to understand at once. When creating a step, you may notice that field values ​​are not inherited well depending on the location of taskExecutor().

TaskletStep step1 = new StepBuilder("step-name", jobRepository)
.chunk(10, transactionManager)
.reader(itemReader)
.processor(itemProcessor)
.writer(itemWriter)
.faultTolerant()
.taskExecutor(taskExecutor)
.build();
TaskletStep step2 = new StepBuilder("step-name", jobRepository)
.chunk(10, transactionManager)
.taskExecutor(taskExecutor)// The task executor is set before faultTolerant()
.reader(itemReader)
.processor(itemProcessor)
.writer(itemWriter)
.faultTolerant()
.build();

Open source analysis

The process of creating TaskletStep can be represented in the following structure diagram.

StepBuilderHelper
|
|-- AbstractTaskletStepBuilder
|
|-- SimpleStepBuilder
|
|-- FaultTolerantStepBuilder
|
|-- TaskletStep

When creating a TaskletStep with StepBuilder```, ```.faultTolerant()``` is influenced by ```FaultTolerantStepBuilder```, and .taskExecutor(taskExecutor)is a method ofAbstractTaskletStepBuilder. By applying faultTolerant(), when TaskletStep``` is created, it will include retry and skip functions for failed items when implementing a chunk-oriented system.

// FaultTolerantStepBuilder faultTolerant()
public FaultTolerantStepBuilder<I, O> faultTolerant() {
return new FaultTolerantStepBuilder<>(this);
}

taskExecutor(taskExecutor) allows you to change the taskExecutor parameter among the AbstractTaskletStepBuilder properties and return a ````SimpleStepBuilderinstance. Here,Brepresents the type ofAbstractTaskletStepBuilder``` or its subclass.

// AbstractTaskletStepBuilder taskExecutor(taskExecutor)
public B taskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
return self();
}

If you look at the constructor of SimpleStepBuilder, you can see that two methods are used: receiving the properties of the parent object to create a new object or creating a copy constructor.

public SimpleStepBuilder(StepBuilderHelper<?> parent) {
super(parent);
}
protected SimpleStepBuilder(SimpleStepBuilder<I, O> parent) {
super(parent);
this.chunkSize = parent.chunkSize;
this.completionPolicy = parent.completionPolicy;
this.chunkOperations = parent.chunkOperations;
this.reader = parent.reader;
this.writer = parent.writer;
this.processor = parent.processor;
this.itemListeners = parent.itemListeners;
this.readerTransactionalQueue = parent.readerTransactionalQueue;
this.meterRegistry = parent.meterRegistry;
}

However, you can see that there is no copy constructor for AbstractTaskletStepBuilder, and there is only a super() processing constructor that simply receives the parent object properties.

public AbstractTaskletStepBuilder(StepBuilderHelper<?> parent) {
super(parent);
}

Solution

Field update was made possible by adding the copy constructor of ```AbstractTaskletStepBuilder````.

public AbstractTaskletStepBuilder(AbstractTaskletStepBuilder<?> parent) {
super(parent);
this.chunkListeners = parent.chunkListeners;
this.stepOperations = parent.stepOperations;
this.transactionManager = parent.transactionManager;
this.transactionAttribute = parent.transactionAttribute;
this.streams.addAll(parent.streams);
this.exceptionHandler = parent.exceptionHandler;
this.throttleLimit = parent.throttleLimit;
this.taskExecutor = parent.taskExecutor;
}

The test code was written as follows. You can write the test code according to the settings set by looking at the contribute``` file or ````readme```` content in the open source contribution. ```@BeforeEach``` creates a ```simpleStepBuilder``` to be commonly used in testing. For code modification, ```AbstractTaskletStepBuilder``` itself is an abstract class, so ```simpleStepBuilder``` was used as an implementation for comparison. I wrote test code to check if the copy constructor works well and to compare the values ​​when ```taskExecutor()``` is done first and ```faultTolerant()``` is done afterwards. Here, because the property values ​​of ```SimpleStepBuilder``` are privateand difficult to access, we used **reflection**, a Java API. <span style="background-color:#fff5b1">Reflection is a Java API that allows operations such as querying class information, accessing object fields or methods, or dynamically creating class objects at runtime.</span> You can adjust the access of the declare field in the same way asfield.setAccessible(true). (accessPrivateField``` method)

@SpringBatchTest
@SpringJUnitConfig
public class AbstractTaskletStepBuilderTests {
private final JobRepository jobRepository = mock(JobRepository.class);
private final int chunkSize = 10;
private final ItemReader itemReader = mock(ItemReader.class);
private final ItemProcessor itemProcessor = mock(ItemProcessor.class);
private final ItemWriter itemWriter = mock(ItemWriter.class);
private final SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
SimpleStepBuilder simpleStepBuilder;
private <T> T accessPrivateField(Object o, String fieldName) throws ReflectiveOperationException {
Field field = o.getClass().getDeclaredField(fieldName);
field.setAccessible(true);
return (T) field. get(o);
}
private <T> T accessSuperClassPrivateField(Object o, String fieldName) throws ReflectiveOperationException {
Field field = o.getClass().getSuperclass().getDeclaredField(fieldName);
field.setAccessible(true);
return (T) field. get(o);
}
// Create simpleStepBuilder for common use
@BeforeEach
void set(){
StepBuilderHelper stepBuilderHelper = new StepBuilderHelper("test", jobRepository) {
@Override
protected StepBuilderHelper self() {
return null;
}
};
simpleStepBuilder = new SimpleStepBuilder(stepBuilderHelper);
simpleStepBuilder.chunk(chunkSize);
simpleStepBuilder.reader(itemReader);
simpleStepBuilder.processor(itemProcessor);
simpleStepBuilder.writer(itemWriter);
}
// 복사가 잘되는지 확인
@Test
void copyConstractorTest() throws ReflectiveOperationException {
Constructor<SimpleStepBuilder> constructor = SimpleStepBuilder.class.getDeclaredConstructor(SimpleStepBuilder.class);
constructor.setAccessible(true);
SimpleStepBuilder copySimpleStepBuilder = constructor.newInstance(simpleStepBuilder);
int copyChunkSize = accessPrivateField(copySimpleStepBuilder, "chunkSize");
ItemReader copyItemReader = accessPrivateField(copySimpleStepBuilder, "reader");
ItemProcessor copyItemProcessor = accessPrivateField(copySimpleStepBuilder, "processor");
ItemWriter copyItemWriter = accessPrivateField(copySimpleStepBuilder, "writer");
assertEquals(chunkSize, copyChunkSize);
assertEquals(itemReader, copyItemReader);
assertEquals(itemProcessor, copyItemProcessor);
assertEquals(itemWriter, copyItemWriter);
}
// taskExecutor를 먼저하고 faultTolerant를 이후에 했을 때 값 비교
@Test
void faultTolerantMethodTest() throws ReflectiveOperationException {
simpleStepBuilder.taskExecutor(taskExecutor); // The task executor is set before faultTolerant()
simpleStepBuilder.faultTolerant();
int afterChunkSize = accessPrivateField(simpleStepBuilder, "chunkSize");
ItemReader afterItemReader = accessPrivateField(simpleStepBuilder, "reader");
ItemProcessor afterItemProcessor = accessPrivateField(simpleStepBuilder, "processor");
ItemWriter afterItemWriter = accessPrivateField(simpleStepBuilder, "writer");
TaskExecutor afterTaskExecutor = accessSuperClassPrivateField(simpleStepBuilder, "taskExecutor");
assertEquals(chunkSize, afterChunkSize);
assertEquals(itemReader, afterItemReader);
assertEquals(itemProcessor, afterItemProcessor);
assertEquals(itemWriter, afterItemWriter);
assertEquals(taskExecutor, afterTaskExecutor);
}
}

Study Retrospective

In fact, I had only thought about contributing to open source, but had never tried it, but through this simple open source, I thought I would like to participate in other open source projects, starting with this one. And there was some psychological pressure about open source itself, but it was nice to be able to contribute as if I were actually participating in the project. If it weren't for the study, if I hadn't been with other people, I think my attempt would have been delayed, but I was grateful to have been able to participate in such a great opportunity. In conclusion, ``spring-batch``` was a bonanza with a lot to contribute!

Spring Batch Contributor!! I received feedback from the maintainer almost 5 months later, and with the help of Inje, who helped me, and Taeik, who suggested corrections to the contribution, I was able to reupload and close the pr. In fact, it was not a part where I had to write a lot of code, but it was a part where the problem of intermediate data loss occurred, so I was very proud of fixing the core bug, and I realized that I could approach it this way and solve the problem by communicating like this. That was the moment I found out. If I had any regrets, no matter how much I thought about it, I could only think of the limited method suggested in the issue at the time, so I wrote the test according to the issue. The direction for writing the test part in the middle was that the maintainer's request could not be satisfactorily modified in a way to make it easier to maintain. Previously, the method was to check whether all values ​​were copied properly, but you can see that the maintainer changed the test by changing the method to compare stepOperations as shown below. (In spring-batch, there are many things that are treated as private, so many parts of the test code are actually written in other tests by simply taking the values with ReflectionTestUtils and comparing them.)

Object stepOperations = ReflectionTestUtils.getField(step, "stepOperations");
assertInstanceOf(TaskExecutorRepeatTemplate.class, stepOperations);