Creating an advanced application with Spring Batch

Technologies that we will use during the implementation of the app:

• Java 11
• Spring 5
• Spring Batch
• PostgreSQL

docker run --name postgres -e POSTGRES_PASSWORD=your_pass -p 5438:5432 -d postgres

What we did? Calling this command

• pulls the postgres Docker image from Docker Hub,
• sets the POSTGRES_PASSWORD environment variable value to your_pass,
• names (--name) the Docker container to be postgres,
• maps container’s internal 5432 to external 5438 port (it is good practice to set the same ports, however on my computer port 5432 is already in use),
• enables to run the Docker container in the background (-d)

Now you can connect to the database using the selected database client (I prefer DataGrip or Valentina Studio) or even with the console. Now we move on to the design of the database. Everything will be done in one schema, so we don’t need to create additional schemas. We will put all the necessary tables, triggers and other functions into the public schema.

At the first stage, we will put the data from the CSV file into the database. To do this, we first need to establish specific tables in the database. Below are the tables which I’ve added.

Our database model begins with the CLUB table, which contains basic information including name, id, email, phone etc. It also has a field for the user to indicate their preference for what type of communication should be used to contact them (club can choose type of communication – email or phone).

create table club
(
   club_id serial
      constraint club_pk
         primary key,
   name varchar(200) not null,
   city varchar(100) not null,
   email_address varchar(100),
   phone char(10),
   is_notified char(1) not null,
   year_of_foundation int
);

Each club has relation to the account table. Club table has many-to-many relationships with the Account table, because Club could have many different Accounts while each Account could have many Clubs. As you can see, the Account table has only few fields – id, current balance and the timestamp of the last executed statement.

create table account
(
	account_id serial,
	balance float not null,
	last_statement_timestamp timestamp
);

create unique index account_account_id_uindex
	on account (account_id);

alter table account
	add constraint account_pk
		primary key (account_id);

create table club_account
(
	club_club_id int not null
		constraint club_account_pk
			primary key
		constraint club_account_club_club_id_fk
			references club,
	account_account_id int not null
		constraint club_account_account_account_id_fk
			references account
);

The last table we need is the Transaction table. It contains fields describing the most important parameters for each transaction. It’s simply to understand that each Account table has a one-to-many relationships with the Transaction table.

create table transaction
(
	transaction_id serial,
	account_id int not null
		constraint transaction_pk
			primary key
		constraint transaction_account_account_id_fk
			references account,
	credit float,
	debit float,
	creation_timestamp timestamp not null
);

create unique index transaction_transaction_id_uindex
	on transaction (transaction_id);

In order to import the club data from the CSV file, we need to understand the specific file format which will be containing all information. In that case, the file consists of three record types:
– for updating name and foundation year fields,
– for updating address fields,
– for updating club communication mechanism.

Each subsequent record should consist of the same number of attributes belonging to a given table. Our task will be to properly transform the received data so that the database can be properly updated. It’s important to note that it is expected that the specific club exist in the database prior to the update record coming through.

Each record type will be indicated in first parameter in the record. Let’s look at each of the three cases:

Record type for updating name and year of foundation:

1. Record type – this always be 1 for a record type 1,
2. Club ID – this is id of the club record, which will be updated,
3. Name – the name of club which should be updated in database (if blank, no update for year of foundation should be done)
4. Year of foundation – year of foundation club which should be updated in database (if blank, no update for year of foundation should be done)

Record type for updating address:

1. Record type – this always be 2 for a record type 2,
2. Club ID – this is id of the club record, which will be updated,
3. Address 1 – The first line of the address to be updated (if blank, no update for address should be done),
4. Address 2 – The optional second line of the address to be updated (if blank, no update for address should be done),
5. City – the city of the club (if blank, no update should be done)
6. State – the state of the club (if blank, no update should be done)
7. Postal code – postal code of the club (if blank, no update should be done)

Record type for updating club communication mechanism:

1. Record type – this always be 3 for a record type 3,
2. Club ID – this is id of the club record, which will be updated,
3. Email address – the email address to update (if blank, no update should be done),
4. Phone – the phone number to update (if blank, no update should be done),
5. Notification option – the notification preference (if blank, no update should be done).

Now let’s prepare the initial two files. The first one will contain the basic data that we want to put in our still empty database. All data will be saved in the database using SQL commands.

