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 Integration, ActiveMQ Dead Letter Queue and Testing

Spring Integration, ActiveMQ Dead Letter Queue and Testing

recently i was working on a design whereby the application would consume messages from a JMS queue, but may encounter a transient exception (e.g. the database connection may not be available temporarily).  in Spring Integration i could use a retry pattern on the endpoint, but as the application is in a cluster, another JVM may have better luck getting to the database.  so, to use the cluster design, if a transient exception should occur, the exception would be thrown up, the transaction would roll back and the message would go back on the queue, awaiting consumption again.  this would allow another node in the cluster an opportunity to get the message and process it.

ok, so this is all very nice, but how do we confirm that this actually works?  we're using ActiveMQ in the test environment (embedded) and wanted to see the message actually got rolled back.  the default behavior of ActiveMQ, (since version 5.4?) is the following;

• no delay in retries
• 6 attempts
• if all attempt fail, put it on the Dead Letter Queue

so, here's a couple of snippets that use this to test.

first the integration setup;

***note: a DefaultMessageListenerContainer is manually configured to support the transactions, setting the transactionManager directly on the channel adapter alone doesn't work (in 2.2.4.RELEASE)

now our test resources;

our testing bean;

package de.incompleteco.spring.integration.service;

public class FailureService {

 public void process(Object payload) {
  throw new RuntimeException("simulated failure");
 }
 
}

and our junit test;

package de.incompleteco.spring.integration;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

import javax.jms.ConnectionFactory;
import javax.jms.Message;
import javax.jms.Queue;
import javax.jms.TextMessage;

import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.test.context.ActiveProfiles;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

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

 private static final String PAYLOAD = "hello world";
 
 @Autowired
 private ConnectionFactory connectionFactory;
 
 @Autowired
 @Qualifier("request.queue")
 private Queue requestQueue;
 
 @Autowired
 @Qualifier("dead.letter.queue")
 private Queue dlQueue;
 
 private JmsTemplate jmsTemplate;
 
 @Before
 public void before() throws Exception {
  jmsTemplate = new JmsTemplate(connectionFactory);
  jmsTemplate.setDefaultDestination(dlQueue);
  jmsTemplate.setReceiveTimeout(2 * 1000);//about 2 seconds
 }
 
 @Test
 public void test() throws Exception {
  //send a message
  jmsTemplate.convertAndSend(requestQueue, PAYLOAD);
  //expect it to fail and, by default, be put back on the dead letter queue
  Message failedMessage = jmsTemplate.receive();
  //now check that it has the same content
  assertTrue(failedMessage instanceof TextMessage);
  assertEquals(PAYLOAD,((TextMessage) failedMessage).getText());
 }
 
}

Spring Integration JMS to Database

Spring Integration JMS to Database

this is a quick design to persist a very basic JMS message directly to a database.  this pattern could be extended to a use case where you want to very quickly (and in an ordered fashion) consume messages from a JMS queue, persist them quickly into a staging table, then raise a subsequent event for secondary processing of the messages.

anyway, here's the consumption part in Spring Integration.

now here's a sample message

MSFT,100.00,1373761697932

here's the resource configuration

and finally a JUnit test

package de.incompleteco.spring.integration;

import static org.junit.Assert.assertTrue;

import java.io.BufferedReader;
import java.io.InputStreamReader;

import javax.annotation.Resource;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Queue;
import javax.jms.Session;
import javax.sql.DataSource;

import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.core.io.DefaultResourceLoader;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

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

 @Resource
 private ConnectionFactory connectionFactory;
 
 @Resource
 private Queue queue;
 
 @Resource
 private DataSource dataSource;
 
 @Before
 public void before() throws Exception {
  new JdbcTemplate(dataSource).execute("delete from target_table");
 }
 
 @Test
 public void test() throws Exception {
  //check in the database
  int count = new JdbcTemplate(dataSource).queryForObject("select count(*) from target_table",Integer.class);
  //check
  assertTrue(count == 0);  
  
  //read in a message and send it
  BufferedReader reader = new BufferedReader(new InputStreamReader(new DefaultResourceLoader().getResource("classpath:/data/msft.msg").getInputStream()));
  final String message = reader.readLine();
  //send the message
  new JmsTemplate(connectionFactory).send(queue,new MessageCreator() {

   @Override
   public Message createMessage(Session session) throws JMSException {
    return session.createTextMessage(message);
   } });
  //wait and check
  Thread.sleep(100);
  //check in the database
  count = new JdbcTemplate(dataSource).queryForObject("select count(*) from target_table",Integer.class);
  //check
  assertTrue(count == 1);
 }

}

