Quarkus JBeret Components

The Quarkus JBeret Components provides reusable, batch components for common data processing tasks.

Installation

To use the JBeret Components module, add the io.quarkiverse.jberet:quarkus-jberet-components extension to your build file:

pom.xml

<dependency>
    <groupId>io.quarkiverse.jberet</groupId>
    <artifactId>quarkus-jberet-components</artifactId>
    <version>2.8.0</version>
</dependency>

build.gradle

implementation("io.quarkiverse.jberet:quarkus-jberet-components:2.8.0")

Read and Write Data through JDBC

The JDBC components provide efficient reading and writing of database records using JDBC Cursors and Batch processing. These components are ideal for:

Processing large database tables without loading all data into memory
Migrating data between databases
Generating aggregated statistics from database records
Bulk insert/update operations with optimal performance

JdbcCursorItemReader

The JdbcCursorItemReader reads data from a database using a JDBC cursor, meaning that it will read every resulting row from the supplied sql statement one row at a time without loading the entire result set into memory.

JdbcCursorItemReader

package org.acme.batch.components.jdbc;

import javax.sql.DataSource;

import jakarta.enterprise.context.Dependent;
import jakarta.enterprise.inject.Produces;
import jakarta.inject.Inject;
import jakarta.inject.Named;
import jakarta.inject.Singleton;

import io.quarkiverse.jberet.components.runtime.item.jdbc.JdbcCursorItemReader;

@Singleton
public class AuctionJdbcCursorItemReaderProducer {
    @Inject
    DataSource dataSource;
    @Inject
    AuctionStatisticsRowMapper rowMapper;

    @Produces
    @Dependent
    @Named("auctionsItemReader")
    public JdbcCursorItemReader<AuctionStatistics> auctionsItemReader() {
        String sql = """
                SELECT
                    itemId,
                    sum(quantity) as totalQuantity,
                    sum(bid) as totalBid,
                    sum(buyout) as totalBuyout,
                    min(bid / quantity) as minBid,
                    min(buyout / quantity) as minBuyout,
                    max(bid / quantity) as maxBid,
                    max(buyout / quantity) as maxBuyout
                FROM Auctions
                GROUP BY itemId
                ORDER BY itemId
                """;
        return new JdbcCursorItemReader<>(dataSource, sql, rowMapper);
    }
}

The JdbcCursorItemReader requires:

A DataSource to read the data
A SQL query to execute to retrieve the data
A RowMapper to convert each ResultSet row into a custom POJO

RowMapper

The RowMapper is a functional interface that maps a JDBC ResultSet row to a POJO:

RowMapper

package org.acme.batch.components.jdbc;

import java.sql.ResultSet;
import java.sql.SQLException;

import jakarta.inject.Named;
import jakarta.inject.Singleton;

import io.quarkiverse.jberet.components.runtime.item.jdbc.RowMapper;

@Singleton
@Named
public class AuctionStatisticsRowMapper implements RowMapper<AuctionStatistics> {
    @Override
    public AuctionStatistics mapRow(ResultSet resultSet) throws SQLException {
        int itemId = resultSet.getInt(1);
        long quantity = resultSet.getLong(2);
        long bid = resultSet.getLong(3);
        long buyout = resultSet.getLong(4);
        long minBid = resultSet.getLong(5);
        long minBuyout = resultSet.getLong(6);
        long maxBid = resultSet.getLong(7);
        long maxBuyout = resultSet.getLong(8);

        Double avgBid = (double) (bid / quantity);
        Double avgBuyout = (double) (buyout / quantity);

        return new AuctionStatistics(itemId, quantity, bid, minBid, maxBid, buyout, minBuyout, maxBuyout, avgBid, avgBuyout);
    }
}

AuctionStatistics

package org.acme.batch.components.jdbc;

public record AuctionStatistics(
        Integer itemId,
        Long quantity,
        Long bid,
        Long minBid,
        Long maxBid,
        Long buyout,
        Long minBuyout,
        Long maxBuyout,
        Double avgBid,
        Double avgBuyout) {
}

The RowMapper retrieves values from the ResultSet by column index and constructs the AuctionStatistics object.

JdbcBatchItemWriter

The JdbcBatchItemWriter writes data to a database using JDBC batch processing. Instead of executing one SQL statement per item, it groups multiple statements together and executes them in a single database operation.

JdbcBatchItemWriter

package org.acme.batch.components.jdbc;

import javax.sql.DataSource;

import jakarta.enterprise.context.Dependent;
import jakarta.enterprise.inject.Produces;
import jakarta.inject.Inject;
import jakarta.inject.Named;
import jakarta.inject.Singleton;

import io.quarkiverse.jberet.components.runtime.item.jdbc.JdbcBatchItemWriter;

@Singleton
public class AuctionJdbcBatchItemWriterProducer {
    @Inject
    DataSource dataSource;

    @Inject
    AuctionStatisticsParameterSetter parameterSetter;