INSERT INTO club(name, address, email_address, phone, is_notified, year_of_foundation)
VALUES
       ('Manchester City', '83 Ducie St, Manchester M1 2JQ', 'manchester_city@gmail.com', '855100432', 'M', 1922),
       ('Paris Saint Germain', '24, Rue du Commandant-Guilbaud 75016 Paris', 'psg@gmail.com', '155101132', 'M', 1924),
       ('Real Madrid', 'Avenida de Concha Espina 1 Madrid', 'real@yahoo.com', '840177901', 'P', 1899),
       ('Legia Warszawa', 'Łazienkowska 3, 00-449 Warszawa', 'legia@interia.pl', '589101655', 'M', 1924),
       ('Czarni Deblin', 'Pułku Piechoty, 08-530 Dęblin', 'czarni@wp.pl', '732011854', 'P', 1931);

INSERT INTO account(balance, last_statement_timestamp)
VALUES
       (100000000, '2021-06-22 19:10:25-01'),
       (250000000, '2021-10-01 06:50:42-02'),
       (110000000, '2020-02-10 11:30:14-07'),
       (250000000, '2021-10-01 12:20:11-06'),
       (1950000, '2016-11-15 12:41:11-02'),
       (999000, '2019-08-22 14:41:21-06'),
       (50000, '2018-04-04 21:41:25-01');

INSERT INTO club_account(club_club_id, account_account_id)
VALUES
       (1, 1),
       (2, 2),
       (3, 4),
       (1, 3),
       (4, 5),
       (4, 6),
       (5, 7),
       (4, 7);

The second file will be in CSV format. This file will contain the records formatted in accordance with the above assumptions.

2,2,,,Paris,,11111
2,2,Rue du Commandant-Guilbaud,,Paris,,41013
3,1,,,,316-510-9138,P
3,4,legiawawa@google.com,785-790-7373,P
1,3,Real Madrid C.F,1902

Even before implementing the main code of our application, we need to configure the database connection to our application. We will do this by using the very popular syntax inside the application.properties file, which should be in the resources folder. Remember that the data must match the values provided when creating the database!

spring.datasource.driverClassName=org.postgresql.Driver
spring.datasource.url=jdbc:postgresql://localhost:5438/postgres
spring.datasource.username=postgres
spring.datasource.password=postgres
spring.batch.initialize-schema=always

To process file, we’ll begin by defining the job and its first step. First we will create a class responsible for the configuration part. The basic version is shown in the code snippet below.

@Configuration
public class ImportDataJobConfiguration {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    public ImportDataJobConfiguration(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
    }

    @Bean
    public Job importClubDataJob() throws Exception {
        return this.jobBuilderFactory
                .get("importClubDataJob")
                .start(importClubUpdates())
                .build();
    }

    @Bean
    public Step importClubDataUpdatesStep() throws Exception {
        return this.stepBuilderFactory
                .get("importClubUpdates")
                .<ClubUpdate, ClubUpdate>chunk(100)
                .reader(clubDataUpdateItemReader(null))
                .processor(clubDataValidatingItemProcessor(null))
                .writer(clubDataUpdateItemWriter())
    }
}

We begin by defining a Spring Configuration class (via the @Configuration) annotation. Then two dependencies of the basic elements of this class are drawn – we import two factories: JobBuilderFactory and StepBuilderFactory. If you cannot find these dependencies (e.g. the compiler signals an error) then make sure that you have added @EnableBatchProcessing annotation to the main class. With the builders wired in, we can define our Job and Step.

@Bean
@StepScope
public ItemReader<? extends ClubUpdate> clubDataUpdateItemReader(@Value("#{jobParameters['clubData']}") Resource inputFile)
        throws Exception {
    return new FlatFileItemReaderBuilder<ClubUpdate>()
            .name("clubDataUpdateItemReader")
            .resource(inputFile)
            .lineTokenizer(clubDataUpdatesLineTokenizer())
            .fieldSetMapper(clubDataUpdateFieldSetMapper())
            .build();
}