one thing to note is that with Spring XD on it's way, this consumption becomes a little redundant; Spring XD does it better :)

Spring Integration, WebSphere Application Server and the TopicConnectionFactory in a Cluster (Active-Active)

Spring Integration, WebSphere Application Server and the TopicConnectionFactory in a Cluster (Active-Active)

so, something i recently came across with when setting up a TopicConnectionFactory on WebSphere is the difference between Cluster level and Node level.  when setting up a QueueConnectionFactory, you can set once at the Cluster level, and all the Nodes replicate the setup and it all works fine.  This is not the case with the TopicConnectionFactory.  

The ClientID concept in a TopicConnectionFactory (that is, the unique ID for the subscription side of things) doesn't play nice across the Cluster if set at the Cluster level.  Essentially only one of the Nodes in the Cluster becomes the subscriber and receives messages where the other Nodes throw warnings that the subscription is in use.  (which is true)

To get around this, set the TopicConnectionFactory at the Node level, giving each one a unique ClientID.  All the rest of the settings can stay the same (JMS name, topic string, etc.).  A suggested format for the ClientID is [application].node.[number] to help you easily track and clean up subscriptions.

Remember, this is for an Active-Active cluster where each Node in it is Active and semi-autonomous (can handle their own processing independently).

Spring XD JMS Source using Spring Integration

Spring XD - JMS input

Spring XD is a fantastic new framework to help ingest data from a variety of inputs and process them out.  there's a bunch of 'source' points that come with Spring XD, but one that i wanted to use was JMS (grab my JMS messages and stick it in Hadoop…)  so, here's a simple one that works with Spring XD 1.0.0.M1

i'm using Spring Integration, so here's the JMS listener

here's the connection factory for the database

now this is what i need to do to add it to my Spring XD;

• add the jms-source-module.xml to $SPRING_XD/modules/source
• add the package jar to $SPRING_XD/lib
• add the following 'supporting' jars
• activemq-core-5.5.1.jar
• spring-integration-jms-2.2.3.RELEASE.jar
• spring-jms-3.2.1.RELEASE.jar
• geronimo-jms_1.1_spec-1.1.1.jar
• geronimo-j2ee-management_1.1_spec-1.0.1.jar

now start up the activemq server

$ACTIVE_MQ/bin/activemq start

and startup spring-xd

$SPRING_XD/bin/xd-singlenode

now to start it the 'listener'

curl -d "jms-source-module | log" http://localhost:8080/streams/jmstest

or if you're going to use hadoop as the consumer

curl -d "jms-source-module | hdfs --rollover=10" http://localhost:8080/streams/jmstest

and send a message (via the console is an easy way to test)

now, you should see it show up in the log file of spring xd (remember, "log" was specified after the pipe)

to see it in hadoop, try the following

./hadoop fs -ls /xd/jmstest

Spring Integration - Losing the ReplyChannel when sending over JMS

Spring Integration - Losing the ReplyChannel when sending over JMS

a scenario you may have encountered is the need to "asynchronously" get a request/response in Spring Integration via JMS.  that is, you want to send multiple, non-blocking JMS messages out, then pair them up on the way back in.

there are a couple of options to do this;

1. use a dispatcher (task executor) on your outbound channel to a JMS outbound-gateway - this will use a thread from the pool to process and wait for the reply.

2. (send) in conjunction with a message-drive-channel-adapter (receive) - this will use a thread pool on the receive.

to support the 2nd option, you can set the replyChannel to the message allowing you to 'route responses'.  an example of this is in org.springframework.batch.integration.partition.MessageChannelPartitionHandler from the Spring Batch Integration project.  the rub is, the replyChannel gets 'lost' when sending over JMS; it doesn't get translated to the wire.  

now nor should it.  the consumer of the message on the other end may be Spring Integration driven too and have it's own replyChannel settings.  so, one solution may be to use an extended HeaderMapper.  

the intent is to map any dynamic channels into the applicationContext (beanFactory as a singleton).  then, set that value into a new JMS-compatible format (String).  this header field will be 'left alone' by the consumer and returned.  on it's return, the replyChannel object is looked up and set to the replyChannel header value, allowing Spring Integration to appropriately route.

here's an example of the header

package de.incompleteco.spring.integration.jms.support;

import java.util.Map;

import javax.jms.JMSException;
import javax.jms.Message;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.jms.DefaultJmsHeaderMapper;

public class JMSReplyChannelJmsHeaderMapper extends DefaultJmsHeaderMapper implements ApplicationContextAware {

