How to Emulate a remote JVM in JUnit using Spring, Ant, H2 and ActiveMQ

How to Emulate a remote JVM in JUnit using Spring, Ant, H2 and ActiveMQ

recently i needed to test a scenario where one JVM was communicating to another using both a shared database and JMS.  i also needed to have each JVM be "invisible" to each other and make sure that the couldn't 'accidentally' share things across and application context.

so, to do this, i needed to kick start a separate JVM from inside a JUnit test, and communicate via JMS and a database... fun.

here's what i needed;

- shared database (standalone server)
- shared JMS (standalone server)
- JVM A - application context with unique configuration
- JVM B - application context with unique configuration

so, here's the solution i put together with some google/stackoverflow help and the following;
- Spring
- JUnit 4
- Ant
- H2
- ActiveMQ

here's the infrastructure utilities class.  it's responsible for setting up 'shared' infrastructure such as H2 and ActiveMQ.  it also provides hooks for binding to JNDI in a standalone context to help with sharing the resources in an 'agnostic' kind of way.

package de.incompleteco.spring.batch.ha;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;

import javax.sql.DataSource;

import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.spring.ActiveMQConnectionFactory;
import org.h2.jdbcx.JdbcConnectionPool;
import org.h2.jdbcx.JdbcDataSource;
import org.h2.tools.Server;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.mock.jndi.SimpleNamingContextBuilder;

public class InfrastructureUtils {

 private static final Logger logger = LoggerFactory.getLogger(InfrastructureUtils.class);
 
 private static Server server;
 
 private static BrokerService broker;
 
 public static void startH2() throws Exception {
  if (server == null) {
   server = Server.createTcpServer("-tcpAllowOthers");
   server.start();
   logger.info(server.getStatus());
  }//end if
 } 
 
 public static void stopH2() throws Exception {
  if (server != null) {
   server.stop();
  }//end if
 } 
 
 public static void startAMQ() throws Exception {
  if (broker == null) {
   broker = new BrokerService();
   broker.addConnector("tcp://localhost:61616");
   broker.setUseJmx(false);
   broker.setUseShutdownHook(true);
   broker.start();
   broker.deleteAllMessages();//clean up
  }//end if
 }
 
 public static void stopAMQ() throws Exception {
  if (broker != null) {
   broker.stop();
  }//end if
 } 
 
 public static void bindLocalAMQ(String connectionFactoryName,String... queueNames) {
  //get jndi
  SimpleNamingContextBuilder builder = SimpleNamingContextBuilder.getCurrentContextBuilder();
  //get the connection
  ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
  connectionFactory.setBrokerURL("tcp://localhost:61616");
  //bind
  builder.bind("jms/" + connectionFactoryName,connectionFactory);
  //setup the queues
  for (String name : queueNames) {
   builder.bind("jms/" + name,new ActiveMQQueue(name));   
  }//end for
 } 
 
 public static DataSource bindLocalH2(String dataSourceName) throws Exception {
  JdbcDataSource dataSource = new JdbcDataSource();
  dataSource.setURL("jdbc:h2:tcp://localhost/~/test");
  //build a pool and bind
  JdbcConnectionPool pool = JdbcConnectionPool.create(dataSource);
  SimpleNamingContextBuilder builder = SimpleNamingContextBuilder.getCurrentContextBuilder();
  builder.bind("jdbc/" + dataSourceName, pool);
  //return it in case there's other uses
  return dataSource;
 } 
 
 public static String[] convertSqlFile(String location) throws Exception {
  List sqlStatements = new ArrayList();
  //load up the file
  BufferedReader reader = new BufferedReader(new InputStreamReader(TestSimpleBatchHAService.class.getResourceAsStream(location)));
  String line;
  StringBuilder statement = new StringBuilder();
  while ((line = reader.readLine()) != null) {
   if (line.contains(";") && !line.contains("--")) {
    statement.append(line);
    sqlStatements.add(statement.toString().replace(';',' '));
    statement = new StringBuilder();//reset the string
   } else if (line.contains("--")) {
    //ignore
   } else {
    statement.append(line);
   }//end if
  }//end while
  reader.close();
  //return
  return sqlStatements.toArray(new String[sqlStatements.size()]);
 } 
}