@Bean
public LineTokenizer clubDataUpdatesLineTokenizer() throws Exception {
    DelimitedLineTokenizer recordTypeOne = new DelimitedLineTokenizer();
    recordTypeOne.setNames("recordId", "clubId", "name", "yearOfFoundation");
    recordTypeOne.afterPropertiesSet();

    DelimitedLineTokenizer recordTypeTwo = new DelimitedLineTokenizer();
    recordTypeTwo.setNames("recordId", "clubId", "address1", "address2", "city", "state", "postalCode");
    recordTypeTwo.afterPropertiesSet();

    DelimitedLineTokenizer recordTypeThree = new DelimitedLineTokenizer();
    recordTypeThree.setNames("recordId", "clubId", "emailAddress", "phone", "notification");
    recordTypeThree.afterPropertiesSet();

    Map<String, LineTokenizer> tokenizerMap = new HashMap<>(3);
    tokenizerMap.put("1*", recordTypeOne);
    tokenizerMap.put("2*", recordTypeTwo);
    tokenizerMap.put("3*", recordTypeThree);

    PatternMatchingCompositeLineTokenizer tokenizer = new PatternMatchingCompositeLineTokenizer();
    tokenizer.setTokenizers(tokenizerMap);
    return tokenizer;
}

@Getter
@Setter
public class ClubDataUpdate {

    protected final long clubId;

    public ClubDataUpdate(long clubId) {
        this.clubId = clubId;
    }
}

@Getter
@Setter
public class ClubDataBaseUpdate extends ClubDataUpdate {

    private final String name;
    private final int yearOfFoundation;

    public ClubDataBaseUpdate(long clubId, String name, int yearOfFoundation) {
        super(clubId);
        this.name = name;
        this.yearOfFoundation = yearOfFoundation;
    }
}

@Getter
@Setter
public class ClubDataAddressUpdate extends ClubDataUpdate {

    private final String address1;
    private final String address2;
    private final String city;
    private final String state;
    private final String postalCode;

    public ClubDataAddressUpdate(final long clubId, final String address1, final String address2,
                                 final String city, final String state, final String postalCode) {
        super(clubId);
        this.address1 = StringUtils.hasText(address1) ? address1 : null;
        this.address2 = StringUtils.hasText(address2) ? address2 : null;
        this.city = StringUtils.hasText(city) ? city : null;
        this.state = StringUtils.hasText(state) ? state : null;
        this.postalCode = StringUtils.hasText(postalCode) ? postalCode : null;
    }
}

@Getter
@Setter
public class ClubDataContactUpdate extends ClubDataUpdate {

    private final String emailAddress;
    private final String phone;
    private final String notification;

    public ClubDataContactUpdate(final long clubId, String emailAddress, String phone, String notification) {
        super(clubId);
        this.emailAddress = StringUtils.hasText(emailAddress) ? emailAddress : null;
        this.phone = StringUtils.hasText(phone) ? phone : null;
        this.notification = StringUtils.hasText(notification) ? notification : null;
    }
}

@Bean
public FieldSetMapper<ClubDataUpdate> clubDataUpdateFieldSetMapper() {
    return fieldSet -> {
        switch (fieldSet.readInt("recordId")) {
            case 1:
                return new ClubDataBaseUpdate(
                        fieldSet.readLong("clubId"),
                        fieldSet.readString("name"),
                        fieldSet.readInt("yearOfFoundation"));
            case 2:
                return new ClubDataAddressUpdate(
                        fieldSet.readLong("clubId"),
                        fieldSet.readString("address1"),
                        fieldSet.readString("address2"),
                        fieldSet.readString("city"),
                        fieldSet.readString("state"),
                        fieldSet.readString("postalCode"));
            case 3:
                return new ClubDataContactUpdate(
                        fieldSet.readLong("clubId"),
                        fieldSet.readString("emailAddress"),
                        fieldSet.readString("phone"),
                        fieldSet.readString("notification"));
            default:
                throw new IllegalArgumentException("Invalid record type was found: " + fieldSet.readInt("recordId"));
        }
    };
}

@Component
public class ClubDataItemValidator implements Validator<ClubDataUpdate> {

    private final NamedParameterJdbcTemplate jdbcTemplate;

    private static final String FIND_CLUB_BY_ID = "SELECT count(*) FROM club WHERE club_id = :id";

    public ClubDataItemValidator(DataSource dataSource) {
        this.jdbcTemplate = new NamedParameterJdbcTemplate(dataSource);
    }