 private final Log logger = LogFactory.getLog(this.getClass());
 
 public static final String SENDER_REPLY_CHANNEL_NAME = "senderReplyChannelName";
 
 private ApplicationContext applicationContext;
 
 @Override
 public void fromHeaders(MessageHeaders headers, Message jmsMessage) {
  //process the parent normally
  super.fromHeaders(headers, jmsMessage);
  //now check if the headers contains a replyChannel
  if (headers.getReplyChannel() != null) {
   //now process it
   Object replyChannel = headers.getReplyChannel();
   String replyChannelName = null;
   //check
   if (replyChannel instanceof String) {
    //it's the name
    replyChannelName = replyChannel.toString();
   } else {
    //test if the object exists in the app context already
    if (!applicationContext.containsBean(replyChannel.toString())) {
     //generate a name --> use the id
     replyChannelName = headers.getId().toString();
     //set into the app context
     ((ConfigurableApplicationContext)applicationContext).getBeanFactory().registerSingleton(replyChannelName, replyChannel);     
    } else {
     replyChannelName = replyChannel.toString();
    }//end if
   }//end if
   //now add a message header
   try {
    jmsMessage.setStringProperty(SENDER_REPLY_CHANNEL_NAME, replyChannelName);
   } catch (JMSException e) {
    logger.info("failed to set senderReplyChannelName, skipping", e);
   }
  }//end if
 }

 @Override
 public Map toHeaders(Message jmsMessage) {
  Map headers = super.toHeaders(jmsMessage);
  //check if there's a SENDER_REPLY_CHANNEL_NAME
  try {
   if (jmsMessage.getStringProperty(SENDER_REPLY_CHANNEL_NAME) != null) {
    //check if the name maps to the app context
    String replyChannelName = jmsMessage.getStringProperty(SENDER_REPLY_CHANNEL_NAME);
    if (applicationContext.containsBean(replyChannelName)) {
     //got it
     logger.info("setting: " + replyChannelName + "as reply channel");
     headers.put(MessageHeaders.REPLY_CHANNEL,applicationContext.getBean(replyChannelName));
    }//end if
   }
  } catch (JMSException e) {
   logger.info("failed to set process the senderReplyChannelName, skipping", e);
  }
  //return
  return headers;
 }

 @Override
 public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
  this.applicationContext = applicationContext;
 }

}

and here's a rather long configuration using the MessageChannelPartitionHandler

Spring Integration, Spring JMS, ActiveMQ and testing Topics

Spring Integration, Spring JMS, ActiveMQ and testing Topics

so everyone knows how to test a JMS Queue in JUnit, ActiveMQ and - to make life easy - Spring JmsTemplate.  but, testing a Topic is a slightly different matter.

first, a topic subscriber has to be 'connected' before you can publish the message.  that's a bit of a challenge to setup.  but, with a little Spring Integration to the rescue we can get a subscriber sorted and sne a message via the JmsTemplate.

another catch comes from ActiveMQ.  to test queues, you can setup the ActiveMQConnectionFactory and you're done.  however, with Topics, you'll need the PooledConnectionFactory as well.

first, a couple dependencies for our pom

   
    org.apache.activemq
    activemq-core
    5.6.0
    test
   
   
    org.apache.activemq
    activemq-pool
    5.6.0
    test
   

then a little Spring Integration to listen to the Topic

(i prefer a queue channel, just to make it easier to get the messages off in testing)
so the next part would be the test class

package de.incompleteco.spring.integration.test;

import static org.junit.Assert.assertNotNull;

import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Session;
import javax.jms.Topic;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.Message;
import org.springframework.integration.core.PollableChannel;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

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

 @Autowired
 private PollableChannel receiveChannel;
 
 @Autowired
 private ConnectionFactory connectionFactory;
 
 @Autowired
 private Topic topic;
 
 private JmsTemplate jmsTemplate;
 
 @Test
 public void test() throws Exception {
  //setup the jmsTemplate and send
  jmsTemplate = new JmsTemplate(connectionFactory);
  jmsTemplate.setDefaultDestination(topic);
  jmsTemplate.setPubSubDomain(true);//it's a topic
  //send
  jmsTemplate.send(new MessageCreator() {

   public javax.jms.Message createMessage(Session session) throws JMSException {
    return session.createTextMessage("hello world");
   } });
  //now grab it off the queue
  Message message = receiveChannel.receive(1000);//let's put a timeout
  //test
  assertNotNull(message);
 }
 
}

and here's the specific apache configuration to support Topics