Breaking up Integration tests

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - [PR-37](https://github.com/salesforce/storm-dynamic-spout/pull/37) Replaced `SidelineRequestIdentifier` with `FilterChainStepIdentifier` in the FilterChain.
 - [PR-37](https://github.com/salesforce/storm-dynamic-spout/pull/37) Added isSidelineStarted() and isSidelineStopped() to the `SidelineController`
 - [PR-52](https://github.com/salesforce/storm-dynamic-spout/pull/52) Remove Kafka-Test-Server.  Replace with Kafka-JUnit external dependency.
+- [PR-66](https://github.com/salesforce/storm-dynamic-spout/pull/66) DynamicSpout.open() now throws IllegalStateException if you attempt to open it more than once.
 
 ### Improvements
 - [PR-38](https://github.com/salesforce/storm-dynamic-spout/pull/38) Removed unused method `Deserializer.getOutputFields()`.

src/main/java/com/salesforce/storm/spout/dynamic/DynamicSpout.java

@@ -140,12 +140,12 @@ public DynamicSpout(Map<String, Object> spoutConfig) {
      * @param topologyConfig The Storm Topology configuration.
      * @param topologyContext The Storm Topology context.
      * @param spoutOutputCollector The output collector to emit tuples via.
+     * @throws IllegalStateException if you attempt to open the spout multiple times.
      */
     @Override
     public void open(Map topologyConfig, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
         if (isOpen) {
-            logger.warn("This spout has already been opened, cowardly refusing to open it again!");
-            return;
+            throw new IllegalStateException("This spout has already been opened.");
         }
 
         // Save references.
@@ -317,9 +317,9 @@ public void close() {
             metricsRecorder = null;
         }
 
-        if (spoutHandler != null) {
-            spoutHandler.onSpoutClose(this);
-            spoutHandler.close();
+        if (getSpoutHandler() != null) {
+            getSpoutHandler().onSpoutClose(this);
+            getSpoutHandler().close();
             spoutHandler = null;
         }
 
@@ -332,8 +332,8 @@ public void close() {
     @Override
     public void activate() {
         logger.debug("Activating spout");
-        if (spoutHandler != null) {
-            spoutHandler.onSpoutActivate(this);
+        if (getSpoutHandler() != null) {
+            getSpoutHandler().onSpoutActivate(this);
         }
     }
 
@@ -343,8 +343,8 @@ public void activate() {
     @Override
     public void deactivate() {
         logger.debug("Deactivate spout");
-        if (spoutHandler != null) {
-            spoutHandler.onSpoutDeactivate(this);
+        if (getSpoutHandler() != null) {
+            getSpoutHandler().onSpoutDeactivate(this);
         }
     }
 
@@ -405,6 +405,9 @@ private TopologyContext getTopologyContext() {
     /**
      * Add a new VirtualSpout to the coordinator, this will get picked up by the coordinator's monitor, opened and
      * managed with teh other currently running spouts.
+     *
+     * This method is blocking.
+     *
      * @param spout New delegate spout
      * @throws SpoutAlreadyExistsException if a spout already exists with the same VirtualSpoutIdentifier.
      */
@@ -489,6 +492,14 @@ SpoutMessageBus getMessageBus() {
         return messageBus;
     }
 
+    /**
+     * @return The SpoutHandler implementation.
+     */
+    SpoutHandler getSpoutHandler() {
+        checkSpoutOpened();
+        return spoutHandler;
+    }
+
     /**
      * @return The stream that tuples will be emitted out.
      */