org.thales.punch.libraries.storm.spout

Class AbstractKafkaNode

java.lang.Object

org.apache.storm.topology.base.BaseComponent

org.apache.storm.topology.base.BaseRichSpout

org.thales.punch.libraries.storm.api.BaseInputNode

org.thales.punch.libraries.storm.spout.AbstractKafkaNode

All Implemented Interfaces:

Serializable, org.apache.storm.spout.ISpout, org.apache.storm.topology.IComponent, org.apache.storm.topology.IRichSpout, org.thales.punch.kafka.api.IRecordHandler<byte[],byte[]>

Direct Known Subclasses:

KafkaInput

public abstract class AbstractKafkaNode
extends org.thales.punch.libraries.storm.api.BaseInputNode
implements org.thales.punch.kafka.api.IRecordHandler<byte[],byte[]>

Common kafka input node base class with standard kafka properties definition.

This is the base class for both the regular and the batch Kafka node. Refer to the KafkaInput documentation.

Both nodes accept most of the standard Kafka consumer properties. Refer to the list of kafka properties available. Here is a quicklist with their (Kafka) default values.

        auto.commit.interval.ms = 5000
 auto.offset.reset = latest
 check.crcs = true
 client.id =
 connections.max.idle.ms = 540000
 enable.auto.commit = false
 exclude.internal.topics = true
 fetch.max.bytes = 52428800
 fetch.max.wait.ms = 500
 fetch.min.bytes = 1
 group.id = mytenant.apache_httpd.test
 heartbeat.interval.ms = 3000
 interceptor.classes = []
 internal.leave.group.on.close = true
 isolation.level = read_uncommitted
 max.partition.fetch.bytes = 1048576
 max.poll.interval.ms = 300000
 max.poll.records = 500
 metadata.max.age.ms = 300000
 metric.reporters = []
 metrics.num.samples = 2
 metrics.recording.level = INFO
 metrics.sample.window.ms = 30000
 partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
 receive.buffer.bytes = 65536
 reconnect.backoff.max.ms = 1000
 reconnect.backoff.ms = 50
 request.timeout.ms = 305000
 retry.backoff.ms = 100
 sasl.jaas.config = null
 sasl.kerberos.kinit.cmd = /usr/bin/kinit
 sasl.kerberos.min.time.before.relogin = 60000
 sasl.kerberos.service.name = null
 sasl.kerberos.ticket.renew.jitter = 0.05
 sasl.kerberos.ticket.renew.window.factor = 0.8
 sasl.mechanism = GSSAPI
 security.protocol = PLAINTEXT
 send.buffer.bytes = 131072
 session.timeout.ms = 30000
 ssl.cipher.suites = null
 ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
 ssl.endpoint.identification.algorithm = null
 ssl.key.password = null
 ssl.keymanager.algorithm = SunX509
 ssl.keystore.location = null
 ssl.keystore.password = null
 ssl.keystore.type = JKS
 ssl.protocol = TLS
 ssl.provider = null
 ssl.secure.random.implementation = null
 ssl.trustmanager.algorithm = PKIX
 ssl.truststore.location = null
 ssl.truststore.password = null
 ssl.truststore.type = JKS
 value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
 

The punch spout will overwrite the following defaults :

  auto.offset.reset to earliest.
  enable.auto.commit to false.
  fetch.max.bytes to 1048576 (1Mb) only for the regular KafkaSpout
 

Author:

dimi

See Also:

Serialized Form

Field Summary

Fields

Modifier and Type	Field and Description
protected org.thales.punch.libraries.storm.api.StreamDeclaration	errorStream The error stream is used to forward error documents
protected boolean	failSleep True to make the input_node sleep in case it receives a failed tuple
protected boolean	failStop True to make the input_node exit in case it receives a failed tuple
protected long	failuresDelayMs Time to sleep in ms after it receives a failed tuple

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

collector, exitCondition, latencyRecordSender, loadController, metricContext, myself, nodeSettings

