Pulsar adaptor for Apache Storm

Pulsar Storm is an adaptor for integrating with Apache Storm topologies. It provides core Storm implementations for sending and receiving data.

An application can inject data into a Storm topology via a generic Pulsar spout, as well as consume data from a Storm topology via a generic Pulsar bolt.

Using the Pulsar Storm Adaptor

To use the Pulsar Storm Adaptor, you need to include dependency for Pulsar Storm Adaptor:

  1. <dependency>
  2.   <groupId>org.apache.pulsar</groupId>
  3.   <artifactId>pulsar-storm</artifactId>
  4.   <version>${pulsar.version}</version>
  5. </dependency>

Pulsar Spout

The Pulsar Spout allows for the data published on a topic to be consumed by a Storm topology. It emits a Storm tuple based on the message received and the MessageToValuesMapper provided by the client.

The tuples that fail to be processed by the downstream bolts will be re-injected by the spout with an exponential backoff, within a configurable timeout (the default is 60 seconds) or a configurable number of retries, whichever comes first, after which it is acknowledged by the consumer. Here’s an example construction of a spout:

  1. MessageToValuesMapper messageToValuesMapper = new MessageToValuesMapper() {
  3.     @Override
  4.     public Values toValues(Message msg) {
  5.         return new Values(new String(msg.getData()));
  6.     }
  8.     @Override
  9.     public void declareOutputFields(OutputFieldsDeclarer declarer) {
  10.         // declare the output fields
  11.         declarer.declare(new Fields("string"));
  12.     }
  13. };
  15. // Configure a Pulsar Spout
  16. PulsarSpoutConfiguration spoutConf = new PulsarSpoutConfiguration();
  17. spoutConf.setServiceUrl("pulsar://broker.messaging.usw.example.com:6650");
  18. spoutConf.setTopic("persistent://my-property/usw/my-ns/my-topic1");
  19. spoutConf.setSubscriptionName("my-subscriber-name1");
  20. spoutConf.setMessageToValuesMapper(messageToValuesMapper);
  22. // Create a Pulsar Spout
  23. PulsarSpout spout = new PulsarSpout(spoutConf);

For a complete example, click here.

Pulsar Bolt

The Pulsar bolt allows data in a Storm topology to be published on a topic. It publishes messages based on the Storm tuple received and the TupleToMessageMapper provided by the client.

A partitioned topic can also be used to publish messages on different topics. In the implementation of the TupleToMessageMapper, a “key” will need to be provided in the message which will send the messages with the same key to the same topic. Here’s an example bolt:

  1. TupleToMessageMapper tupleToMessageMapper = new TupleToMessageMapper() {
  3.     @Override
  4.     public TypedMessageBuilder<byte[]> toMessage(TypedMessageBuilder<byte[]> msgBuilder, Tuple tuple) {
  5.         String receivedMessage = tuple.getString(0);
  6.         // message processing
  7.         String processedMsg = receivedMessage + "-processed";
  8.         return msgBuilder.value(processedMsg.getBytes());
  9.     }
  11.     @Override
  12.     public void declareOutputFields(OutputFieldsDeclarer declarer) {
  13.         // declare the output fields
  14.     }
  15. };
  17. // Configure a Pulsar Bolt
  18. PulsarBoltConfiguration boltConf = new PulsarBoltConfiguration();
  19. boltConf.setServiceUrl("pulsar://broker.messaging.usw.example.com:6650");
  20. boltConf.setTopic("persistent://my-property/usw/my-ns/my-topic2");
  21. boltConf.setTupleToMessageMapper(tupleToMessageMapper);
  23. // Create a Pulsar Bolt
  24. PulsarBolt bolt = new PulsarBolt(boltConf);