Update to support new namespace for Storm 1.0.0

pom.xml

@@ -24,11 +24,11 @@
 
 
     <properties>
-        <aws-java-sdk.version>1.7.13</aws-java-sdk.version>
-        <storm.version>0.9.2-incubating</storm.version>
-        <curator-framework.version>1.1.3</curator-framework.version>
-        <guava.version>13.0-final</guava.version>
-        <commons-lang3.version>3.0</commons-lang3.version>
+        <aws-java-sdk.version>1.10.74</aws-java-sdk.version>
+        <storm.version>1.0.0</storm.version>
+        <curator-framework.version>1.3.3</curator-framework.version>
+        <guava.version>19.0</guava.version>
+        <commons-lang3.version>3.4</commons-lang3.version>
     </properties>
 
 

src/main/java/com/amazonaws/services/kinesis/stormspout/DefaultKinesisRecordScheme.java

@@ -18,7 +18,7 @@
 import java.util.ArrayList;
 import java.util.List;
 
-import backtype.storm.tuple.Fields;
+import org.apache.storm.tuple.Fields;
 
 import com.amazonaws.services.kinesis.model.Record;
 

src/main/java/com/amazonaws/services/kinesis/stormspout/IKinesisRecordScheme.java

@@ -17,7 +17,7 @@
 
 import java.util.List;
 
-import backtype.storm.tuple.Fields;
+import org.apache.storm.tuple.Fields;
 
 import com.amazonaws.services.kinesis.model.Record;
 

src/main/java/com/amazonaws/services/kinesis/stormspout/KinesisHelper.java

@@ -20,6 +20,7 @@
 import java.util.Map;
 import java.util.concurrent.Callable;
 
+import org.objenesis.strategy.StdInstantiatorStrategy;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

src/main/java/com/amazonaws/services/kinesis/stormspout/KinesisSpout.java

@@ -24,11 +24,11 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import backtype.storm.Config;
-import backtype.storm.spout.SpoutOutputCollector;
-import backtype.storm.task.TopologyContext;
-import backtype.storm.topology.IRichSpout;
-import backtype.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.Config;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
 
 import com.amazonaws.ClientConfiguration;
 import com.amazonaws.auth.AWSCredentialsProvider;

src/main/java/com/amazonaws/services/kinesis/stormspout/SerializationHelper.java

@@ -59,7 +59,8 @@ public static byte[] kryoSerializeObject(final Object obj) {
         final ByteArrayOutputStream os = new ByteArrayOutputStream();
         final Output output = new Output(os);
 
-        kryo.setInstantiatorStrategy(new StdInstantiatorStrategy());
+        // Better initialization to keep a fallback and prevent crash
+        ((Kryo.DefaultInstantiatorStrategy) kryo.getInstantiatorStrategy()).setFallbackInstantiatorStrategy(new StdInstantiatorStrategy());
         kryo.writeClassAndObject(output, obj);
 
         output.flush();
@@ -70,7 +71,9 @@ public static Object kryoDeserializeObject(final byte[] ser) {
         final Kryo kryo = new Kryo();
         final Input input = new Input(new ByteArrayInputStream(ser));
 
-        kryo.setInstantiatorStrategy(new StdInstantiatorStrategy());
+        // Better initialization to keep a fallback and prevent crash
+        ((Kryo.DefaultInstantiatorStrategy) kryo.getInstantiatorStrategy()).setFallbackInstantiatorStrategy(new StdInstantiatorStrategy());
+
         return kryo.readClassAndObject(input);
     }
 }

src/main/java/com/amazonaws/services/kinesis/stormspout/state/zookeeper/ZookeeperShardState.java

@@ -64,11 +64,11 @@ class ZookeeperShardState {
         try {
             zk = CuratorFrameworkFactory.newClient(config.getZookeeperConnectionString(),
                     new ExponentialBackoffRetry(BASE_SLEEP_TIME_MS, MAX_NUM_RETRIES));
-        } catch (IOException e) {
+            zk.start();
+        } catch (Exception e) {
             LOG.error("Could not connect to ZooKeeper", e);
             throw new KinesisSpoutException(e);
         }
-        zk.start();
     }
 
     /**

src/main/samples/SampleBolt.java

@@ -23,11 +23,11 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import backtype.storm.task.TopologyContext;
-import backtype.storm.topology.BasicOutputCollector;
-import backtype.storm.topology.OutputFieldsDeclarer;
-import backtype.storm.topology.base.BaseBasicBolt;
-import backtype.storm.tuple.Tuple;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.BasicOutputCollector;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseBasicBolt;
+import org.apache.storm.tuple.Tuple;
 
 import com.amazonaws.services.kinesis.model.Record;
 

src/main/samples/SampleKinesisRecordScheme.java

@@ -16,7 +16,7 @@
 import java.util.ArrayList;
 import java.util.List;
 
-import backtype.storm.tuple.Fields;
+import org.apache.storm.tuple.Fields;
 
 import com.amazonaws.services.kinesis.model.Record;
 import com.amazonaws.services.kinesis.stormspout.IKinesisRecordScheme;

src/main/samples/SampleTopology.java

@@ -18,17 +18,18 @@
 import java.io.IOException;
 import java.util.Properties;
 
+import org.apache.storm.generated.AuthorizationException;
 import org.apache.zookeeper.KeeperException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import backtype.storm.Config;
-import backtype.storm.LocalCluster;
-import backtype.storm.StormSubmitter;
-import backtype.storm.generated.AlreadyAliveException;
-import backtype.storm.generated.InvalidTopologyException;
-import backtype.storm.topology.TopologyBuilder;
-import backtype.storm.tuple.Fields;
+import org.apache.storm.Config;
+import org.apache.storm.LocalCluster;
+import org.apache.storm.StormSubmitter;
+import org.apache.storm.generated.AlreadyAliveException;
+import org.apache.storm.generated.InvalidTopologyException;
+import org.apache.storm.topology.TopologyBuilder;
+import org.apache.storm.tuple.Fields;
 
 import com.amazonaws.ClientConfiguration;
 import com.amazonaws.regions.Regions;
@@ -46,7 +47,7 @@ public class SampleTopology {
     private static String zookeeperEndpoint;
     private static String zookeeperPrefix;
 
-    public static void main(String[] args) throws IllegalArgumentException, KeeperException, InterruptedException, AlreadyAliveException, InvalidTopologyException, IOException {
+    public static void main(String[] args) throws IllegalArgumentException, KeeperException, InterruptedException, AlreadyAliveException, InvalidTopologyException, IOException, AuthorizationException {
         String propertiesFile = null;
         String mode = null;