here's the RemoteJVMEmulator class.  it's a 'standalone' class (static void main(..)) that bootstraps and application context.  it also uses the infrastructure utilities to bind the resources to the JNDI.

package de.incompleteco.spring.batch.ha;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.mock.jndi.SimpleNamingContextBuilder;

public class RemoteJVMEmulator {

 private static final Logger logger = LoggerFactory.getLogger(RemoteJVMEmulator.class);
 
 //bind shared resources to JNDI
 public static void setupJNDI() throws Exception {
  SimpleNamingContextBuilder.emptyActivatedContextBuilder();
  //bind AMQ
  InfrastructureUtils.bindLocalAMQ("ConnectionFactory","request.queue","reply.queue");
  //bind h2
  InfrastructureUtils.bindLocalH2("DataSource");
 }
 
 public static void main(String[] args) throws Exception {
  //setup jndi
  setupJNDI();
  //start the app context
  ApplicationContext context = new ClassPathXmlApplicationContext(args);
  //print a statement saying "it's up"
  logger.info(RemoteJVMEmulator.class.getSimpleName() + " is up " + context.getStartupDate());
 }

}

and here's the test runner.  this uses the infrastructure utilities to startup the shared services and bind them locally.  it then uses and inner class to build and run and Ant task that will kick off the RemoteJVMEmulator class.  a couple of things to note about it though;
- the Ant task is started in a separate thread - mainly because it blocks
- the Ant task is started first to act as the 'remote client'/'listener'
- the 'remote client'/'listener' needs to be setup such that you don't need to be in the same JVM to access any information from it.  (treat it as a 'remote' server deployed and running without any interaction services - you could connect to it via JMX or expose other services, that's up to you)

package de.incompleteco.spring.batch.ha;

import static org.junit.Assert.assertFalse;

import java.io.PrintStream;

import javax.sql.DataSource;

import org.apache.tools.ant.DefaultLogger;
import org.apache.tools.ant.DemuxOutputStream;
import org.apache.tools.ant.Project;
import org.apache.tools.ant.taskdefs.Java;
import org.apache.tools.ant.types.Path;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.mock.jndi.SimpleNamingContextBuilder;

//don't run this test in CI
@Ignore
public class TestSimpleBatchHAService {

 
 @BeforeClass
 public static void beforeClass() throws Exception {
  //start a builder
  SimpleNamingContextBuilder.emptyActivatedContextBuilder();
  //setup the services (h2 and amq)
  InfrastructureUtils.startH2();
  //setup the database
  DataSource dataSource = InfrastructureUtils.bindLocalH2("DataSource");
  setupH2Data(dataSource);
  //start amq
  InfrastructureUtils.startAMQ();
  //bind
  InfrastructureUtils.bindLocalAMQ("ConnectionFactory","request.queue","reply.queue");
 }
 
 @AfterClass
 public static void afterClass() throws Exception {
  //shutdown the services
  InfrastructureUtils.stopH2();
  InfrastructureUtils.stopAMQ();
 }
 
