Spring Batch and Visual Rules

Spring Batch and Visual Rules

Today I want to show how to use the Spring Batch to execute batch jobs which make use of business rules.

Spring Batch is a framework for managing Enterprise jobs, that is reading, transforming and writing data sets. Batch jobs are often run asynchronously at night and process large amounts of data.

Business rules are used to separate business logic into separate modules, which can be tested and management without touching application source code at all. That also means business rules can be managed and modified by non-developers and even by non-technical business people to some degree.

As business rules, i am using the Movie Ticket Pricing example rule code generated by Visual Rules.

This blog post is rather long and contains a lot of code. To set up our test case, we need to define some dependencies on the project's pom.xml (Maven):

 
  de.innovations.visualrules
  visualrules-runtime
  4.0.0.v20080414-1927
 
 
  org.springframework.batch
  spring-batch-core
  1.0.1.RELEASE
 
 
  org.springframework
  spring-aop
  2.5.2
 
 
  org.springframework
  spring-tx
  2.5.2
 
 
  org.springframework
  spring-jdbc
  2.5.4
 

The dependencies include the visual rules runtime, required to run the generated rule code. The Spring Batch Core dependency also requires Spring AOP, Spring Transactions and Spring JDBC, as we load input data from the database.

Let's start by filling our empty test case:

import javax.sql.DataSource;
import org.junit.Test;
public class MoviePricingBatchTest {
 @Test
 public void testMoviePricingBatch() throws Exception {
 DataSource dataSource = setupDatabase();
 }
}

Implementation of the setupDatabase() method creates an In-Memory database using HSQLDB, creates the table schema for our input data and inserts some example data to work with:

private DataSource setupDatabase() {
 DataSource dataSource =
 new SingleConnectionDataSource(jdbcDriver.class.getName(),
     "jdbc:hsqldb:mem:db_movieticket", "sa", "", true);
 SQLRunner.runScript(getClass().
  getResourceAsStream("/movieticket.sql"),dataSource);
 return dataSource;
}

Second step is to instantiate the Visual Rules IRuleModel we want to work with:

import javax.sql.DataSource;
import org.junit.Test;
public class MoviePricingBatchTest {
 @Test
 public void testMoviePricingBatch() throws Exception {
 DataSource dataSource = setupDatabase();
 IRuleModel ruleModel = createRuleModel();
 }
}

The createRuleModel() method is very simple in our case:

 private IRuleModel createRuleModel() {
 return new Movie_Ticket_PricingRuleModel();
 }

Now comes the intersting part - set up of Spring Batch. The design of Spring Batch jobs is rather simple: read data, (transform data), write data. We'll have a look on how to read data from the database first:

import javax.sql.DataSource;
import org.junit.Test;
public class MoviePricingBatchTest {
 @Test
 public void testMoviePricingBatch() throws Exception {
 DataSource dataSource = setupDatabase();
 IRuleModel ruleModel = createRuleModel();
 ItemReader itemReader =
      createItemReader(dataSource, ruleModel);
 }
}

The createItemReader() method creates a Spring Batch ItemReader which reads data from a JDBC database and maps the input records into Visual Rules IRequestData objects. These request data objects hold input and output parameters, and some additional meta information.

private JdbcCursorItemReader createItemReader(DataSource dataSource,
  IRuleModel ruleModel) {
 String sql = "SELECT show_date," + " coupon," + " auditorium_no,"
  + " seat_no, student," + " bonus_card" + " FROM reservations";
 JdbcCursorItemReader jdbcCursorItemReader
       = new JdbcCursorItemReader();
 jdbcCursorItemReader.setDataSource(dataSource);
 jdbcCursorItemReader.
     setMapper(new MovieTicketRowMapper(ruleModel,IPricing.RULE_NAME));
 jdbcCursorItemReader.setSql(sql);
 return jdbcCursorItemReader;
 }

Note the MovieTicketRowMapper implementation, which extends an abstract utility class AbstractRequestDataRowMapper. The AbstractRequestDataRowMapper helps us to dynamically map columns into IRequestData In-Parameters.

public class MovieTicketRowMapper extends AbstractRequestDataRowMapper
 implements RowMapper {
 public MovieTicketRowMapper(IRuleModel ruleModel, String ruleName) {
 super(ruleModel, ruleName);
 }
 @Override
 protected void mapResultSetIntoRequestData(ResultSet rs, int rowNum,
  IRequestData requestData) throws SQLException {
 mapAllColumnsByName(rs, requestData);
 set(requestData, "show_date",
      new Date(rs.getDate("show_date").getTime()));
 String strBonusCard = rs.getString("bonus_card");
 BONUS_CARD bonusCard = BONUS_CARD.byName(strBonusCard);
 set(requestData, "bonus_card", bonusCard);
 }
}

The two special parameters "show_date" and "bonus_card" are handled manually, because their types need to be converted. All other parameters are of primitive types (boolean and int) and can be automatically converted by the AbstractRequestDataRowMapper.

Back to the implementation of our testcase, we now have the ItemReader. Next, we need to have a transformer. The transformer uses the IRequestData created by the MovieTicketRowMapper and executes the IRuleModel with that data for each data item read from the database.

import javax.sql.DataSource;
import org.junit.Test;
public class MoviePricingBatchTest {
 @Test
 public void testMoviePricingBatch() throws Exception {
 DataSource dataSource = setupDatabase();
 IRuleModel ruleModel = createRuleModel();
 ItemReader itemReader = createItemReader(dataSource, ruleModel);
 ItemWriter transformer = createTransformer(ruleModel);
 }
}

