Rework when consumer thread is started

Author: Crim

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

@@ -84,16 +84,8 @@ public void open(
         // Create funnel instance.
         this.funnel = new MemoryFunnel(config, spoutConfig, topologyContext);
 
-        // Create consumer and client
-        final int taskIndex = topologyContext.getThisTaskIndex();
-        final Client client = new ClientFactory().createClient(config, taskIndex);
-        final Consumer consumer = new Consumer(config, client, (ConsumerFunnel) funnel);
-
-        // Create background consuming thread.
-        consumerThread = new Thread(
-            consumer,
-            "RedisStreamSpout-ConsumerThread[" + taskIndex + "]"
-        );
+        // Create and start consumer thread.
+        createAndStartConsumerThread();
     }
 
     @Override
@@ -104,12 +96,15 @@ public void close() {
 
     @Override
     public void activate() {
-        if (consumerThread.isAlive()) {
+        // If the thread is already running and alive
+        if (consumerThread != null && consumerThread.isAlive()) {
             // No-op.  It's already running, and deactivate() is a no-op for us.
             return;
         }
-        // Start thread, this should return immediately, but start a background processing thread.
-        consumerThread.start();
+
+        // If we haven't created the consumer thread yet, or it has previously died.
+        // Create and start it
+        createAndStartConsumerThread();
     }
 
     @Override
@@ -187,4 +182,21 @@ public void declareOutputFields(final OutputFieldsDeclarer declarer) {
     public Map<String, Object> getComponentConfiguration() {
         return new HashMap<>();
     }
+
+    /**
+     * Create background consumer thread.
+     */
+    private void createAndStartConsumerThread() {
+        // Create consumer and client
+        final int taskIndex = topologyContext.getThisTaskIndex();
+        final Client client = new ClientFactory().createClient(config, taskIndex);
+        final Consumer consumer = new Consumer(config, client, (ConsumerFunnel) funnel);
+
+        // Create background consuming thread.
+        consumerThread = new Thread(
+            consumer,
+            "RedisStreamSpout-ConsumerThread[" + taskIndex + "]"
+        );
+        consumerThread.start();
+    }
 }

src/test/java/org/sourcelab/storm/spout/redis/AbstractRedisStreamSpoutIntegrationTest.java

@@ -119,6 +119,8 @@ void smokeTest_openAndClose(final ClientType clientType) {
 
             // Open spout
             spout.open(stormConfig, mockTopologyContext, new SpoutOutputCollector(collector));
+            spout.activate();
+            spout.deactivate();
 
             // Close spout via autocloseable
         }
@@ -158,14 +160,49 @@ void smokeTest_openActivateDeactivateAndClose(final ClientType clientType) throw
         verify(mockTopologyContext, times(1)).getThisTaskIndex();
     }
 
+    /**
+     * Basic lifecycle smoke test.
+     * Cycle calling activate/deactivate a few times.
+     */
+    @ParameterizedTest
+    @EnumSource(ClientType.class)
+    void smokeTest_cycleActivateDeactivate(final ClientType clientType) throws InterruptedException {
+        // Inject client type into config
+        configBuilder.withClientType(clientType);
+
+        // Create spout
+        try (final RedisStreamSpout spout = new RedisStreamSpout(configBuilder.build())) {
+            final StubSpoutCollector collector = new StubSpoutCollector();
+
+            // Open spout
+            spout.open(stormConfig, mockTopologyContext, new SpoutOutputCollector(collector));
+
+            // Cycle Activate and Deactivate a few times
+            spout.activate();
+            Thread.sleep(2000L);
+            spout.deactivate();
+
+            spout.activate();
+            Thread.sleep(2000L);
+            spout.deactivate();
+
+            spout.activate();
+            Thread.sleep(2000L);
+            spout.deactivate();
+
+            // Close via Autocloseable.
+        }
+
+        // Verify mocks
+        verify(mockTopologyContext, times(1)).getThisTaskIndex();
+    }
+
     /**
      * Verifies the behavior when you attempt to connect to a redis instance
      * that does not exist.  Looks like nothing. You get errors in the logs.
      *
      * Disabled for now.
      */
-//    @ParameterizedTest
-//    @EnumSource(ClientType.class)
     void smokeTest_configureInvalidRedisHost(final ClientType clientType) throws InterruptedException {
         // Inject client type into config
         configBuilder.withClientType(clientType);
@@ -199,7 +236,7 @@ void smokeTest_configureInvalidRedisHost(final ClientType clientType) throws Int
         }
 
         // Verify mocks
-        verify(mockTopologyContext, times(1)).getThisTaskIndex();
+        verify(mockTopologyContext, times(2)).getThisTaskIndex();
     }
 
     /**
@@ -516,6 +553,7 @@ void test_EmitDownSeparateStreams(final ClientType clientType) {
 
             // Open spout
             spout.open(stormConfig, mockTopologyContext, new SpoutOutputCollector(collector));
+            spout.activate();
 
             // Ask for stream names
             final OutputFieldsGetter getter = new OutputFieldsGetter();