org.thales.punch.libraries.storm.bolt

Class RetentionNode

java.lang.Object

org.thales.punch.libraries.storm.api.BaseProcessingNode

org.thales.punch.libraries.storm.bolt.RetentionNode

All Implemented Interfaces:

Serializable, org.apache.storm.task.IBolt, org.apache.storm.topology.IComponent, org.apache.storm.topology.IRichBolt

public class RetentionNode
extends org.thales.punch.libraries.storm.api.BaseProcessingNode

Overview

The Retention Bolt stores all incoming tuples into a memory table a until a defined event arrives. It then let them flow downstream

This bolt can subscribe to one or several Storm stream(s).

Problems and limits

WARNING: To work properly, this bolt need the topology setting variable 'topology.enable.message.timeouts' to be set to 'false'. This disable the spout tuple re-emitting on timeouts. Thanks to this parameter, we deal with the tuple duplication issue.

Limit: Currently, all the pending tuples are stored in RAM so you are limited by the JVM size. Before using this bolt in production, you should test it (for example by sending a large number of tuple) and see when it fails.

Bolt settings

property	default	description
trigger_field	MANDATORY	Name of the Storm stream field containing the expected value to trigger
trigger_value	MANDATORY	The value that the variable received have to match to trigger the Tuples flushing
batch_id_field	"default"	(Optional) Create multiple "Tuple flushing list" based on the value received. This way, you can flush only tuples associated to a specific ID
retention_sec	0	The maximum time allowed between the last event received and the current tick tuple in seconds. If this limit is reach, the trigger is pulled : all tuples are acked and (depending on the 'emit_on_expired' value) emitted
emit_on_expire	true	When the 'retention_sec' is reached, this boolean choose what to do with tuples. If set to 'true', each tuple will be emited and acked to the next bolt. Else if set to 'false', tuples will be acked but NOT emitted which lead to a loss of all these tuples

Storm configuration example

 {
         "tenant": ...,
         "channel": ...,
         "name": ...,
         "meta": {...},

         "spouts": [
                ...
         ],

         "bolts": [
                ...,
                {
                         "type": "retention_bolt",
                         "bolt_settings": {
                                 "trigger_field": "trigger",
                                 "trigger_value": "whateveryouwant",
                                 "batch_id_field" : "batch_id",
                                 "retention_sec" : 120,
                                 "emit_on_expire" : true
                         },
                         "storm_settings": {
                                 "executors": 1,
                                 "component": "retention_bolt",
                                 "subscribe": [{
                                 "component": "retention_kafka_spout",
                                 "stream": "logs",
                                 "grouping": "localOrShuffle"
                         }],
                         "publish": [{
                                 "stream": "logs",
                                 "fields": ["raw_log","log","local_uuid","es_index", "trigger", "batch_id"]
                                 }]
                         }
                 },
                ...
         ],

         "storm_settings": {
                 ...
                 "topology.enable.message.timeouts": false,
                 ...
         }
 }
 

See Also:

Serialized Form

Field Summary

Fields inherited from class org.thales.punch.libraries.storm.api.BaseProcessingNode

ackRate, collector, componentId, every, failRate, metricContext, nodeSettings, topologyContext, traversalTime

Constructor Summary

Constructors

Constructor and Description
RetentionNode(org.thales.punch.libraries.storm.api.NodeSettings boltSettings, org.thales.punch.libraries.storm.api.ITopologySettings topoConfig) Create a new retention bolt

Method Summary

Modifier and Type	Method and Description
void	cleanup()
Map<String,Object>	getComponentConfiguration()
void	prepare(Map stormConf, org.apache.storm.task.TopologyContext context, org.apache.storm.task.OutputCollector collector) Prepare this bolt.
void	process(org.apache.storm.tuple.Tuple tuple) The execute method is called by storm whenever some new tuples traverse the topology.

Methods inherited from class org.thales.punch.libraries.storm.api.BaseProcessingNode

ack, declareOutputFields, enrichAndForwardMonitoringTuple, execute, fail, getNewPoint

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

RetentionNode

public RetentionNode(org.thales.punch.libraries.storm.api.NodeSettings boltSettings,
                     org.thales.punch.libraries.storm.api.ITopologySettings topoConfig)
              throws UnknownHostException

Create a new retention bolt

Parameters:

boltSettings - the bolt configuration

topoConfig - the topo configuration

Throws:

UnknownHostException - if one of your settings contains an invalid host

Method Detail

prepare

public void prepare(Map stormConf,
                    org.apache.storm.task.TopologyContext context,
                    org.apache.storm.task.OutputCollector collector)

Prepare this bolt. Use this directly only for testing, and only after setting what you need in terms of configuration.

Specified by:

prepare in interface org.apache.storm.task.IBolt

Overrides:

prepare in class org.thales.punch.libraries.storm.api.BaseProcessingNode

process

public void process(org.apache.storm.tuple.Tuple tuple)

The execute method is called by storm whenever some new tuples traverse the topology.

Specified by:

process in class org.thales.punch.libraries.storm.api.BaseProcessingNode

Parameters:

tuple - the input storm tuple.

getComponentConfiguration

public Map<String,Object> getComponentConfiguration()

Specified by:

getComponentConfiguration in interface org.apache.storm.topology.IComponent

Overrides:

getComponentConfiguration in class org.thales.punch.libraries.storm.api.BaseProcessingNode

cleanup

public void cleanup()

Specified by:

cleanup in interface org.apache.storm.task.IBolt

Overrides:

cleanup in class org.thales.punch.libraries.storm.api.BaseProcessingNode