 private static void setupH2Data(DataSource dataSource) throws Exception {
  //execute the statements
  JdbcTemplate template = new JdbcTemplate(dataSource);
  //drop statements
  String[] statements = InfrastructureUtils.convertSqlFile("/org/springframework/batch/core/schema-drop-h2.sql");
  try {
   for (String statement : statements) {
    template.execute(statement);
   }//end for
  }
  catch (Exception e) { }
  statements = InfrastructureUtils.convertSqlFile("/META-INF/sql/schema-ext-drop-h2.sql");
  try {
   for (String statement : statements) {
    template.execute(statement);
   }//end for
  }
  catch (Exception e) { }  
  //create statements
  statements = InfrastructureUtils.convertSqlFile("/org/springframework/batch/core/schema-h2.sql");
  for (String statement : statements) {
   template.execute(statement);
  }//end for  
  //create statements
  statements = InfrastructureUtils.convertSqlFile("/META-INF/sql/schema-ext-h2.sql");
  for (String statement : statements) {
   template.execute(statement);
  }//end for    
 }
 
 
 @Test
 public void testExecute() throws Exception {
  //start a thread to run the remote
  new Thread(new RemoteJVMRunner()).start();
  //now start the 'server'
  ApplicationContext context = new ClassPathXmlApplicationContext("classpath:/META-INF/spring/server-context.xml");
  //now that it's started, run the job
  Job job = context.getBean(Job.class);
  JobParameters parameters = new JobParametersBuilder().addLong("runtime",System.currentTimeMillis()).toJobParameters();
  JobLauncher launcher = context.getBean("remoteJobLauncher", JobLauncher.class);
  JobExecution execution = launcher.run(job,parameters);
  JobExplorer explorer = context.getBean(JobExplorer.class);
  //monitor
  while (explorer.getJobExecution(execution.getId()).isRunning()) {
   Thread.sleep(500);
  }//end while
  //reload the execution
  execution = explorer.getJobExecution(execution.getId());
  //check
  assertFalse(execution.getStatus().isUnsuccessful());
 }

 
 class RemoteJVMRunner implements Runnable {
  
  @Override
  public void run() {
   Project project = new Project();
   project.setName("remote-jvm");
   project.init();
   //setup the logger
   DefaultLogger logger = new DefaultLogger();
   project.addBuildListener(logger);
   logger.setOutputPrintStream(System.out);
   logger.setErrorPrintStream(System.err);
   logger.setMessageOutputLevel(Project.MSG_INFO);
   System.setOut(new PrintStream(new DemuxOutputStream(project,false)));
   System.setErr(new PrintStream(new DemuxOutputStream(project,true)));
   //start the project
   project.fireBuildStarted();
   
   Java java = new Java();
   java.setProject(project);;
   java.setTaskName("run-remote-jvm");
   java.setFork(true);
   java.setFailonerror(true);
   //set the classname
   java.setClassname(RemoteJVMEmulator.class.getName());
   java.setClasspath(new Path(project,System.getProperty("java.class.path")));
   //create arguments
   java.createArg().setValue("classpath:/META-INF/spring/client-context.xml");
   //init
   java.init();
   //execute
   java.executeJava();
  }
 }
}

Spring Batch High Availability

Spring Batch High Availability

here's a quick project i recently threw together to demonstrate how to setup a high availability cluster for processing Spring Batch.  HA doesn't mean what you typically think, but in a batch environment, it allows for a configurable period of time in which should a job 'stop running' (that is, seem to be running but the underlying JVM has failed), the job will be 'failed' and restarted immediately on another cluster. the design is as follows;

Master/Heartbeat Consumer/Launcher

Overview

- JVM that starts Jobs on itself or other nodes
- Heartbeat Consumer - runs a TCP socket server that listens for 'heartbeats' from other JVM's
- Is responsible for failing over a job to another server should the job fail where it is

Key Components

- MessageChannelJobLauncher - uses Spring Integration to serialize the Job's name and it's parameters and send it out as a start request
- BatchHAService - service to be periodically invoked and check for that executions registered are still running/stable - responsible for failing and restarting jobs that become 'ghosts'
- BatchHeartbeatConsumerService - responsible for consuming heartbeats and registering job execution ids

Client/Heartbeat Producer

Overview

- JVM that actually runs a Job
- produces Heartbeats - a list of registered/in-flight job execution ids
- uses a RemoteJobRegistry to persist job names in a shared batch database

Key Components

- JobExecutionRegisterListener - wraps any jobs and registers the job execution id in the current JVM
- JobExecutionRegisterListenerPostProcessor - responsible for implementing the Listener on all Jobs
- RemoteJobRegistry - wraps the 'JVM local' JobRegistry and persists in a shared database, any Job names on the local JVM
- BatchHeartbeatClientService - responsible for accepting jobExecution registers and publishing

The infrastructure pieces are;
- Job "start" requests are sent via JMS
- there is a shared Job Database (batch database) that includes a new table, Job_Entity for persisting job names
- TCP server/client implemented in Spring Integration to publish the Job Execution Ids as part of the heart beat

the code will be up on github soon and i'll flesh it out properly, but it really is magic what Spring Integration can do.