    @Override
    public void validate(ClubDataUpdate clubDataUpdate) throws ValidationException {
        Map<String, Long> map = Collections.singletonMap("id", clubDataUpdate.getClubId());
        Long count = jdbcTemplate.queryForObject(FIND_CLUB_BY_ID, map, Long.class);
        if (count == 0) {
            throw new ValidationException(String.format("Club with id %s has not found in DB", clubDataUpdate.getClubId()));
        }
    }
}

@Bean
public ItemProcessor<? super ClubDataUpdate, ? extends ClubDataUpdate> clubDataValidatingItemProcessor(
        ClubDataItemValidator validator) {
    ValidatingItemProcessor<ClubDataUpdate> clubDataUpdateValidatingItemProcessor = new ValidatingItemProcessor<>(validator);
    clubDataUpdateValidatingItemProcessor.setFilter(true);
    return clubDataUpdateValidatingItemProcessor;
}

The last element of this first three-part game with the batch process is updating the data directly in the database. Our functions should respond appropriately to the given values depending on the type of process (one of these three types of data updates). For this reason, you will need to build three different ItemWriter components that will be managed by the parent function. This function managing the selection of these items will be implemented using ClassifierCompositeItemWriter. In fact, all three ItemWriter methods will be similar to each other. The main difference between them will be the SQL procedures that will have to import the data to the appropriate places in the database. Below I’ve added examples of how to implement these three methods:

@Bean
public JdbcBatchItemWriter<ClubDataUpdate> clubDataBaseUpdateJdbcBatchItemWriter(DataSource dataSource) {
    return new JdbcBatchItemWriterBuilder<ClubDataUpdate>()
            .beanMapped()
            .sql("UPDATE club " +
                    "SET name = COALESCE(:name, name), " +
                    "year_of_foundation = COALESCE(:yearOfFoundation, year_of_foundation) " +
                    "WHERE club_id = :clubId")
            .dataSource(dataSource)
            .build();
}

@Bean
public JdbcBatchItemWriter<ClubDataUpdate> clubDataAddressUpdateJdbcBatchItemWriter(DataSource dataSource) {
    return new JdbcBatchItemWriterBuilder<ClubDataUpdate>()
            .beanMapped()
            .sql("UPDATE club " +
                    "SET address = COALESCE(:address1, address), " +
                    "WHERE club_id = :clubId")
            .dataSource(dataSource)
            .build();
}

@Bean
public JdbcBatchItemWriter<ClubDataUpdate> clubDataContactUpdateJdbcBatchItemWriter(DataSource dataSource) {
    return new JdbcBatchItemWriterBuilder<ClubDataUpdate>()
            .beanMapped()
            .sql("UPDATE club " +
                    "SET email_address = COALESCE(:emailAddress, email_address), " +
                    "phone = COALESCE(:phone, phone), " +
                    "is_notified = COALESCE(:notification, is_notified) " +
                    "WHERE club_id = :clubId")
            .dataSource(dataSource)
            .build();
}

public class ClubDataUpdateClassifier implements Classifier<ClubDataUpdate, ItemWriter<? super ClubDataUpdate>> {

    private final JdbcBatchItemWriter<ClubDataUpdate> recordTypeOneItemWriter;
    private final JdbcBatchItemWriter<ClubDataUpdate> recordTypeTwoItemWriter;
    private final JdbcBatchItemWriter<ClubDataUpdate> recordTypeThreeItemWriter;

    public ClubDataUpdateClassifier(JdbcBatchItemWriter<ClubDataUpdate> recordTypeOneItemWriter,
                                    JdbcBatchItemWriter<ClubDataUpdate> recordTypeTwoItemWriter,
                                    JdbcBatchItemWriter<ClubDataUpdate> recordTypeThreeItemWriter) {
        this.recordTypeOneItemWriter = recordTypeOneItemWriter;
        this.recordTypeTwoItemWriter = recordTypeTwoItemWriter;
        this.recordTypeThreeItemWriter = recordTypeThreeItemWriter;
    }

    @Override
    public ItemWriter<? super ClubDataUpdate> classify(ClubDataUpdate classifiable) {
        if (classifiable instanceof ClubDataBaseUpdate) {
            return recordTypeOneItemWriter;
        } else if (classifiable instanceof ClubDataAddressUpdate) {
            return recordTypeTwoItemWriter;
        } else if (classifiable instanceof ClubDataContactUpdate) {
            return recordTypeThreeItemWriter;
        } else {
            throw new IllegalArgumentException(String.format("Invalid type: %s", classifiable.getClass().getCanonicalName()));
        }
    }
}

