cleanup

Author: Crim

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

@@ -63,6 +63,14 @@ public static Builder newBuilder() {
         return new Builder();
     }
 
+    /**
+     * Create a new default configuration.
+     * @return Default configuration.
+     */
+    public static ExponentialBackoffConfig defaultConfig() {
+        return new Builder().build();
+    }
+
     public int getRetryLimit() {
         return retryLimit;
     }

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

@@ -28,7 +28,7 @@ public class ExponentialBackoffFailureHandler implements FailureHandler {
      * Used to control timing around retries.
      * Also allows for injecting a mock clock for testing.
      */
-    private transient Clock clock = Clock.systemUTC();
+    private transient Clock clock;
 
     /**
      * This map hows how many times each messageId has failed.
@@ -253,10 +253,10 @@ public MetricHandler() {
          * @param retryQueue Our internal retry queue to collect size data.
          */
         public MetricHandler(final TopologyContext topologyContext, final TreeMap<Long, Queue<Message>> retryQueue) {
-            metricExceededRetryLimitCount = topologyContext.registerCounter("failureHandler.exceededRetryLimit");
-            metricRetriedMessagesCount = topologyContext.registerCounter("failureHandler.retriedMessages");
-            metricSuccessfulRetriedMessagesCounter = topologyContext.registerCounter("failureHandler.successfulRetriedMessages");
-            topologyContext.registerGauge("failureHandler.queuedForRetry", () ->
+            metricExceededRetryLimitCount = topologyContext.registerCounter("failureHandler_exceededRetryLimit");
+            metricRetriedMessagesCount = topologyContext.registerCounter("failureHandler_retriedMessages");
+            metricSuccessfulRetriedMessagesCounter = topologyContext.registerCounter("failureHandler_successfulRetriedMessages");
+            topologyContext.registerGauge("failureHandler_queuedForRetry", () ->
                 retryQueue.values().stream()
                 .mapToLong(Collection::size)
                 .sum()

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

@@ -27,17 +27,17 @@ public class RetryFailedTuples implements FailureHandler, Serializable {
      */
     private final Map<String, Long> messageCounter = new HashMap<>();
 
-    /**
-     * Contains a FIFO queue for failed messages.
-     */
-    private LinkedBlockingQueue<Message> messageQueue = new LinkedBlockingQueue<>();
-
     /**
      * How many times a failed message should be replayed.
      * A value of 0 means never give up on a message and always replay it.
      */
     private final int maxRetries;
 
+    /**
+     * Contains a FIFO queue for failed messages.
+     */
+    private transient LinkedBlockingQueue<Message> messageQueue;
+
     /**
      * Constructor.
      * @param maxRetries Maximum number of times to retry a failed tuple.
@@ -48,7 +48,8 @@ public RetryFailedTuples(final int maxRetries) {
 
     @Override
     public void open(final Map<String, Object> stormConfig, final TopologyContext topologyContext) {
-        // no-op
+        // Create queue
+        messageQueue = new LinkedBlockingQueue<>();
     }
 
     @Override

src/test/java/org/sourcelab/storm/spout/redis/example/ExampleLocalTopology.java

@@ -3,10 +3,12 @@
 import org.apache.storm.Config;
 import org.apache.storm.LocalCluster;
 import org.apache.storm.generated.StormTopology;
+import org.apache.storm.metric.LoggingMetricsConsumer;
 import org.apache.storm.topology.TopologyBuilder;
 import org.sourcelab.storm.spout.redis.RedisStreamSpout;
 import org.sourcelab.storm.spout.redis.RedisStreamSpoutConfig;
-import org.sourcelab.storm.spout.redis.failhandler.RetryFailedTuples;
+import org.sourcelab.storm.spout.redis.failhandler.ExponentialBackoffConfig;
+import org.sourcelab.storm.spout.redis.failhandler.ExponentialBackoffFailureHandler;
 import org.sourcelab.storm.spout.redis.util.test.RedisTestHelper;
 import org.testcontainers.containers.GenericContainer;
 
@@ -70,7 +72,7 @@ public void runExample(final boolean enableDebug) throws Exception {
                 .withConsumerIdPrefix(consumerPrefix)
                 .withStreamKey(streamKey)
                 // Failure Handler
-                .withFailureHandler(new RetryFailedTuples(1))
+                .withFailureHandler(new ExponentialBackoffFailureHandler(ExponentialBackoffConfig.defaultConfig()))
                 // Tuple Handler Class
                 .withTupleConverter(new TestTupleConverter("value"));
 
@@ -83,6 +85,10 @@ public void runExample(final boolean enableDebug) throws Exception {
             // Create configuration
             final Config config = new Config();
             config.setDebug(enableDebug);
+            // Never fall back to java serialization
+            config.setFallBackOnJavaSerialization(false);
+            // Logging metrics consumer for metrics validation.
+            config.registerMetricsConsumer(LoggingMetricsConsumer.class);
 
             // Submits the topology to the local cluster
             localCluster.submitTopology(

src/test/java/org/sourcelab/storm/spout/redis/failhandler/ExponentialBackoffFailureHandlerTest.java

@@ -626,12 +626,12 @@ private void validateTupleIsNotBeingTracked(final ExponentialBackoffFailureHandl
 
     private void verifyMetricInteractions() {
         verify(mockTopologyContext, times(1))
-            .registerCounter(eq("failureHandler.exceededRetryLimit"));
+            .registerCounter(eq("failureHandler_exceededRetryLimit"));
         verify(mockTopologyContext, times(1))
-            .registerCounter(eq("failureHandler.retriedMessages"));
+            .registerCounter(eq("failureHandler_retriedMessages"));
         verify(mockTopologyContext, times(1))
-            .registerCounter(eq("failureHandler.successfulRetriedMessages"));
+            .registerCounter(eq("failureHandler_successfulRetriedMessages"));
         verify(mockTopologyContext, times(1))
-            .registerGauge(eq("failureHandler.queuedForRetry"), any(Gauge.class));
+            .registerGauge(eq("failureHandler_queuedForRetry"), any(Gauge.class));
     }
 }