Merge pull request #2 from SourceLabOrg/sp/exponentialBackoff

Implements an Exponential Backoff Failure Handler

Author: Crim

README.md

@@ -77,7 +77,8 @@ will handle Failed Tuples. The following implementations are provided out of th
 | Implementation | Description |
 |----------------|-------------|
 | [NoRetryHandler](src/main/java/org/sourcelab/storm/spout/redis/failhandler/NoRetryHandler.java) | Will never retry failed tuples. |
-| [RetryFailedTuples](src/main/java/org/sourcelab/storm/spout/redis/failhandler/RetryFailedTuples.java) | Rudimentary implementation that can be configured to replay failed tuples forever, or for a configured number of attempts. |
+| [ExponentialBackoffFailureHandler](src/main/java/org/sourcelab/storm/spout/redis/failhandler/ExponentialBackoffFailureHandler.java) | Will attempt to retry failed messages using an exponential backoff strategy. |
+| [RetryFailedTuples](src/main/java/org/sourcelab/storm/spout/redis/failhandler/RetryFailedTuples.java) | Rudimentary implementation that can be configured to replay failed tuples for a configured number of attempts. |
 
 #### Example Topology
 

src/main/java/org/sourcelab/storm/spout/redis/FailureHandler.java

@@ -8,12 +8,22 @@
  * Does NOT need to be thread safe as only accessed via a single thread.
  */
 public interface FailureHandler extends Serializable {
- void open(Map<String, Object> stormConfig);
+ /**
+ * Lifecycle method. Called once the Spout has started.
+ * @param stormConfig Configuration map passed from the spout.
+ */
+ void open(final Map<String, Object> stormConfig);
 
  /**
  * Handle a failed message.
- * If the implementation wants to replay this tuple again in the future, it should return value of TRUE.
- * If the implementation does NOT want to handle/replay this tuple in the future, it should return a value of FALSE.
+ * A return value of TRUE means:
+ * This implementation wants to replay this message again in the future.
+ * The spout will NOT mark the message as processed/ack the message.
+ *
+ * A return value of FALSE means:
+ * This implementation does NOT want to replay this message in the future.
+ * The spout WILL mark the message as processed/ack the message.
+ *
  * @param message The failed message.
  * @return True if the implementation will replay this tuple again later, false if not.
  */

src/main/java/org/sourcelab/storm/spout/redis/Message.java

@@ -31,6 +31,23 @@ public Map<String, String> getBody() {
  return body;
  }
 
+ @Override
+ public boolean equals(final Object other) {
+ if (this == other) {
+ return true;
+ }
+ if (other == null || getClass() != other.getClass()) {
+ return false;
+ }
+ final Message message = (Message) other;
+ return getId().equals(message.getId());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(id);
+ }
+
  @Override
  public String toString() {
  return "Message{"

src/main/java/org/sourcelab/storm/spout/redis/failhandler/ExponentialBackoffConfig.java

@@ -0,0 +1,159 @@
+package org.sourcelab.storm.spout.redis.failhandler;
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Configuration for {@link ExponentialBackoffFailureHandler}.
+ */
+public class ExponentialBackoffConfig implements Serializable {
+ /**
+ * Define our retry limit.
+ * A value of less than 0 will mean we'll retry forever
+ * A value of 0 means we'll never retry.
+ * A value of greater than 0 sets an upper bound of number of retries.
+ */
+ private final int retryLimit;
+
+ /**
+ * Initial delay after a tuple fails for the first time, in milliseconds.
+ */
+ private final long initialRetryDelayMs;
+
+ /**
+ * Each time we fail, double our delay, so 4, 8, 16 seconds, etc.
+ */
+ private final double retryDelayMultiplier;
+
+ /**
+ * Maximum delay between successive retries, defaults to 15 minutes.
+ */
+ private final long retryDelayMaxMs;
+
+ /**
+ * Constructor. See Builder instance.
+ */
+ public ExponentialBackoffConfig(
+ final int retryLimit,
+ final long initialRetryDelayMs,
+ final double retryDelayMultiplier,
+ final long retryDelayMaxMs
+ ) {
+ this.retryLimit = retryLimit;
+ this.initialRetryDelayMs = initialRetryDelayMs;
+ this.retryDelayMultiplier = retryDelayMultiplier;
+ this.retryDelayMaxMs = retryDelayMaxMs;
+ }
+
+ /**
+ * New Builder instance for ExponentialBackoffConfig.
+ * @return New Builder instance for ExponentialBackoffConfig.
+ */
+ public static Builder newBuilder() {
+ return new Builder();
+ }
+
+ public int getRetryLimit() {
+ return retryLimit;
+ }
+
+ public long getInitialRetryDelayMs() {
+ return initialRetryDelayMs;
+ }
+
+ public double getRetryDelayMultiplier() {
+ return retryDelayMultiplier;
+ }
+
+ public long getRetryDelayMaxMs() {
+ return retryDelayMaxMs;
+ }
+
+ @Override
+ public String toString() {
+ return "ExponentialBackoffConfig{"
+ + "retryLimit=" + retryLimit
+ + ", initialRetryDelayMs=" + initialRetryDelayMs
+ + ", retryDelayMultiplier=" + retryDelayMultiplier
+ + ", retryDelayMaxMs=" + retryDelayMaxMs
+ + '}';
+ }
+
+ /**
+ * Builder for {@link ExponentialBackoffConfig}.
+ */
+ public static final class Builder {
+ /**
+ * Default retry limit is 10 attempts.
+ */
+ private int retryLimit = 10;
+
+ /**
+ * The first time a message has failed, defines the minimum delay.
+ * Default initial delay is 2 seconds (2000 milliseconds).
+ */
+ private long initialRetryDelayMs = TimeUnit.SECONDS.toMillis(2);
+
+ /**
+ * Default multiplier to 2.0.
+ */
+ private double retryDelayMultiplier = 2.0;
+
+ /**
+ * Upperbound that we'll limit retries within.
+ * Defaults to a max of 15 minutes.
+ */
+ private long retryDelayMaxMs = TimeUnit.MINUTES.toMillis(15);
+
+ private Builder() {
+ }
+
+ public Builder withRetryForever() {
+ return withRetryLimit(-1);
+ }
+
+ public Builder withRetryNever() {
+ return withRetryLimit(0);
+ }
+
+ public Builder withRetryLimit(int retryLimit) {
+ this.retryLimit = retryLimit;
+ return this;
+ }
+
+ public Builder withInitialRetryDelay(final Duration duration) {
+ Objects.requireNonNull(duration);
+ return withInitialRetryDelayMs(duration.toMillis());
+ }
+
+ public Builder withInitialRetryDelayMs(long initialRetryDelayMs) {
+ this.initialRetryDelayMs = initialRetryDelayMs;
+ return this;
+ }
+
+ public Builder withRetryDelayMultiplier(double retryDelayMultiplier) {
+ this.retryDelayMultiplier = retryDelayMultiplier;
+ return this;
+ }
+
+ public Builder withRetryDelayMax(final Duration duration) {
+ Objects.requireNonNull(duration);
+ return withRetryDelayMaxMs(duration.toMillis());
+ }
+
+ public Builder withRetryDelayMaxMs(long retryDelayMaxMs) {
+ this.retryDelayMaxMs = retryDelayMaxMs;
+ return this;
+ }
+
+ /**
+ * Create new ExponentialBackoffConfig instance.
+ * @return ExponentialBackoffConfig.
+ */
+ public ExponentialBackoffConfig build() {
+ return new ExponentialBackoffConfig(retryLimit, initialRetryDelayMs, retryDelayMultiplier, retryDelayMaxMs);
+ }
+ }
+}