@Bean
public ItemWriter<? super ClubDataUpdate> clubDataUpdateItemWriter() {
    ClubDataUpdateClassifier classifier = new ClubDataUpdateClassifier(
            clubDataBaseUpdateJdbcBatchItemWriter(null),
            clubDataAddressUpdateJdbcBatchItemWriter(null),
            clubDataContactUpdateJdbcBatchItemWriter(null));
   ClassifierCompositeItemWriter<ClubDataUpdate> compositeItemWriter = new ClassifierCompositeItemWriter<>();
    compositeItemWriter.setClassifier(classifier);
    return compositeItemWriter;
}

A job instance already exists and is complete for parameters={clubData=clubDataFile.csv}. If you want to run this job again, change the parameters.

If we have already prepared our data, we can try to handle several transactions made between clubs. Suppose the transfer season is about to begin and clubs start buying and selling their players. For this project, the full list of transactions carried out will be kept in an XML file (I changed the file format to show you different ways of dealing with data). In this part of the guide, we will try to import the changes contained in the file to the database and process the data accordingly.

The transaction file begins with a transactions element that wraps all of the individual transaction elements. Each transaction chunk represents a single bank transaction and will result in an item in our batch job. What was included in the file can now be mapped to a domain object inside the application.

@XmlRootElement(name = "transaction")
@NoArgsConstructor
@AllArgsConstructor
@Getter
@Setter
public class Transaction {

    private long transactionId;
    private long accountId;
    private String description;
    private BigDecimal credit;
    private BigDecimal debit;
    private Date timestamp;

    @XmlJavaTypeAdapter(JaxbDateSerializer.class)
    public void setTimestamp(Date timestamp) {
        this.timestamp = timestamp;
    }

    public BigDecimal getTransactionAmount() {
        if (credit != null) {
            if (debit != null) {
                return credit.add(debit);
            }
            else {
                return credit;
            }
        } else return Objects.requireNonNullElseGet(debit, () -> new BigDecimal(0));
    }
}

As you can see, in our Transaction class we have fields that map directly to the XML chunks (what we’ve added in xml file). The three significant items on this class are the @XmlRootElement, @XmlJavaTypeAdapter and the additional method getTransactionAmount(). The @XmlRootElement is a JAX-B annotation that defines what the root tag is for that domain object. In our case, it is the tag transaction. The @XmlJavaTypeAdapter is used on the setter for the timestamp field because JAX-B doesn’t have a nice and simple way to handle the conversion from a
String to a java.util.Date. Because of that, we need to provide a bit of code that JAX-B will use to do that conversion which is the JaxbDateSerializer. As usual, to minimize the amount of boilerplate code, I added the lombok annotation – these are related to constructor creation and the get and set methods.

public class JaxbDateSerializer extends XmlAdapter<String, Date> {

    private SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");

    @Override
    public Date unmarshal(String date) throws Exception {
        return dateFormat.parse(date);
    }

    @Override
    public String marshal(Date date) throws Exception {
        return dateFormat.format(date);
    }
}

This JaxbDateSerializer class will help us transform a date value (from string to date and from date to string). In addition, the previously mentioned getTransactionAmount() method, which is embedded inside the Transaction class, will help us calculate the final amounts that we operate during the transaction.

Let’s start the importing of our bank transactions by configuring the Step and adding it to our Job. We will use StaxEventItemReader to read XML files. In this step, we will compose two individual processes – reading first and then writing. ItemReader was implemented using a StaxEventItemReaderBuilder. While building, you should provide the name of the object, the location of the XML file placed in the parameters, the type of the main domain entity in the XML file and a reference to the unmarshaller.

@Bean
@StepScope
public StaxEventItemReader<Transaction> transactionItemReader(
        @Value("#{jobParameters['transactionFile']}") Resource transactionFile) {
    Jaxb2Marshaller unmarshaller = new Jaxb2Marshaller();
    unmarshaller.setClassesToBeBound(Transaction.class);
    return new StaxEventItemReaderBuilder<Transaction>()
            .name("transactionItemReader")
            .resource(transactionFile)
            .addFragmentRootElements("transaction")
            .unmarshaller(unmarshaller)
            .build();
}

