docs/storm-jms-example.md - storm - Git at Google


            
            ---

            
            title: Storm JMS Integration

            
            layout: documentation

            
            documentation: true

            
            ---

            
            ## Example Storm JMS Topology


            The storm-jms source code contains an example project (in the "examples" directory)

            
            builds a multi-bolt/multi-spout topology (depicted below) that uses the JMS Spout and JMS Bolt components.


            ![picture alt](images/Storm-JMS-Example.png "Example JMS Topology")


            The green components represent instances of the storm-jms components. White components represent

            
            "standard" Storm bolts (in the example these bolts are instances of `GenericBolt` which simply logs

            
            information about the tuples it receives and emits).


            Grey arrows represent JMS messages, while black arrows represent the flow of Storm tuple objects.


            ### JMS Transactions and Guaranteed Processing

            
            The example is set up to be "transactional," meaning the JMS Spout will use Storm's guaranteed

            
            processing capabilities to determine if a JMS Message should be acknowledged. Each bolt in the

            
            topology will anchor to each tuple it receives. If every bolt successfully processes and acks

            
            each tuple in the chain, the original JMS Message will be acknowledged, and the underlying

            
            JMS implementation will not attempt re-delivery of the message. If a bolt fails to process/ack

            
            a tuple, the JMS message will not be acknowledged, and the JMS implementation will queue the

            
            message for redelivery.


            ### Data Flow

            
            The topology contains two chains: One originating from a JMS Spout connected to a Queue, and

            
            another originating from a JMS Spout connected to a Topic.


            **Chain #1**


            1. The "JMS Queue Spout" receives a JMS Message object from the queue, and emits a

            
            tuple to the "Intermediate Bolt"

            
            2. The "Intermediate Bolt" emits a tuple to the "Final Bolt" and the "JMS Topic Bolt", and acks

            
            the tuple it received.

            
            3. The "Final Bolt" receives the tuple and simply acks it, it does not emit anything.

            
            4. The "JMS Topic Bolt" receives a tuple, constructs a JMS Message from the tuple's values,

            
            and publishes the message to a JMS Topic.

            
            5. If the "JMS Topic Bolt" successfully publishes the JMS message, it will ack the tuple.

            
            6. The "JMS Queue Spout" will receive notification if all bolts in the chain have acked

            
            and acknowledge the original JMS Message. If one or more bolts in the chain fail to ack a tuple, the

            
            "JMS Queue Spout" will not acknowledge the JMS message.


            **Chain #2**


            1. The "JMS Topic Spout" receives a JMS message from the topic and emits a tuple to "Another Bolt."

            
            2. The "Another Bolt" receives and acks the tuple.

            
            3. The "JMS Topic Spout" acknowledges the JMS message.


            ### Building the Example Topology


            	$ cd storm-jms

            
            	$ mvn clean install


            ### Running the Example Topology Locally


            The example uses ApacheMQ 5.4.0 for JMS. Download and install it from

            
            [Apache ActiveMQ Downloads](http://activemq.apache.org/download.html).


            There is no specific configuration required, simply start ActiveMQ:


            	$ [ACTIVEMQ_HOME]/bin/activemq


            Run the example topology from the `examples` directory:


            	$ mvn exec:java


            When the topology runs, it will connect to ActiveMQ and the following JMS Destinations will be created:


            	backtype.storm.contrib.example.queue

            
            	backtype.storm.contrib.example.topic


            To publish a message to the `backtype.storm.contrib.example.queue` queue:


            1. Open the ActiveMQ Queue admin console: http://localhost:8161/admin/queues.jsp

            
            2. Click the [Send To](http://localhost:8161/admin/send.jsp?JMSDestination=backtype.storm.contrib.example.queue&JMSDestinationType=queue)

            
            link for the `backtupe.storm.example.queue` queue entry.

            
            3. On the "Send a JMS Message" form, select the "Persistent Delivery" checkbox, enter

            
            some text for the message body, and click "Send".


            In the terminal you should see the following among the output:


            	DEBUG (backtype.storm.contrib.jms.bolt.JmsBolt:183) - Connecting JMS..

            
            	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:213) - sending tuple: ActiveMQTextMessage {commandId = 5, responseRequired = true, messageId = ID:budreau.home-51286-1321074044423-2:4:1:1:1, originalDestination = null, originalTransactionId = null, producerId = ID:budreau.home-51286-1321074044423-2:4:1:1, destination = queue://backtype.storm.contrib.example.queue, transactionId = null, expiration = 0, timestamp = 1321735055910, arrival = 0, brokerInTime = 1321735055910, brokerOutTime = 1321735055921, correlationId = , replyTo = null, persistent = true, type = , priority = 0, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = org.apache.activemq.util.ByteSequence@6c27ca12, dataStructure = null, redeliveryCounter = 0, size = 0, properties = {secret=880412b7-de71-45dd-8a80-8132589ccd22}, readOnlyProperties = true, readOnlyBody = true, droppable = false, text = Hello storm-jms!}

            
            	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:219) - Requested deliveryMode: CLIENT_ACKNOWLEDGE

            
            	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:220) - Our deliveryMode: CLIENT_ACKNOWLEDGE

            
            	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:224) - Requesting acks.

            
            	DEBUG (backtype.storm.contrib.jms.example.GenericBolt:60) - [INTERMEDIATE_BOLT] Received message: source: 1:10, stream: 1, id: {-7100026097570233628=-7100026097570233628}, [Hello storm-jms!]

            
            	DEBUG (backtype.storm.contrib.jms.example.GenericBolt:66) - [INTERMEDIATE_BOLT] emitting: source: 1:10, stream: 1, id: {-7100026097570233628=-7100026097570233628}, [Hello storm-jms!]

            
            	DEBUG (backtype.storm.contrib.jms.example.GenericBolt:75) - [INTERMEDIATE_BOLT] ACKing tuple: source: 1:10, stream: 1, id: {-7100026097570233628=-7100026097570233628}, [Hello storm-jms!]

            
            	DEBUG (backtype.storm.contrib.jms.bolt.JmsBolt:136) - Tuple received. Sending JMS message.

            
            	DEBUG (backtype.storm.contrib.jms.example.GenericBolt:60) - [FINAL_BOLT] Received message: source: 2:2, stream: 1, id: {-7100026097570233628=-5393763013502927792}, [Hello storm-jms!]

            
            	DEBUG (backtype.storm.contrib.jms.example.GenericBolt:75) - [FINAL_BOLT] ACKing tuple: source: 2:2, stream: 1, id: {-7100026097570233628=-5393763013502927792}, [Hello storm-jms!]

            
            	DEBUG (backtype.storm.contrib.jms.bolt.JmsBolt:144) - ACKing tuple: source: 2:2, stream: 1, id: {-7100026097570233628=-9118586029611278300}, [Hello storm-jms!]

            
            	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:251) - JMS Message acked: ID:budreau.home-51286-1321074044423-2:4:1:1:1

            
            	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:213) - sending tuple: ActiveMQTextMessage {commandId = 5, responseRequired = true, messageId = ID:budreau.home-60117-1321735025796-0:0:1:1:1, originalDestination = null, originalTransactionId = null, producerId = ID:budreau.home-60117-1321735025796-0:0:1:1, destination = topic://backtype.storm.contrib.example.topic, transactionId = null, expiration = 0, timestamp = 1321735056258, arrival = 0, brokerInTime = 1321735056260, brokerOutTime = 1321735056260, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, text = source: 2:2, stream: 1, id: {-710002609757023... storm-jms!]}

            
            	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:219) - Requested deliveryMode: CLIENT_ACKNOWLEDGE

            
            	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:220) - Our deliveryMode: CLIENT_ACKNOWLEDGE

            
            	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:224) - Requesting acks.

            
            	DEBUG (backtype.storm.contrib.jms.example.GenericBolt:60) - [ANOTHER_BOLT] Received message: source: 5:9, stream: 1, id: {-5117078009445186058=-5117078009445186058}, [source: 2:2, stream: 1, id: {-7100026097570233628=-9118586029611278300}, [Hello storm-jms!]]

            
            	DEBUG (backtype.storm.contrib.jms.example.GenericBolt:75) - [ANOTHER_BOLT] ACKing tuple: source: 5:9, stream: 1, id: {-5117078009445186058=-5117078009445186058}, [source: 2:2, stream: 1, id: {-7100026097570233628=-9118586029611278300}, [Hello storm-jms!]]

            
            	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:251) - JMS Message acked: ID:budreau.home-60117-1321735025796-0:0:1:1:1


            The topology will run for 2 minutes, then gracefully shut down.
	---
	title: Storm JMS Integration
	layout: documentation
	documentation: true
	---
	## Example Storm JMS Topology

	The storm-jms source code contains an example project (in the "examples" directory)
	builds a multi-bolt/multi-spout topology (depicted below) that uses the JMS Spout and JMS Bolt components.

	![picture alt](images/Storm-JMS-Example.png "Example JMS Topology")

	The green components represent instances of the storm-jms components. White components represent
	"standard" Storm bolts (in the example these bolts are instances of `GenericBolt` which simply logs
	information about the tuples it receives and emits).

	Grey arrows represent JMS messages, while black arrows represent the flow of Storm tuple objects.

	### JMS Transactions and Guaranteed Processing
	The example is set up to be "transactional," meaning the JMS Spout will use Storm's guaranteed
	processing capabilities to determine if a JMS Message should be acknowledged. Each bolt in the
	topology will anchor to each tuple it receives. If every bolt successfully processes and acks
	each tuple in the chain, the original JMS Message will be acknowledged, and the underlying
	JMS implementation will not attempt re-delivery of the message. If a bolt fails to process/ack
	a tuple, the JMS message will not be acknowledged, and the JMS implementation will queue the
	message for redelivery.

	### Data Flow
	The topology contains two chains: One originating from a JMS Spout connected to a Queue, and
	another originating from a JMS Spout connected to a Topic.

	Chain #1

	1. The "JMS Queue Spout" receives a JMS Message object from the queue, and emits a
	tuple to the "Intermediate Bolt"
	2. The "Intermediate Bolt" emits a tuple to the "Final Bolt" and the "JMS Topic Bolt", and acks
	the tuple it received.
	3. The "Final Bolt" receives the tuple and simply acks it, it does not emit anything.
	4. The "JMS Topic Bolt" receives a tuple, constructs a JMS Message from the tuple's values,
	and publishes the message to a JMS Topic.
	5. If the "JMS Topic Bolt" successfully publishes the JMS message, it will ack the tuple.
	6. The "JMS Queue Spout" will receive notification if all bolts in the chain have acked
	and acknowledge the original JMS Message. If one or more bolts in the chain fail to ack a tuple, the
	"JMS Queue Spout" will not acknowledge the JMS message.

	Chain #2

	1. The "JMS Topic Spout" receives a JMS message from the topic and emits a tuple to "Another Bolt."
	2. The "Another Bolt" receives and acks the tuple.
	3. The "JMS Topic Spout" acknowledges the JMS message.


	### Building the Example Topology

	$ cd storm-jms
	$ mvn clean install



	### Running the Example Topology Locally

	The example uses ApacheMQ 5.4.0 for JMS. Download and install it from
	[Apache ActiveMQ Downloads](http://activemq.apache.org/download.html).

	There is no specific configuration required, simply start ActiveMQ:

	$ [ACTIVEMQ_HOME]/bin/activemq

	Run the example topology from the `examples` directory:

	$ mvn exec:java

	When the topology runs, it will connect to ActiveMQ and the following JMS Destinations will be created:

	backtype.storm.contrib.example.queue
	backtype.storm.contrib.example.topic

	To publish a message to the `backtype.storm.contrib.example.queue` queue:

	1. Open the ActiveMQ Queue admin console: http://localhost:8161/admin/queues.jsp
	2. Click the [Send To](http://localhost:8161/admin/send.jsp?JMSDestination=backtype.storm.contrib.example.queue&JMSDestinationType=queue)
	link for the `backtupe.storm.example.queue` queue entry.
	3. On the "Send a JMS Message" form, select the "Persistent Delivery" checkbox, enter
	some text for the message body, and click "Send".


	In the terminal you should see the following among the output:

	DEBUG (backtype.storm.contrib.jms.bolt.JmsBolt:183) - Connecting JMS..
	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:213) - sending tuple: ActiveMQTextMessage {commandId = 5, responseRequired = true, messageId = ID:budreau.home-51286-1321074044423-2:4:1:1:1, originalDestination = null, originalTransactionId = null, producerId = ID:budreau.home-51286-1321074044423-2:4:1:1, destination = queue://backtype.storm.contrib.example.queue, transactionId = null, expiration = 0, timestamp = 1321735055910, arrival = 0, brokerInTime = 1321735055910, brokerOutTime = 1321735055921, correlationId = , replyTo = null, persistent = true, type = , priority = 0, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = org.apache.activemq.util.ByteSequence@6c27ca12, dataStructure = null, redeliveryCounter = 0, size = 0, properties = {secret=880412b7-de71-45dd-8a80-8132589ccd22}, readOnlyProperties = true, readOnlyBody = true, droppable = false, text = Hello storm-jms!}
	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:219) - Requested deliveryMode: CLIENT_ACKNOWLEDGE
	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:220) - Our deliveryMode: CLIENT_ACKNOWLEDGE
	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:224) - Requesting acks.
	DEBUG (backtype.storm.contrib.jms.example.GenericBolt:60) - [INTERMEDIATE_BOLT] Received message: source: 1:10, stream: 1, id: {-7100026097570233628=-7100026097570233628}, [Hello storm-jms!]
	DEBUG (backtype.storm.contrib.jms.example.GenericBolt:66) - [INTERMEDIATE_BOLT] emitting: source: 1:10, stream: 1, id: {-7100026097570233628=-7100026097570233628}, [Hello storm-jms!]
	DEBUG (backtype.storm.contrib.jms.example.GenericBolt:75) - [INTERMEDIATE_BOLT] ACKing tuple: source: 1:10, stream: 1, id: {-7100026097570233628=-7100026097570233628}, [Hello storm-jms!]
	DEBUG (backtype.storm.contrib.jms.bolt.JmsBolt:136) - Tuple received. Sending JMS message.
	DEBUG (backtype.storm.contrib.jms.example.GenericBolt:60) - [FINAL_BOLT] Received message: source: 2:2, stream: 1, id: {-7100026097570233628=-5393763013502927792}, [Hello storm-jms!]
	DEBUG (backtype.storm.contrib.jms.example.GenericBolt:75) - [FINAL_BOLT] ACKing tuple: source: 2:2, stream: 1, id: {-7100026097570233628=-5393763013502927792}, [Hello storm-jms!]
	DEBUG (backtype.storm.contrib.jms.bolt.JmsBolt:144) - ACKing tuple: source: 2:2, stream: 1, id: {-7100026097570233628=-9118586029611278300}, [Hello storm-jms!]
	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:251) - JMS Message acked: ID:budreau.home-51286-1321074044423-2:4:1:1:1
	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:213) - sending tuple: ActiveMQTextMessage {commandId = 5, responseRequired = true, messageId = ID:budreau.home-60117-1321735025796-0:0:1:1:1, originalDestination = null, originalTransactionId = null, producerId = ID:budreau.home-60117-1321735025796-0:0:1:1, destination = topic://backtype.storm.contrib.example.topic, transactionId = null, expiration = 0, timestamp = 1321735056258, arrival = 0, brokerInTime = 1321735056260, brokerOutTime = 1321735056260, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, text = source: 2:2, stream: 1, id: {-710002609757023... storm-jms!]}
	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:219) - Requested deliveryMode: CLIENT_ACKNOWLEDGE
	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:220) - Our deliveryMode: CLIENT_ACKNOWLEDGE
	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:224) - Requesting acks.
	DEBUG (backtype.storm.contrib.jms.example.GenericBolt:60) - [ANOTHER_BOLT] Received message: source: 5:9, stream: 1, id: {-5117078009445186058=-5117078009445186058}, [source: 2:2, stream: 1, id: {-7100026097570233628=-9118586029611278300}, [Hello storm-jms!]]
	DEBUG (backtype.storm.contrib.jms.example.GenericBolt:75) - [ANOTHER_BOLT] ACKing tuple: source: 5:9, stream: 1, id: {-5117078009445186058=-5117078009445186058}, [source: 2:2, stream: 1, id: {-7100026097570233628=-9118586029611278300}, [Hello storm-jms!]]
	DEBUG (backtype.storm.contrib.jms.spout.JmsSpout:251) - JMS Message acked: ID:budreau.home-60117-1321735025796-0:0:1:1:1

	The topology will run for 2 minutes, then gracefully shut down.