    @Produces
    @Dependent
    @Named("auctionsItemWriter")
    public JdbcBatchItemWriter<AuctionStatistics> auctionsItemWriter() {
        String sql = """
                INSERT INTO AuctionStatistics (
                    id, itemId, quantity, bid, minBid, maxBid,
                    buyout, minBuyout, maxBuyout, avgBid, avgBuyout, timestamp
                ) VALUES (nextval('auction_statistics_id'), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """;
        return new JdbcBatchItemWriter<>(dataSource, sql, parameterSetter);
    }
}

The JdbcBatchItemWriter requires:

A DataSource to write the data
A parameterized SQL statement to execute for each item to write
A ParameterSetter to map objects into SQL parameters

ParameterSetter

The ParameterSetter is a functional interface that sets PreparedStatement parameters from a POJO:

ParameterSetter

package org.acme.batch.components.jdbc;

import java.sql.SQLException;

import jakarta.inject.Named;
import jakarta.inject.Singleton;

import io.quarkiverse.jberet.components.runtime.item.jdbc.ParameterSetter;

@Singleton
@Named
public class AuctionStatisticsParameterSetter implements ParameterSetter<AuctionStatistics> {
    @Override
    public void setValues(Parameters parameters, AuctionStatistics value) throws SQLException {
        parameters.setInt(1, value.itemId());
        parameters.setLong(2, value.quantity());
        parameters.setLong(3, value.bid());
        parameters.setLong(4, value.minBid());
        parameters.setLong(5, value.maxBid());
        parameters.setLong(6, value.buyout());
        parameters.setLong(7, value.minBuyout());
        parameters.setLong(8, value.maxBuyout());
        parameters.setDouble(9, value.avgBid());
        parameters.setDouble(10, value.avgBuyout());
        parameters.setLong(11, System.currentTimeMillis());
    }
}

The ParameterSetter extracts values from an object and sets them as PreparedStatement parameters by index in Parameters.

The Job

All JDBC components must be assembled in a Job definition:

auctionsJob.xml

<?xml version="1.0" encoding="UTF-8"?>
<job id="auctionsJob" xmlns="https://jakarta.ee/xml/ns/jakartaee" version="2.0">
    <step id="processAuctions">
        <chunk>
            <reader ref="auctionsItemReader"/>    (1)
            <writer ref="auctionsItemWriter"/>    (2)
        </chunk>
    </step>
</job>

1	The `auctionsItemReader` is the CDI bean name of the `JdbcCursorItemReader` produced by the `AuctionJdbcCursorItemReaderProducer`
2	The `auctionsItemWriter` is the CDI bean name of the `JdbcBatchItemWriter` produced by the `AuctionJdbcBatchItemWriterProducer`

To execute this Job:

@Inject
JobOperator jobOperator;

void execute() {
    long executionId = jobOperator.start("auctionsJob", new Properties());
}

Configuration with Batch Properties

Instead of using CDI producers, the JdbcCursorItemReader and JdbcBatchItemWriter and can be configured directly in the Job XML using batch properties and their built-in reference names jdbcItemReader and jdbcItemWriter:

auctionsJob.xml

<?xml version="1.0" encoding="UTF-8"?>
<job id="auctionsJob" xmlns="https://jakarta.ee/xml/ns/jakartaee" version="2.0">
	<step id="processAuctions">
		<chunk item-count="100">
			<reader ref="jdbcItemReader">
				(1)
				<properties>
					(2)
					<property name="sql"
						value="SELECT itemId, sum(quantity), sum(bid), sum(buyout), min(bid / quantity), min(buyout / quantity), max(bid / quantity), max(buyout / quantity) FROM Auctions GROUP BY itemId ORDER BY itemId" />
					(3)
					<property name="rowMapper" value="auctionStatisticsRowMapper" />
				</properties>
			</reader>
			<writer ref="jdbcItemWriter">
				(4)
				<properties>
					(5)
					<property name="sql"
						value="INSERT INTO AuctionStatistics (id, itemId, quantity, bid, minBid, maxBid, buyout, minBuyout, maxBuyout, avgBid, avgBuyout, timestamp) VALUES (nextval('auction_statistics_id'), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" />
					(6)
					<property name="parameterSetter" value="auctionStatisticsParameterSetter" />
				</properties>
			</writer>
		</chunk>
	</step>
</job>

1	Reference the built-in `jdbcItemReader` `JdbcCursorItemReader`
2	Specify the SQL query to execute to retrieve the data
3	Specify the CDI bean name of the `RowMapper`
4	Reference the built-in `jdbcItemWriter` `JdbcBatchItemWriter`
5	Specify SQL statement to execute for each item to write
6	Specify the CDI bean name of the `ParameterSetter`

When using batch properties, the dataSource property is optional. If not specified, the default (unnamed) datasource is used. For named datasources, use: <property name="dataSource" value="namedDatasource"/>

Fetch Size

The fetchSize property hints to the JDBC driver how many rows to fetch from the database:

<property name="fetchSize" value="1"/>

Higher fetch sizes reduce network overhead but increase memory usage. The optimal value depends on your network latency and row size.