Now that we have the prepared part of the code responsible for reading the file, now we can start saving transactions. We will create a new ItemWriter method called transactionItemWriter(). Inside this function I have built a simple SQL query that we will use to add new records in the database.

@Bean
public JdbcBatchItemWriter<Transaction> transactionItemWriter(DataSource dataSource) {
    return new JdbcBatchItemWriterBuilder<Transaction>()
            .dataSource(dataSource)
            .sql("INSERT INTO transaction(transaction_id, account_id, description, credit, debit, creation_timestamp) " +
                    "VALUES (:transactionId, :accountId, :description, :credit, :debit, :creationTimestamp)")
            .beanMapped()
            .build();
}

If we have both elements ready, we can build the next step and insert it into the previously prepared Job.

@Bean
public Step importTransactionsStep() {
    return this.stepBuilderFactory
            .get("importTransactionsStep")
            .<Transaction, Transaction>chunk(100)
            .reader(transactionItemReader(null))
            .writer(transactionItemWriter(null))
            .build();
}

@Bean
public Job importClubDataJob() throws Exception {
    return this.jobBuilderFactory
            .get("importClubDataJob")
            .start(importClubUpdatesStep())
            .next(importTransactionsStep())
            .incrementer(new RunIdIncrementer())
            .build();
}

Two parts ready, so let’s try to start our machine a second time. With the ItemReader and ItemWriter configured, we can build our job via ./mvnw clean install and run it with the same command as last time plus our new input file
parameter: java -jar SoccerBank-0.0.1-SNAPSHOT.jar clubData=<path to file with club data> transactionFile=<path to file with transactions>. After our job’s successful run, we can validate that the values in the transaction XML file got into the transaction table in our database.

Job: [SimpleJob: [name=importClubDataJob]] completed with the following parameters: [{clubData=clubDataFile.csv, transactionFile=transactions.xml, clubDataFile=clubDataFile.csv, run.id=5}] and the following status: [COMPLETED] in 294ms

@Bean
public Job importClubDataJob() throws Exception {
    return this.jobBuilderFactory
            .get("importClubDataJob")
            .start(importClubUpdatesStep())
            .next(importTransactionsStep())
            .next(applyClubTransactionStep())
            .incrementer(new RunIdIncrementer())
            .build();
}

@Bean
public Step applyClubTransactionsStep() {
    return stepBuilderFactory
            .get("applyClubTransactionsStep")
            .<Transaction, Transaction>chunk(100)
            .reader(applyClubTransactionReader(null))
            .writer(applyClubTransactionWriter(null))
            .build();
}

@Bean
public JdbcCursorItemReader<Transaction> applyClubTransactionReader(DataSource dataSource) {
    return new JdbcCursorItemReaderBuilder<Transaction>()
            .name("applyClubTransactionReader")
            .dataSource(dataSource)
            .sql("SELECT transaction_id, account_id, description, credit, debit, creation_timestamp " +
                    "FROM transaction ORDER BY creation_timestamp")
            .rowMapper((resultSet, i) ->
                    new Transaction(
                            resultSet.getLong("transaction_id"),
                            resultSet.getLong("account_id"),
                            resultSet.getString("description"),
                            resultSet.getBigDecimal("credit"),
                            resultSet.getBigDecimal("debit"),
                            resultSet.getTimestamp("creation_timestamp")))
            .build();
}

If we have read what we need, now we can take up the most interesting task – we will now introduce changes to our database based on transactions made by the clubs.

To save the changes to the database, I configured ItemWriter with JdbcBatchItemWriter. During the initialization of this method, the DataSource and the SQL statement were declared. This SQL query is responsible for adding the transaction amount to the current balance and identifies that the parameters in our SQL query can be filled by calling the seed property.

@Bean
public JdbcBatchItemWriter applyClubTransactionWriter(DataSource dataSource) {
    return new JdbcBatchItemWriterBuilder<Transaction>()
            .dataSource(dataSource)
            .sql("UPDATE account SET balance = balance + :transactionAmount WHERE account_id = accountId")
            .beanMapped()
            .assertUpdates(false)
            .build();
}

Job: [SimpleJob: [name=importClubDataJob]] completed with the following parameters: [{clubData=clubDataFile.csv, transactionFile=transactions.xml, clubDataFile=clubDataFile.csv, run.id=12}] and the following status: [COMPLETED] in 311ms