Constructor Summary

Constructors

Constructor and Description
AbstractKafkaNode(org.thales.punch.libraries.storm.api.NodeSettings spoutConfig, String kafkaClusterId, org.apache.logging.log4j.Logger subLogger) Create a new Kafka spout

Method Summary

Modifier and Type	Method and Description
protected org.thales.punch.libraries.storm.spout.impl.kafka.TupleId	getTupleId(Object o, boolean acked) Every tuple is acked or failed, even when working with batches.
void	nextTuple()
void	open(Map conf, org.apache.storm.task.TopologyContext topologyContext, org.apache.storm.spout.SpoutOutputCollector collector)
protected boolean	process(Object attachment, org.apache.kafka.clients.consumer.ConsumerRecord<byte[],byte[]> record, org.thales.punch.libraries.storm.spout.impl.kafka.KafkaBatchAttachement batchAttch) Process a received record.

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

ack, close, deactivate, declareOutputFields, fail, getPublishedStreams, regulate, sendLatencyRecord

Methods inherited from class org.apache.storm.topology.base.BaseRichSpout

activate

Methods inherited from class org.apache.storm.topology.base.BaseComponent

getComponentConfiguration

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.thales.punch.kafka.api.IRecordHandler

onPartitionAssigned, onPartitionRevoked, onReceive, onTick

Methods inherited from interface org.thales.punch.libraries.storm.api.ISpout

registerNextTupleCallback

Methods inherited from interface org.apache.storm.spout.ISpout

activate

Methods inherited from interface org.apache.storm.topology.IComponent

getComponentConfiguration

Field Detail

errorStream

protected org.thales.punch.libraries.storm.api.StreamDeclaration errorStream

The error stream is used to forward error documents

failStop

protected boolean failStop

True to make the input_node exit in case it receives a failed tuple

failSleep

protected boolean failSleep

True to make the input_node sleep in case it receives a failed tuple

failuresDelayMs

protected final long failuresDelayMs

Time to sleep in ms after it receives a failed tuple

Constructor Detail

AbstractKafkaNode

public AbstractKafkaNode(org.thales.punch.libraries.storm.api.NodeSettings spoutConfig,
                         String kafkaClusterId,
                         org.apache.logging.log4j.Logger subLogger)

Create a new Kafka spout

Parameters:

spoutConfig - the punchplatform spout configuration. It includes the declared streams and fields.

kafkaClusterId - an id of the used kafka cluster. This is used for metrics namings

subLogger - a logger to make it easier to keep track of the child class

Method Detail

open

public void open(Map conf,
                 org.apache.storm.task.TopologyContext topologyContext,
                 org.apache.storm.spout.SpoutOutputCollector collector)

Specified by:

open in interface org.apache.storm.spout.ISpout

Overrides:

open in class org.thales.punch.libraries.storm.api.BaseInputNode

nextTuple

public void nextTuple()

Specified by:

nextTuple in interface org.apache.storm.spout.ISpout

process

protected boolean process(Object attachment,
                          org.apache.kafka.clients.consumer.ConsumerRecord<byte[],byte[]> record,
                          org.thales.punch.libraries.storm.spout.impl.kafka.KafkaBatchAttachement batchAttch)

Process a received record. Watch out : we must either emit the message into Storm, or not if we decide not to send it in, but in all case it must eventually be acknowledged. Use the return value to know what to do

Parameters:

attachment - the attachment object

record - the input kafka record

batchAttch - optional additional longs should it be required to add batch information

Returns:

true if the record is in the pipe of being emitted.

getTupleId

protected org.thales.punch.libraries.storm.spout.impl.kafka.TupleId getTupleId(Object o,
                                                                               boolean acked)

Every tuple is acked or failed, even when working with batches.

Parameters:

o - the object attached to the acked?failed tuple

acked - true if the tuple is acked

Returns:

the tupleId associated to this LoadControlItem