Package org.apache.beam.sdk.io.jms

Class JmsIO

java.lang.Object
org.apache.beam.sdk.io.jms.JmsIO
public class JmsIO extends Object
An unbounded source for JMS destinations (queues or topics).

Reading from a JMS destination

JmsIO source returns unbounded collection of JMS records as PCollection<JmsRecord>. A JmsRecord includes JMS headers and properties, along with the JMS TextMessage payload.

To configure a JMS source, you have to provide a ConnectionFactory and the destination (queue or topic) where to consume. The following example illustrates various options for configuring the source: 


 pipeline.apply(JmsIO.read()
    .withConnectionFactory(myConnectionFactory)
    .withQueue("my-queue")
    // above two are required configuration, returns PCollection<JmsRecord>

    // rest of the settings are optional

It is possible to read any type of JMS Message into a custom POJO using the following configuration: 


 pipeline.apply(JmsIO.<T>readMessage()
    .withConnectionFactory(myConnectionFactory)
    .withQueue("my-queue")
    .withMessageMapper((MessageMapper<T>) message -> {
      // code that maps message to T
    })
    .withCoder(
      // a coder for T
    )

Acknowledgment Modes and Client Prefetch Configuration

By default, JmsIO consumes messages using JmsIO.AcknowledgeMode.CLIENT_ACKNOWLEDGE where a new Session is created for each checkpoint to prevent premature acknowledgments across bundles. When using JmsIO.AcknowledgeMode.CLIENT_ACKNOWLEDGE, if your JMS broker or client library utilizes client-side message prefetch buffers (such as Apache ActiveMQ), you should configure prefetch=0 on your ConnectionFactory (e.g., via ?jms.prefetchPolicy.all=0 in the broker URL or ActiveMQPrefetchPolicy.setAll(0)). Otherwise, unconsumed messages could be held inside old consumers in low throughput scenario and could lead to message backlog.

Alternatively, if your JMS broker supports individual message acknowledgment (such as ActiveMQ or Amazon MQ ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE = 4), you can specify JmsIO.Read.withAcknowledgeMode(AcknowledgeMode) with JmsIO.AcknowledgeMode.INDIVIDUAL_ACKNOWLEDGE. In this mode, a single shared session and consumer are reused across all checkpoints.

Writing to a JMS destination

JmsIO sink supports writing text messages to a JMS destination on a broker. To configure a JMS sink, you must specify a ConnectionFactory and a Destination name. For instance: 


 pipeline
   .apply(...) // returns PCollection<String>
   .apply(JmsIO.write()
       .withConnectionFactory(myConnectionFactory)
       .withQueue("my-queue")