Spring Batch - looping over multiple files

Spring Batch - looping over multiple files

here's a quick solution to a scenario coming from this question.  essentially, an array of file names will be passed in as a job parameter and the job needs to process each of these files.  one option is to have the 'looping' part managed by the job orchestrator (it loops, gets a single file, then kicks off the job for that file) the other is to pass the array in, and then loop internally in the job.

which option to choose would be dependent on other aspects of the use case, but here's an example of looping internally in the job

here's the job definition;

now here's the decider code;

package de.incompleteco.spring.batch.decider;

import java.util.Arrays;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;

import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.job.flow.FlowExecutionStatus;
import org.springframework.batch.core.job.flow.JobExecutionDecider;

public class FileDecision implements JobExecutionDecider {

 public static final String INPUT_FILE = "input.file";
 public static final String INPUT_FILES = "input.files";
 public static final String DELIMITER = ",";
 
 private Queue < String > inputFiles;
 
 @Override
 public FlowExecutionStatus decide(JobExecution jobExecution,StepExecution stepExecution) {
  //check if the jobExecution has the input.file in it's context
  if (!jobExecution.getExecutionContext().containsKey(INPUT_FILE)) {
   //build the queue
   inputFiles = new LinkedBlockingQueue < String >(Arrays.asList(jobExecution.getJobParameters().getString(INPUT_FILES).split(DELIMITER)));
  }//end if
  //pop and add
  String file = inputFiles.poll();
  if (file != null) {
   jobExecution.getExecutionContext().put(INPUT_FILE, file);
   return FlowExecutionStatus.UNKNOWN;
  }//end if
  //return 'done'
  return FlowExecutionStatus.COMPLETED;
 }

}

our helper writer;

package de.incompleteco.spring.batch.item.writer;

import java.util.List;

import org.springframework.batch.item.ItemWriter;

public class SystemOutItemWriter implements ItemWriter < String > {

 @Override
 public void write(List items) throws Exception {
  for (String item : items) {
   System.out.println("this is what was received:" + item);
  }//end for
 }

}

and our resources.

finally, our unit test to see if it all works;

package de.incompleteco.spring.batch;

import static org.junit.Assert.assertFalse;

import java.io.File;
import java.io.FileWriter;

import javax.annotation.Resource;

import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import de.incompleteco.spring.batch.decider.FileDecision;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration("classpath:/META-INF/spring/*-context.xml")
public class MultipleFileProcessIntegrationTest {

 @Rule
 public TemporaryFolder folder = new TemporaryFolder();
 
 @Resource
 private Job job;
 
 @Resource
 private JobLauncher jobLauncher;
 
 @Resource
 private JobExplorer jobExplorer;
 
 @Test
 public void test() throws Exception {
  //somewhere to hold the filenames
  StringBuilder builder = new StringBuilder();
  //create 3 files
  for (int i=0;i<3 add="" content:="" content="" file="" filename="" filewriter="" i="" if="" some="" test="" testfile="" the="" txt="" write="" writer.close="" writer.flush="" writer.write="" writer="new"> 0) {
    builder.append(FileDecision.DELIMITER);
   }//end if
   builder.append(file.getAbsolutePath());
   //show it
   System.out.println(file.getAbsolutePath());
  }//end loop
  
  
  //now build the job parameters
  JobParameters parameters = new JobParametersBuilder().addString(FileDecision.INPUT_FILES,builder.toString()).toJobParameters();
  //execution
  JobExecution execution = jobLauncher.run(job,parameters);
  //check
  while (jobExplorer.getJobExecution(execution.getId()).isRunning()) {
   Thread.sleep(100);
  }//end while
  //load
  execution = jobExplorer.getJobExecution(execution.getId());
  //check
  assertFalse(execution.getStatus().isUnsuccessful());
 }
 
}

.

Spring Batch StringToJobLaunchRequestAdapter

Spring Batch StringToJobLaunchRequestAdapter

so i don't forget this, the proper string format for Spring Batch StringToJobLaunchRequestAdapter is

foo[bar=spam,count(long)=123]

(am always getting it wrong...)