org.apache.storm.kafka.spout

Class KafkaSpoutRetryExponentialBackoff

java.lang.Object

org.apache.storm.kafka.spout.KafkaSpoutRetryExponentialBackoff

All Implemented Interfaces:

Serializable, KafkaSpoutRetryService

public class KafkaSpoutRetryExponentialBackoff
extends Object
implements KafkaSpoutRetryService

Implementation of KafkaSpoutRetryService using the exponential backoff formula. The time of the nextRetry is set as follows: nextRetry = failCount == 1 ? currentTime + initialDelay : currentTime + delayPeriod*2^(failCount-1) where failCount = 1, 2, 3, … nextRetry = Min(nextRetry, currentTime + maxDelay)

See Also:

Serialized Form

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	KafkaSpoutRetryExponentialBackoff.TimeInterval

Constructor Summary

Constructors

Constructor and Description

KafkaSpoutRetryExponentialBackoff(KafkaSpoutRetryExponentialBackoff.TimeInterval initialDelay, KafkaSpoutRetryExponentialBackoff.TimeInterval delayPeriod, int maxRetries, KafkaSpoutRetryExponentialBackoff.TimeInterval maxDelay) The time stamp of the next retry is scheduled according to the exponential backoff formula ( geometric progression): nextRetry = failCount == 1 ? currentTime + initialDelay : currentTime + delayPeriod^(failCount-1) where failCount = 1, 2, 3, … nextRetry = Min(nextRetry, currentTime + maxDelay).

Method Summary

Modifier and Type	Method and Description
Map<org.apache.kafka.common.TopicPartition,Long>	earliestRetriableOffsets()
KafkaSpoutMessageId	getMessageId(org.apache.kafka.clients.consumer.ConsumerRecord<?,?> record) Gets the KafkaSpoutMessageId for the given record.
boolean	isReady(KafkaSpoutMessageId msgId) Checks if a specific failed KafkaSpoutMessageId is ready to be retried, i.e is scheduled and has retry time that is less than current time.
boolean	isScheduled(KafkaSpoutMessageId msgId) Checks if a specific failed KafkaSpoutMessageId is scheduled to be retried.
int	readyMessageCount()
boolean	remove(KafkaSpoutMessageId msgId) Removes a message from the list of messages scheduled for retrial
boolean	retainAll(Collection<org.apache.kafka.common.TopicPartition> topicPartitions) Retains all the messages whose TopicPartition belongs to the specified Collection<TopicPartition>.
boolean	schedule(KafkaSpoutMessageId msgId) Schedules this KafkaSpoutMessageId if not yet scheduled, or updates retry time if it has already been scheduled.
String	toString()

Methods inherited from class java.lang.Object

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

Constructor Detail

KafkaSpoutRetryExponentialBackoff

public KafkaSpoutRetryExponentialBackoff(KafkaSpoutRetryExponentialBackoff.TimeInterval initialDelay,
                                         KafkaSpoutRetryExponentialBackoff.TimeInterval delayPeriod,
                                         int maxRetries,
                                         KafkaSpoutRetryExponentialBackoff.TimeInterval maxDelay)

The time stamp of the next retry is scheduled according to the exponential backoff formula ( geometric progression): nextRetry = failCount == 1 ? currentTime + initialDelay : currentTime + delayPeriod^(failCount-1) where failCount = 1, 2, 3, … nextRetry = Min(nextRetry, currentTime + maxDelay).

By specifying a value for maxRetries lower than Integer.MAX_VALUE, the user decides to sacrifice guarantee of delivery for the previous polled records in favor of processing more records.

Parameters:

initialDelay - initial delay of the first retry

delayPeriod - the time interval that is the ratio of the exponential backoff formula (geometric progression)

maxRetries - maximum number of times a tuple is retried before being acked and scheduled for commit

maxDelay - maximum amount of time waiting before retrying

Method Detail

earliestRetriableOffsets

public Map<org.apache.kafka.common.TopicPartition,Long> earliestRetriableOffsets()

Specified by:

earliestRetriableOffsets in interface KafkaSpoutRetryService

Returns:

The earliest retriable offset for each TopicPartition that has offsets ready to be retried, i.e. for which a tuple has failed and has retry time less than current time

isReady

public boolean isReady(KafkaSpoutMessageId msgId)

Description copied from interface: KafkaSpoutRetryService

Checks if a specific failed KafkaSpoutMessageId is ready to be retried, i.e is scheduled and has retry time that is less than current time.

Specified by:

isReady in interface KafkaSpoutRetryService

Parameters:

msgId - message to check for readiness

Returns:

true if message is ready to be retried, false otherwise

isScheduled

public boolean isScheduled(KafkaSpoutMessageId msgId)

Description copied from interface: KafkaSpoutRetryService

Checks if a specific failed KafkaSpoutMessageId is scheduled to be retried. The message may or may not be ready to be retried yet.

Specified by:

isScheduled in interface KafkaSpoutRetryService

Parameters:

msgId - message to check for scheduling status

Returns:

true if the message is scheduled to be retried, regardless of being or not ready to be retried. Returns false is this message is not scheduled for retrial

remove

public boolean remove(KafkaSpoutMessageId msgId)

Description copied from interface: KafkaSpoutRetryService

Removes a message from the list of messages scheduled for retrial

Specified by:

remove in interface KafkaSpoutRetryService

Parameters:

msgId - message to remove from retrial

Returns:

true if the message was scheduled for retrial, false otherwise

retainAll

public boolean retainAll(Collection<org.apache.kafka.common.TopicPartition> topicPartitions)

Description copied from interface: KafkaSpoutRetryService

Retains all the messages whose TopicPartition belongs to the specified Collection<TopicPartition>. All messages that come from a TopicPartition NOT existing in the collection will be removed. This method is useful to cleanup state following partition rebalance.

Specified by:

retainAll in interface KafkaSpoutRetryService

Parameters:

topicPartitions - Collection of TopicPartition for which to keep messages

Returns:

true if at least one message was removed, false otherwise

schedule

public boolean schedule(KafkaSpoutMessageId msgId)

Description copied from interface: KafkaSpoutRetryService

Schedules this KafkaSpoutMessageId if not yet scheduled, or updates retry time if it has already been scheduled. It may also indicate that the message should not be retried, in which case the message will not be scheduled.

Specified by:

schedule in interface KafkaSpoutRetryService

Parameters:

msgId - message to schedule for retrial

Returns:

true if the message will be retried, false otherwise

readyMessageCount

public int readyMessageCount()

Specified by:

readyMessageCount in interface KafkaSpoutRetryService

Returns:

The number of messages that are ready for retry

getMessageId

public KafkaSpoutMessageId getMessageId(org.apache.kafka.clients.consumer.ConsumerRecord<?,?> record)

Description copied from interface: KafkaSpoutRetryService

Gets the KafkaSpoutMessageId for the given record.

Specified by:

getMessageId in interface KafkaSpoutRetryService

Parameters:

record - The record to fetch the id for

Returns:

The id the record was scheduled for retry with, or a new KafkaSpoutMessageId if the record was not scheduled for retry.

toString

public String toString()

Overrides:

toString in class Object