@Getter
@Setter
@NoArgsConstructor
public class Statement {

    private Club club;
    private List<Account> accounts = new ArrayList<>();

    public Statement(Club club, List<Account> accounts) {
        this.club = club;
        this.accounts.addAll(accounts);
    }
}

@Getter
@Setter
public class Club {

    private final long id;
    private final String name;
    private final String address;
    private final String emailAddress;
    private final String phone;
    private final String notification;
    private final int yearOfFoundation;

    public Club(long id, String name, String address, String emailAddress, String phone, String notification, int yearOfFoundation) {
        this.id = id;
        this.name = name;
        this.address = address;
        this.emailAddress = emailAddress;
        this.phone = phone;
        this.notification = notification;
        this.yearOfFoundation = yearOfFoundation;
    }
}

@Getter
@Setter
public class Account {

    private final long id;
    private final BigDecimal balance;
    private final Date lastStatementTimestamp;
    private final List<Transaction> transactions = new ArrayList<>();

    public Account(long id, BigDecimal balance, Date lastStatementTimestamp) {
        this.id = id;
        this.balance = balance;
        this.lastStatementTimestamp = lastStatementTimestamp;
    }
}

@Bean
public Job importClubDataJob() throws Exception {
    return this.jobBuilderFactory
            .get("importClubDataJob")
            .start(importClubUpdatesStep())
            .next(importTransactionsStep())
            .next(applyClubTransactionsStep())
            .next(generateStatementsStep())
            .incrementer(new RunIdIncrementer())
            .build();
}

@Bean
public Step generateStatements(AccountItemProcessor itemProcessor) {
    return this.stepBuilderFactory
            .get("generateStatements")
            .<Statement, Statement>chunk(1)
            .reader(statementItemReader(null))
            .processor(itemProcessor)
            .writer(statementItemWriter(null))
            .build();
}

@Bean
public JdbcCursorItemReader statementItemReader(DataSource dataSource) {
    return new JdbcCursorItemReaderBuilder<Statement>()
            .name("statementItemReader")
            .dataSource(dataSource)
            .sql("SELECT * FROM club")
            .rowMapper((resultSet, i) -> {
                Club club = Club.builder()
                        .id(resultSet.getLong("club_id"))
                        .name(resultSet.getString("name"))
                        .address(resultSet.getString("address"))
                        .emailAddress(resultSet.getString("email_address"))
                        .phone(resultSet.getString("phone"))
                        .notification(resultSet.getString("is_notified"))
                        .yearOfFoundation(resultSet.getInt("year_of_foundation"))
                        .build();
                return new Statement(club);
            })
            .build();
}

@Component
public class AccountItemProcessor implements ItemProcessor<Statement, Statement> {

    private final JdbcTemplate jdbcTemplate;

    public AccountItemProcessor(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }

    @Override
    public Statement process(Statement statement) throws Exception {
        statement.setAccounts(this.jdbcTemplate
                .query(
                        "SELECT a.account_id, a.balance, a.last_statement_timestamp, " +
                                "t.transaction_id, t.description, t.credit, t.debit, t.creation_timestamp " +
                                "FROM account a " +
                                "LEFT JOIN transaction t ON a.account_id = t.account_id " +
                                "WHERE a.account_id IN " +
                                "(SELECT account_account_id FROM club_account WHERE club_club_id = ? " +
                                "ORDER BY t.creation_timestamp)",
                new Object[] {statement.getClub().getId()}, new AccountResultSetExtractor()));
        return statement;
    }
}

public class AccountResultSetExtractor implements ResultSetExtractor<List<Account>> {

    private List<Account> accounts = new ArrayList<>();
    private Account currentAccount;