createTransformer() creates an ItemTransformerItemWriter instance, which uses a SimpleRuleModelExecutionTransformer to execute the IRuleModel instance. Then, it delegates the result to the LogRequestDataItemWriter. The LogRequestItemWriter prints out the result as log.info() on the console.

private ItemTransformerItemWriter createTransformer(IRuleModel ruleModel) {
 ItemTransformerItemWriter transformer
     = new ItemTransformerItemWriter();
 transformer.setItemTransformer(
     new SimpleRuleModelExecutionTransformer(ruleModel));
 ItemWriter writer = new LogRequestDataItemWriter();
 transformer.setDelegate(writer);
 return transformer;
 }

Now it's time to set up the Spring Batch stuff: the Job Repository, the Job Launcher, the Step and the Job, as long with the JobParameters:

import javax.sql.DataSource;
import org.junit.Test;
public class MoviePricingBatchTest {
 @Test
 public void testMoviePricingBatch() throws Exception {
 DataSource dataSource = setupDatabase();
 IRuleModel ruleModel = createRuleModel();
 ItemReader itemReader = 
     createItemReader(dataSource, ruleModel);
 ItemWriter transformer = 
     createTransformer(ruleModel);
 JobRepository jobRepository = createRepository();
 JobLauncher launcher = createLauncher(jobRepository);
 Step step =
     createStep(itemReader, transformer, jobRepository);
 Job job = createJob(jobRepository, step);
 JobParameters jobParameters = createJobParameters();
 }
}

The JobRepository uses in-memory DAOs to store Job states:

private SimpleJobRepository createRepository() {
 return new SimpleJobRepository(new MapJobInstanceDao(),
  new MapJobExecutionDao(), new MapStepExecutionDao());
 }

The JobLauncher uses a synchronous Task Executor for the sake of simplicity of the unit test. This can be easily reconfigured to make use of multi-threaded parallel processing.

private JobLauncher createLauncher(JobRepository jobRepository)
  throws Exception {
 SimpleJobLauncher launcher = new SimpleJobLauncher();
 launcher.setJobRepository(jobRepository);
 launcher.setTaskExecutor(new SyncTaskExecutor());
 launcher.afterPropertiesSet();
 return launcher;
}

The Step is more complex and needs some additional configuration. We'll use a Step Factory provided by the Spring Batch Framework. The factory will create a Step which is able to support retries and processing chunks. After each chunk, the process is automatically committed.

private Step createStep(ItemReader itemReader, ItemWriter itemWriter,
  JobRepository jobRepository) throws Exception {
 PlatformTransactionManager transactionManager
   = new DummyPlatformTransactionManager();
 ItemFailureLoggerListener itemFailureLoggerListener
   = new ItemFailureLoggerListener();
 StatefulRetryStepFactoryBean factory 
   = new StatefulRetryStepFactoryBean();
 factory.setTransactionManager(transactionManager);
 factory.setItemReader(itemReader);
 factory.setItemWriter(itemWriter);
 factory.setJobRepository(jobRepository);
 factory.setCommitInterval(5);
 factory.setListeners(new StepListener[] {
   itemFailureLoggerListener });
 factory.setRetryableExceptionClasses(new Class[] {
   IOException.class });
 factory.setRetryLimit(5);
 factory.setBeanName("simpleStepFactoryBean");
 factory
  .setSkippableExceptionClasses(new Class[] {
    ValidationException.class });
 factory.setSkipLimit(3); // Allow 3 validation failures
 Step step = (Step) factory.getObject();
 return step;
 }

The Job is straight forward: it has a name and contains only one Step:

private SimpleJob createJob(JobRepository jobRepository, Step step) {
 SimpleJob job = new SimpleJob();
 job.setName("Movie Ticket Recalculation");
 job.setJobRepository(jobRepository);
 job.addStep(step);
 return job;
 }

The JobParameters are global parameters. We don't use them in our example, but I wanted to show how they are constructed:

private JobParameters createJobParameters() {
 JobParameters jobParameters
    = new JobParametersBuilder().addDate("now",
  new Date()).addString("comment", "Unit Test")
    .toJobParameters();
 return jobParameters;
 }

We're finished! The only thing left is to actually execute the Job within the test case and test for the result being finished successfully:

import javax.sql.DataSource;
import org.junit.Test;
public class MoviePricingBatchTest {
 @Test
 public void testMoviePricingBatch() throws Exception {
 DataSource dataSource = setupDatabase();
 IRuleModel ruleModel = createRuleModel();
 ItemReader itemReader = createItemReader(dataSource, ruleModel);
 ItemWriter transformer = createTransformer(ruleModel);
 JobRepository jobRepository = createRepository();
 JobLauncher launcher = createLauncher(jobRepository);
 Step step = createStep(itemReader, transformer, jobRepository);
 Job job = createJob(jobRepository, step);
 JobParameters jobParameters = createJobParameters();
 
 JobExecution execution = launcher.run(job, jobParameters);

 ExitStatus exitStatus = execution.getExitStatus();
 Assert.assertEquals(ExitStatus.FINISHED, exitStatus);
 }
}

The implementation of the SimpleRuleModelExecutionTransformer:

public Object transform(Object item) throws Exception {
 ruleModel.perform((IRequestData) item);
 return item;
 }

I don't want to bloat this blog post any more, so I'll leave the implementation of the AbstractRequestDataRowMapper and the LogRequestDataItemWriter to the interested reader.

• Comments (0)
• Trackbacks (0)