There are plenty of Spring guides that get you started on lots of topics: https://spring.io/guides. But every now and then, you need to combine topics to achieve your goal. This is such an example, where we want to:
- invoke a REST Service that returns a number of records
- process these records one-by-one (=batch processing)
- ... and do that on a scheduled interval (=task scheduling)
As always, this example is situated in the Terra 10 beer brewery:
In the above figure:
Beer catalog
An external REST API that returns a list of beer records:
developer@developer-VirtualBox:~$ curl https://virtserver.swaggerhub.com/TerraX_Brewery/Beercatalog/1.0.0/beers
[ {
"id" : "d290f1ee-6c54-4b01-90e6-d701748f0851",
"name" : "TerraX Tripel",
"releaseDate" : "2019-01-28T09:12:33.001Z",
"manufacturer" : {
"name" : "TerraX Brewery",
"homePage" : "https://terra10.io",
"phone" : "123-555-666"
}
}, {
"id" : "d290f1ee-6c54-4b01-90e6-d701748f0852",
"name" : "TerraX Golden Dragon",
.....
It is implemented as an mock service in SwaggerHub. Note that all sources can be found on GitHub: https://github.com/lgorissen/sb-scheduled-batch-jobs.git. The Swagger definition can be found there as file TerraX_Brewery-Beercatalog-1.0.0-swagger.yaml
.
Scheduled Jobs
This is the core part of this blog: a Spring Boot service that schedules a job every 10 seconds, that:
- Queries the Beer catalog service for a list of beer records
- Processes each beer record, i.e. one-by-one
The goal is to achieve the above functionality with as little code as possible, i.e. maximal leveraging of the Spring Boot features.
The table below shows the java files:
File | Description |
---|---|
io/terrax/brewery/beer/Beer.java | Defines the records from the Beer catalog API - Beer |
io/terrax/brewery/beer/Manufacturer.java | Defines the records from the Beer catalog API - Manufacturer |
io/terrax/brewery/beer/inventory/Application.java | Schedules Jobs - the Spring Boot application |
io/terrax/brewery/beer/inventory/JobCompletionNotificationListener.java | Logic that is executed when job is completed |
io/terrax/brewery/beer/inventory/ScheduledTasks.java | Job definition and scheduling |
io/terrax/brewery/beer/inventory/job/BeerCatalogReader.java | Job step - for reading from Beer catalog API |
io/terrax/brewery/beer/inventory/job/BeerInventoryWriter.java | Job step - for updating the Beer inventory TerraX |
io/terrax/brewery/beer/inventory/job/BeerProcessor.java | Job step - for internal transformation of Beer records |
Both Beer.java
and Manufacturer.java
are business classes that represent the beer records that are received from the Beer catalog API. What more is there to say ...
The Application.java
starts the Spring Boot application. Note that it enables both Scheduling and BatchProcessing:
...
@SpringBootApplication
@EnableScheduling // enable scheduling
@EnableBatchProcessing // enable batch processing
public class Application {
public static void main(String[] args) throws Exception {
SpringApplication.run(Application.class, args);
}
}
By far the most interesting part: the job scheduling happens here. The most interesting parts:
- ScheduledTasks is a SpringBoot Component, thus registering this class as a Bean
- A JobLauncher, JobBuilderFactory and StepBuilderFactory are autowired as they are needed for handling jobs
- A JobCompletionNotificationListener is needed for executing logic upon Job completion
- @Scheduled annotation schedules a void method for execution. In this example, the method is executed 5000msec after the previous execution has finished
@Component
public class ScheduledTasks {
private static final Logger log = LoggerFactory.getLogger(ScheduledTasks.class);
private static final SimpleDateFormat dateFormat = new SimpleDateFormat("HH:mm:ss");
@Autowired
JobLauncher jobLauncher;
@Autowired
private JobBuilderFactory jobs;
@Autowired
private StepBuilderFactory steps;
@Autowired
JobCompletionNotificationListener listener;
@Autowired
Environment environment;
@Scheduled(fixedDelay = 5000)
public void updateBeerInventory() {
log.info("Updating beer inventory for Terra10 {}", dateFormat.format(new Date()));
try {
RestTemplate restTemplate = new RestTemplate(); // for REST call to Beer catalog API
// A Job consists of
// - an ItemReader
// - an ItemProcessor
// - an ItemWriter
ItemReader<Beer> restBeerReaderLocal = new BeerCatalogReader(environment.getRequiredProperty("rest.api.to.beer.catalog.url"), restTemplate);
ItemProcessor<Beer, Beer> restBeerProcessorLocal = new BeerProcessor();
ItemWriter<Beer> restBeerWriterLocal = new BeerInventoryWriter();
// Define a Job Step
Step restBeerStepLocal = steps.get("restBeerStepLocal") // get new StepBuilder
.allowStartIfComplete(true) // allow job to be re-started
.<Beer, Beer>chunk(1) // step processes chunks of size 1
.reader(restBeerReaderLocal) // add the reader
.processor(restBeerProcessorLocal) // add the processer
.writer(restBeerWriterLocal) // add the writer
.build(); // build the step
// Define a Job
Job restBeerJobLocal = jobs.get("restBeerJobLocal") // get new JobBuilder
.incrementer(new RunIdIncrementer()) // increment Run.id parameter
.listener(listener) // at job listener to JobBuilder
.flow(restBeerStepLocal) // add step to JobBuilder
.end() // let JobFlowBuilder build
.build(); // inject into parent builder
jobLauncher.run(restBeerJobLocal, new JobParameters());
} catch (Exception e) {
// TODO: handle exception
e.printStackTrace();
}
}
The BeerCatalogReader is the ItemReader of the Job. It reads the records from the Beer catalog API.
package io.terrax.brewery.beer.inventory.job;
...
public class BeerCatalogReader implements ItemReader<Beer> {
public BeerCatalogReader(String apiUrl, RestTemplate restTemplate) {
this.apiUrl = apiUrl; // api endpoint needed for doing the REST call
this.restTemplate = restTemplate; // restTeample needed for doing the REST call
nextBeerIndex = 0;
}
@Override
public Beer read() throws Exception { // ItemReader needs a read() method
if (beerDataIsNotInitialized()) {
beerData = fetchBeerDataFromAPI();
}
Beer nextBeer = null;
if (nextBeerIndex < beerData.size()) {
nextBeer = beerData.get(nextBeerIndex);
nextBeerIndex++;
}
return nextBeer; // the read() method returns the beers one-by-one
}
...
}
The BeerInventoryWriter is the ItemWriter of the Job. It writes the records to the Beer inventory ... well: we only write a log line
package io.terrax.brewery.beer.inventory.job;
...
public class BeerInventoryWriter implements ItemWriter<Beer> {
private static final Logger log = LoggerFactory.getLogger(BeerInventoryWriter.class);
@Override
public void write(List<? extends Beer> items) throws Exception { // in the write method, beer records are processed
items.forEach(i -> log.info("Adding beer to inventory: " + i.getName())); // instead of writing them to an inventory ...
// we only write a log line
}
}
In the BeerProcessor, it is possible to do some processing, like e.g. data enrichment on the received Beer records.
package io.terrax.brewery.beer.inventory.job;
...
public class BeerProcessor implements ItemProcessor<Beer, Beer> {
private static final Logger log = LoggerFactory.getLogger(BeerProcessor.class);
@Override
public Beer process(Beer item) throws Exception {
log.info("Processing beer information: {}", item.getName());
// you could do some work on the Beer item here
return item;
}
}
All three together, the BeerCatalogReader, BeerProcessor and BeerInventoryWriter are a typical ETL pattern, where:
- Extraction: the BeerCatalogReader
- Transformation: the BeerProcessor
- Load: the BeerInventoryWriter
The application first has to be build:
developer@developer-VirtualBox:~/projects/scheduled-batch-jobs$ ls
mvnw overview.png README.md target
mvnw.cmd pom.xml src TerraX_Brewery-Beercatalog-1.0.0-swagger.yaml
developer@developer-VirtualBox:~/projects/scheduled-batch-jobs$ ./mvnw clean package
[INFO] Scanning for projects...
[INFO]
[INFO] -------------------< io.terrax:scheduled-batch-jobs >-------------------
[INFO] Building scheduled-batch-jobs 0.0.1-SNAPSHOT
[INFO] --------------------------------[ jar ]---------------------------------
[INFO]
[INFO] --- maven-clean-plugin:3.1.0:clean (default-clean) @ scheduled-batch-jobs ---
[INFO] Deleting /home/developer/projects/scheduled-batch-jobs/target
[INFO]
...
...
[INFO] --- spring-boot-maven-plugin:2.1.2.RELEASE:repackage (repackage) @ scheduled-batch-jobs ---
[INFO] Replacing main artifact with repackaged archive
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 3.599 s
[INFO] Finished at: 2019-02-09T20:20:57+01:00
[INFO] ------------------------------------------------------------------------
developer@developer-VirtualBox:~/projects/scheduled-batch-jobs$
After a successful build, the application can be run.
developer@developer-VirtualBox:~/projects/scheduled-batch-jobs$ java -jar target/scheduled-batch-jobs-0.0.1-SNAPSHOT.jar
. ____ _ __ _ _
/\\ / ___'_ __ _ _(_)_ __ __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
\\/ ___)| |_)| | | | | || (_| | ) ) ) )
' |____| .__|_| |_|_| |_\__, | / / / /
=========|_|==============|___/=/_/_/_/
:: Spring Boot :: (v2.1.2.RELEASE)
2019-02-09 20:25:00.747 INFO 22523 --- [ main] i.t.brewery.beer.inventory.Application : Starting Application v0.0.1-SNAPSHOT on developer-VirtualBox with PID 22523 (/home/developer/projects/scheduled-batch-jobs/target/scheduled-batch-jobs-0.0.1-SNAPSHOT.jar started by developer in /home/developer/projects/scheduled-batch-jobs)
2019-02-09 20:25:00.752 INFO 22523 --- [ main] i.t.brewery.beer.inventory.Application : No active profile set, falling back to default profiles: default
2019-02-09 20:25:01.744 INFO 22523 --- [ main] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Starting...
2019-02-09 20:25:01.749 WARN 22523 --- [ main] com.zaxxer.hikari.util.DriverDataSource : Registered driver with driverClassName=org.hsqldb.jdbcDriver was not found, trying direct instantiation.
2019-02-09 20:25:02.074 INFO 22523 --- [ main] com.zaxxer.hikari.pool.PoolBase : HikariPool-1 - Driver does not support get/set network timeout for connections. (feature not supported)
2019-02-09 20:25:02.080 INFO 22523 --- [ main] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Start completed.
2019-02-09 20:25:02.103 INFO 22523 --- [ main] o.s.b.c.r.s.JobRepositoryFactoryBean : No database type set, using meta data indicating: HSQL
2019-02-09 20:25:02.135 INFO 22523 --- [ main] o.s.b.c.l.support.SimpleJobLauncher : No TaskExecutor has been set, defaulting to synchronous executor.
2019-02-09 20:25:02.363 INFO 22523 --- [ main] o.s.s.c.ThreadPoolTaskScheduler : Initializing ExecutorService 'taskScheduler'
2019-02-09 20:25:02.414 INFO 22523 --- [ main] i.t.brewery.beer.inventory.Application : Started Application in 2.106 seconds (JVM running for 2.629)
2019-02-09 20:25:02.416 INFO 22523 --- [ main] o.s.b.a.b.JobLauncherCommandLineRunner : Running default command line with: []
2019-02-09 20:25:02.417 INFO 22523 --- [ scheduling-1] i.t.b.beer.inventory.ScheduledTasks : Updating beer inventory for Terra10 20:25:02
2019-02-09 20:25:02.559 INFO 22523 --- [ scheduling-1] o.s.b.c.l.support.SimpleJobLauncher : Job: [FlowJob: [name=restBeerJobLocal]] launched with the following parameters: [{}]
2019-02-09 20:25:02.596 INFO 22523 --- [ scheduling-1] o.s.batch.core.job.SimpleStepHandler : Executing step: [restBeerStepLocal]
2019-02-09 20:25:03.569 INFO 22523 --- [ scheduling-1] i.t.b.beer.inventory.job.BeerProcessor : Processing beer information: TerraX Tripel
2019-02-09 20:25:03.570 INFO 22523 --- [ scheduling-1] i.t.b.b.i.job.BeerInventoryWriter : Adding beer to inventory: TerraX Tripel
2019-02-09 20:25:03.573 INFO 22523 --- [ scheduling-1] i.t.b.beer.inventory.job.BeerProcessor : Processing beer information: TerraX Golden Dragon
2019-02-09 20:25:03.573 INFO 22523 --- [ scheduling-1] i.t.b.b.i.job.BeerInventoryWriter : Adding beer to inventory: TerraX Golden Dragon
2019-02-09 20:25:03.575 INFO 22523 --- [ scheduling-1] i.t.b.beer.inventory.job.BeerProcessor : Processing beer information: TerraX Bock
2019-02-09 20:25:03.575 INFO 22523 --- [ scheduling-1] i.t.b.b.i.job.BeerInventoryWriter : Adding beer to inventory: TerraX Bock
2019-02-09 20:25:03.577 INFO 22523 --- [ scheduling-1] i.t.b.beer.inventory.job.BeerProcessor : Processing beer information: TerraX Double
2019-02-09 20:25:03.577 INFO 22523 --- [ scheduling-1] i.t.b.b.i.job.BeerInventoryWriter : Adding beer to inventory: TerraX Double
2019-02-09 20:25:03.579 INFO 22523 --- [ scheduling-1] i.t.b.beer.inventory.job.BeerProcessor : Processing beer information: TerraX Weizen
2019-02-09 20:25:03.580 INFO 22523 --- [ scheduling-1] i.t.b.b.i.job.BeerInventoryWriter : Adding beer to inventory: TerraX Weizen
2019-02-09 20:25:03.589 INFO 22523 --- [ scheduling-1] .b.b.i.JobCompletionNotificationListener : Job Completed: beer inventory updated!
2019-02-09 20:25:03.591 INFO 22523 --- [ scheduling-1] o.s.b.c.l.support.SimpleJobLauncher : Job: [FlowJob: [name=restBeerJobLocal]] completed with the following parameters: [{}] and the following status: [COMPLETED]
2019-02-09 20:25:08.592 INFO 22523 --- [ scheduling-1] i.t.b.beer.inventory.ScheduledTasks : Updating beer inventory for Terra10 20:25:08
2019-02-09 20:25:08.628 INFO 22523 --- [ scheduling-1] o.s.b.c.l.support.SimpleJobLauncher : Job: [FlowJob: [name=restBeerJobLocal]] launched with the following parameters: [{}]
2019-02-09 20:25:08.639 INFO 22523 --- [ scheduling-1] o.s.batch.core.job.SimpleStepHandler : Executing step: [restBeerStepLocal]
Watch the log lines to see the beer records being processed.
All the sources can be found on github: https://github.com/lgorissen/sb-scheduled-batch-jobs
Enjoy!