    @Nullable
    @Override
    public List<Account> extractData(ResultSet rs) throws SQLException, DataAccessException {
        while (rs.next()) {
            if (currentAccount == null) {
                currentAccount = new Account(
                        rs.getLong("account_id"),
                        rs.getBigDecimal("balance"),
                        rs.getDate("last_statement_date"));
            } else if (rs.getLong("account_id") != currentAccount.getId()) {
                accounts.add(currentAccount);
                currentAccount = new Account(
                        rs.getLong("account_id"),
                        rs.getBigDecimal("balance"),
                        rs.getDate("last_statement_date"));
            }

            if(StringUtils.hasText(rs.getString("description"))) {
                currentAccount.getTransactions().add(new Transaction(
                        rs.getLong("transaction_id"),
                        rs.getLong("account_id"),
                        rs.getString("descritpion"),
                        rs.getBigDecimal("credit"),
                        rs.getBigDecimal("debit"),
                        new Date(rs.getTimestamp("creation_timestamp").getTime())));
            }
        }
        if (currentAccount != null) {
            accounts.add(currentAccount);
        }
        return accounts;
    }
}

public class StatementLineAggregator implements LineAggregator<Statement> {

    private static final String ADDRESS_LINE = String.format("%121s\n", "Soccer Manager Banking");
    private static final String STATEMENT_DATE_LINE = String.format("Club Account Summary %78s ", "Statement Period")
            + "%tD to %tD\n\n";

    @Override
    public String aggregate(Statement statement) {
        StringBuilder output = new StringBuilder();
        formatHeader(statement, output);
        formatAccount(statement, output);
        return output.toString();
    }
}

private void formatAccount(Statement statement, StringBuilder output) {
    if (!CollectionUtils.isEmpty(statement.getAccounts())) {
        for (Account account : statement.getAccounts()) {
            output.append(String.format(STATEMENT_DATE_LINE, account.getLastStatementTimestamp(), new Date()));
            BigDecimal creditAmount = new BigDecimal(0);
            BigDecimal debitAmount = new BigDecimal(0);
            for (Transaction transaction : account.getTransactions()) {
                if (transaction.getCredit() != null) {
                    creditAmount = creditAmount.add(transaction.getCredit());
                }
                if (transaction.getDebit() != null) {
                    debitAmount = debitAmount.add(transaction.getDebit());
                }
                output.append(String.format("%tD %-50s %8.2f\n",
                        transaction.getCreationTimestamp(), transaction.getDescription(), transaction.getTransactionAmount()));
            }
            output.append(String.format("%80s %14.2f\n", "Total Debit:" , debitAmount));
            output.append(String.format("%81s %13.2f\n", "Total Credit:", creditAmount));
            output.append(String.format("%76s %18.2f\n\n", "Balance:", account.getBalance()));
        }
    }
}

private void formatHeader(Statement statement, StringBuilder output) {
    Club club = statement.getClub();
    String clubName = String.format("\n%s", club.getName());
    output.append(clubName).append(ADDRESS_LINE.substring(clubName.length()));
}

@Bean
@StepScope
public MultiResourceItemWriter<Statement> statementItemWriter(
        @Value("#{jobParameters['outputDirectory']}") Resource outputDirectory) {
    final String ROOT_PATH = "file://home/oskarro/";
    ClassPathResource classPathResource = new ClassPathResource(ROOT_PATH + outputDirectory.getFilename());
    Resource resource = resourceLoader.getResource(classPathResource.getPath());
    return new MultiResourceItemWriterBuilder<Statement>()
            .name("statementItemWriter")
            .resource(resource)
            .itemCountLimitPerResource(1)
            .delegate(singleStatementItemWriter())
            .build();
}

                                                                           Club Service Number
                                                                               (48) 333-8401
                                                                                Available 24/7



Czarni Deblin                                                   Soccer Manager Banking
Your Account Summary                                      Statement Period 04/04/18 to 10/10/21

                                                                    Total Debit:           0.00
                                                                    Total Credit:          0.00
                                                                    Balance:           50000.00

                                                                                                     Club Service Number
                                                                                                           (48) 333-8401
                                                                                                          Available 24/7



Legia Warszawa                                                   Soccer Manager Banking
Your Account Summary                                     Statement Period 08/22/19 to 10/10/21

                                                                    Total Debit:           0.00
                                                                    Total Credit:          0.00
                                                                    Balance:          999000.00

Your Account Summary                                    Statement Period 04/04/18 to 10/10/21

                                                                    Total Debit:           0.00
                                                                    Total Credit:          0.00
                                                                    Balance:           50000.00

Your Account Summary                                     Statement Period 11/15/16 to 10/10/21

02/01/21 Improve local area                                 -60111.00
                                                                    Total Debit:      -60111.00
                                                                    Total Credit:          0.00
                                                                    Balance:        -1055550.00