From 4471d2afd59687ac99bc136a1b5acc5ae179cf53 Mon Sep 17 00:00:00 2001 From: Andrey Pereverzin Date: Tue, 26 Apr 2016 15:03:46 +0100 Subject: [PATCH] Apache Storm version 1.0.0 used --- .../storm/jms/example/ExampleJmsTopology.java | 14 +++++----- .../apache/storm/jms/example/GenericBolt.java | 16 ++++++------ .../storm/jms/example/JsonTupleProducer.java | 6 ++--- examples/src/main/resources/jms-activemq.xml | 4 +-- examples/src/main/resources/log4j.properties | 2 +- pom.xml | 11 ++++++-- .../apache/storm/jms/JmsMessageProducer.java | 8 +++--- .../apache/storm/jms/JmsTupleProducer.java | 4 +-- .../org/apache/storm/jms/bolt/JmsBolt.java | 14 +++++----- .../org/apache/storm/jms/spout/JmsSpout.java | 16 ++++++------ .../apache/storm/jms/trident/JmsState.java | 8 +++--- .../storm/jms/trident/JmsStateFactory.java | 6 ++--- .../apache/storm/jms/trident/JmsUpdater.java | 8 +++--- .../trident/TridentJmsMessageProducer.java | 6 ++--- .../storm/jms/trident/TridentJmsSpout.java | 26 +++++++++---------- .../apache/storm/jms/spout/JmsSpoutTest.java | 2 +- .../jms/spout/MockSpoutOutputCollector.java | 7 ++++- .../storm/jms/spout/MockTupleProducer.java | 6 ++--- 18 files changed, 88 insertions(+), 76 deletions(-) diff --git a/examples/src/main/java/org/apache/storm/jms/example/ExampleJmsTopology.java b/examples/src/main/java/org/apache/storm/jms/example/ExampleJmsTopology.java index 7db35a8..1216f1a 100644 --- a/examples/src/main/java/org/apache/storm/jms/example/ExampleJmsTopology.java +++ b/examples/src/main/java/org/apache/storm/jms/example/ExampleJmsTopology.java @@ -22,18 +22,18 @@ import javax.jms.Session; import javax.jms.TextMessage; -import backtype.storm.Config; -import backtype.storm.LocalCluster; -import backtype.storm.StormSubmitter; +import org.apache.storm.Config; +import org.apache.storm.LocalCluster; +import org.apache.storm.StormSubmitter; import org.apache.storm.jms.JmsMessageProducer; import org.apache.storm.jms.JmsProvider; import org.apache.storm.jms.JmsTupleProducer; import org.apache.storm.jms.bolt.JmsBolt; import org.apache.storm.jms.spout.JmsSpout; -import backtype.storm.topology.TopologyBuilder; -import backtype.storm.tuple.Fields; -import backtype.storm.tuple.Tuple; -import backtype.storm.utils.Utils; +import org.apache.storm.topology.TopologyBuilder; +import org.apache.storm.tuple.Fields; +import org.apache.storm.tuple.Tuple; +import org.apache.storm.utils.Utils; public class ExampleJmsTopology { public static final String JMS_QUEUE_SPOUT = "JMS_QUEUE_SPOUT"; diff --git a/examples/src/main/java/org/apache/storm/jms/example/GenericBolt.java b/examples/src/main/java/org/apache/storm/jms/example/GenericBolt.java index f2f7799..90c8b2d 100644 --- a/examples/src/main/java/org/apache/storm/jms/example/GenericBolt.java +++ b/examples/src/main/java/org/apache/storm/jms/example/GenericBolt.java @@ -19,21 +19,21 @@ import java.util.Map; -import backtype.storm.topology.base.BaseRichBolt; +import org.apache.storm.topology.base.BaseRichBolt; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import backtype.storm.task.OutputCollector; -import backtype.storm.task.TopologyContext; -import backtype.storm.topology.OutputFieldsDeclarer; -import backtype.storm.tuple.Fields; -import backtype.storm.tuple.Tuple; +import org.apache.storm.task.OutputCollector; +import org.apache.storm.task.TopologyContext; +import org.apache.storm.topology.OutputFieldsDeclarer; +import org.apache.storm.tuple.Fields; +import org.apache.storm.tuple.Tuple; /** - * A generic backtype.storm.topology.IRichBolt implementation + * A generic org.apache.storm.topology.IRichBolt implementation * for testing/debugging the Storm JMS Spout and example topologies. *

* For debugging purposes, set the log level of the - * backtype.storm.contrib.jms package to DEBUG for debugging + * org.apache.storm.contrib.jms package to DEBUG for debugging * output. * @author tgoetz * diff --git a/examples/src/main/java/org/apache/storm/jms/example/JsonTupleProducer.java b/examples/src/main/java/org/apache/storm/jms/example/JsonTupleProducer.java index cb11f3d..9ee175e 100644 --- a/examples/src/main/java/org/apache/storm/jms/example/JsonTupleProducer.java +++ b/examples/src/main/java/org/apache/storm/jms/example/JsonTupleProducer.java @@ -22,9 +22,9 @@ import javax.jms.TextMessage; import org.apache.storm.jms.JmsTupleProducer; -import backtype.storm.topology.OutputFieldsDeclarer; -import backtype.storm.tuple.Fields; -import backtype.storm.tuple.Values; +import org.apache.storm.topology.OutputFieldsDeclarer; +import org.apache.storm.tuple.Fields; +import org.apache.storm.tuple.Values; /** * A simple JmsTupleProducer that expects to receive diff --git a/examples/src/main/resources/jms-activemq.xml b/examples/src/main/resources/jms-activemq.xml index 1a845b8..1e5ab53 100644 --- a/examples/src/main/resources/jms-activemq.xml +++ b/examples/src/main/resources/jms-activemq.xml @@ -31,9 +31,9 @@ --> - + - + diff --git a/examples/src/main/resources/log4j.properties b/examples/src/main/resources/log4j.properties index 079b195..6ddc227 100644 --- a/examples/src/main/resources/log4j.properties +++ b/examples/src/main/resources/log4j.properties @@ -22,7 +22,7 @@ log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%5p (%C:%L) - %m%n -log4j.logger.backtype.storm.contrib=DEBUG +log4j.logger.org.apache.storm.contrib=DEBUG log4j.logger.clojure.contrib=WARN log4j.logger.org.springframework=WARN log4j.logger.org.apache.zookeeper=WARN diff --git a/pom.xml b/pom.xml index aa87b75..30e7658 100644 --- a/pom.xml +++ b/pom.xml @@ -27,7 +27,7 @@ 4.0.0 com.github.ptgoetz storm-jms - 0.9.2-SNAPSHOT + 1.0.0 Storm JMS Storm JMS Components @@ -61,8 +61,9 @@ - 0.9.3 + 1.0.0 + org.apache.storm @@ -82,6 +83,12 @@ 4.10 test + + org.mortbay.jetty + jetty + 6.1.26 + test + diff --git a/src/main/java/org/apache/storm/jms/JmsMessageProducer.java b/src/main/java/org/apache/storm/jms/JmsMessageProducer.java index e9ad52d..4fd31ab 100644 --- a/src/main/java/org/apache/storm/jms/JmsMessageProducer.java +++ b/src/main/java/org/apache/storm/jms/JmsMessageProducer.java @@ -6,11 +6,11 @@ import javax.jms.Message; import javax.jms.Session; -import backtype.storm.tuple.Tuple; -import backtype.storm.tuple.Values; +import org.apache.storm.tuple.Tuple; +import org.apache.storm.tuple.Values; /** * JmsMessageProducer implementations are responsible for translating - * a backtype.storm.tuple.Values instance into a + * a org.apache.storm.tuple.Values instance into a * javax.jms.Message object. *

* @@ -21,7 +21,7 @@ public interface JmsMessageProducer extends Serializable{ /** - * Translate a backtype.storm.tuple.Tuple object + * Translate a org.apache.storm.tuple.Tuple object * to a javax.jms.MessageValues objects diff --git a/src/main/java/org/apache/storm/jms/bolt/JmsBolt.java b/src/main/java/org/apache/storm/jms/bolt/JmsBolt.java index 896f932..50a4e01 100644 --- a/src/main/java/org/apache/storm/jms/bolt/JmsBolt.java +++ b/src/main/java/org/apache/storm/jms/bolt/JmsBolt.java @@ -10,19 +10,19 @@ import javax.jms.MessageProducer; import javax.jms.Session; -import backtype.storm.topology.base.BaseRichBolt; +import org.apache.storm.topology.base.BaseRichBolt; import org.apache.storm.jms.JmsMessageProducer; import org.apache.storm.jms.JmsProvider; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import backtype.storm.task.OutputCollector; -import backtype.storm.task.TopologyContext; -import backtype.storm.topology.OutputFieldsDeclarer; -import backtype.storm.tuple.Tuple; +import org.apache.storm.task.OutputCollector; +import org.apache.storm.task.TopologyContext; +import org.apache.storm.topology.OutputFieldsDeclarer; +import org.apache.storm.tuple.Tuple; /** - * A JmsBolt receives backtype.storm.tuple.Tuple objects from a Storm + * A JmsBolt receives org.apache.storm.tuple.Tuple objects from a Storm * topology and publishes JMS Messages to a destination (topic or queue). *

* To use a JmsBolt in a topology, the following must be supplied: @@ -34,7 +34,7 @@ * and javax.jms.Destination objects requied to publish JMS messages. *

* The JmsBolt uses a JmsMessageProducer to translate - * backtype.storm.tuple.Tuple objects into + * org.apache.storm.tuple.Tuple objects into * javax.jms.Message objects for publishing. *

* Both JmsProvider and JmsMessageProducer must be set, or the bolt will diff --git a/src/main/java/org/apache/storm/jms/spout/JmsSpout.java b/src/main/java/org/apache/storm/jms/spout/JmsSpout.java index 03ab77b..38a8a7e 100644 --- a/src/main/java/org/apache/storm/jms/spout/JmsSpout.java +++ b/src/main/java/org/apache/storm/jms/spout/JmsSpout.java @@ -13,17 +13,17 @@ import javax.jms.MessageListener; import javax.jms.Session; -import backtype.storm.topology.base.BaseRichSpout; +import org.apache.storm.topology.base.BaseRichSpout; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.storm.jms.JmsProvider; import org.apache.storm.jms.JmsTupleProducer; -import backtype.storm.spout.SpoutOutputCollector; -import backtype.storm.task.TopologyContext; -import backtype.storm.topology.OutputFieldsDeclarer; -import backtype.storm.tuple.Values; -import backtype.storm.utils.Utils; +import org.apache.storm.spout.SpoutOutputCollector; +import org.apache.storm.task.TopologyContext; +import org.apache.storm.topology.OutputFieldsDeclarer; +import org.apache.storm.tuple.Values; +import org.apache.storm.utils.Utils; /** * A Storm Spout implementation that listens to a JMS topic or queue @@ -117,7 +117,7 @@ public void setJmsProvider(JmsProvider provider){ /** * Set the JmsTupleProducer * implementation that will convert javax.jms.Message - * object to backtype.storm.tuple.Values objects + * object to org.apache.storm.tuple.Values objects * to be emitted. * * @param producer @@ -156,7 +156,7 @@ public void open(Map conf, TopologyContext context, if(this.tupleProducer == null){ throw new IllegalStateException("JMS Tuple Producer has not been set."); } - Integer topologyTimeout = (Integer)conf.get("topology.message.timeout.secs"); + Long topologyTimeout = (Long)conf.get("topology.message.timeout.secs"); // TODO fine a way to get the default timeout from storm, so we're not hard-coding to 30 seconds (it could change) topologyTimeout = topologyTimeout == null ? 30 : topologyTimeout; if( (topologyTimeout.intValue() * 1000 )> this.recoveryPeriod){ diff --git a/src/main/java/org/apache/storm/jms/trident/JmsState.java b/src/main/java/org/apache/storm/jms/trident/JmsState.java index 0694098..3159cfa 100644 --- a/src/main/java/org/apache/storm/jms/trident/JmsState.java +++ b/src/main/java/org/apache/storm/jms/trident/JmsState.java @@ -18,12 +18,12 @@ package org.apache.storm.jms.trident; import org.apache.storm.jms.JmsProvider; -import backtype.storm.topology.FailedException; +import org.apache.storm.topology.FailedException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import storm.trident.operation.TridentCollector; -import storm.trident.state.State; -import storm.trident.tuple.TridentTuple; +import org.apache.storm.trident.operation.TridentCollector; +import org.apache.storm.trident.state.State; +import org.apache.storm.trident.tuple.TridentTuple; import javax.jms.*; import java.io.Serializable; diff --git a/src/main/java/org/apache/storm/jms/trident/JmsStateFactory.java b/src/main/java/org/apache/storm/jms/trident/JmsStateFactory.java index 7fe4d85..9a02ba9 100644 --- a/src/main/java/org/apache/storm/jms/trident/JmsStateFactory.java +++ b/src/main/java/org/apache/storm/jms/trident/JmsStateFactory.java @@ -17,9 +17,9 @@ */ package org.apache.storm.jms.trident; -import backtype.storm.task.IMetricsContext; -import storm.trident.state.State; -import storm.trident.state.StateFactory; +import org.apache.storm.task.IMetricsContext; +import org.apache.storm.trident.state.State; +import org.apache.storm.trident.state.StateFactory; import java.util.Map; diff --git a/src/main/java/org/apache/storm/jms/trident/JmsUpdater.java b/src/main/java/org/apache/storm/jms/trident/JmsUpdater.java index 3269521..a2709a4 100644 --- a/src/main/java/org/apache/storm/jms/trident/JmsUpdater.java +++ b/src/main/java/org/apache/storm/jms/trident/JmsUpdater.java @@ -17,10 +17,10 @@ */ package org.apache.storm.jms.trident; -import backtype.storm.topology.FailedException; -import storm.trident.operation.TridentCollector; -import storm.trident.state.BaseStateUpdater; -import storm.trident.tuple.TridentTuple; +import org.apache.storm.topology.FailedException; +import org.apache.storm.trident.operation.TridentCollector; +import org.apache.storm.trident.state.BaseStateUpdater; +import org.apache.storm.trident.tuple.TridentTuple; import javax.jms.JMSException; import java.util.List; diff --git a/src/main/java/org/apache/storm/jms/trident/TridentJmsMessageProducer.java b/src/main/java/org/apache/storm/jms/trident/TridentJmsMessageProducer.java index 8312947..8f9c301 100644 --- a/src/main/java/org/apache/storm/jms/trident/TridentJmsMessageProducer.java +++ b/src/main/java/org/apache/storm/jms/trident/TridentJmsMessageProducer.java @@ -1,7 +1,7 @@ package org.apache.storm.jms.trident; -import backtype.storm.tuple.Tuple; -import storm.trident.tuple.TridentTuple; +import org.apache.storm.tuple.Tuple; +import org.apache.storm.trident.tuple.TridentTuple; import javax.jms.JMSException; import javax.jms.Message; @@ -11,7 +11,7 @@ public interface TridentJmsMessageProducer extends Serializable{ /** - * Translate a backtype.storm.tuple.TridentTuple object + * Translate a org.apache.storm.tuple.TridentTuple object * to a javax.jms.MessageJmsTupleProducer * implementation that will convert javax.jms.Message - * object to backtype.storm.tuple.Values objects + * object to org.apache.storm.tuple.Values objects * to be emitted. * * @param tupleProducer @@ -146,7 +146,7 @@ private static final String toDeliveryModeString(int acknowledgeMode) { } @Override - public storm.trident.spout.ITridentSpout.BatchCoordinator getCoordinator( + public org.apache.storm.trident.spout.ITridentSpout.BatchCoordinator getCoordinator( String txStateId, @SuppressWarnings("rawtypes") Map conf, TopologyContext context) { return new JmsBatchCoordinator(name); } diff --git a/src/test/java/org/apache/storm/jms/spout/JmsSpoutTest.java b/src/test/java/org/apache/storm/jms/spout/JmsSpoutTest.java index 42ce573..8438952 100644 --- a/src/test/java/org/apache/storm/jms/spout/JmsSpoutTest.java +++ b/src/test/java/org/apache/storm/jms/spout/JmsSpoutTest.java @@ -35,7 +35,7 @@ import org.mortbay.log.Log; import org.apache.storm.jms.JmsProvider; -import backtype.storm.spout.SpoutOutputCollector; +import org.apache.storm.spout.SpoutOutputCollector; public class JmsSpoutTest { @Test diff --git a/src/test/java/org/apache/storm/jms/spout/MockSpoutOutputCollector.java b/src/test/java/org/apache/storm/jms/spout/MockSpoutOutputCollector.java index 8a535ce..5c9e1a4 100644 --- a/src/test/java/org/apache/storm/jms/spout/MockSpoutOutputCollector.java +++ b/src/test/java/org/apache/storm/jms/spout/MockSpoutOutputCollector.java @@ -20,7 +20,7 @@ import java.util.ArrayList; import java.util.List; -import backtype.storm.spout.ISpoutOutputCollector; +import org.apache.storm.spout.ISpoutOutputCollector; public class MockSpoutOutputCollector implements ISpoutOutputCollector { boolean emitted = false; @@ -40,6 +40,11 @@ public void emitDirect(int taskId, String streamId, List tuple, Object m public void reportError(Throwable error) { } + @Override + public long getPendingCount() { + return 0L; + } + public boolean emitted(){ return this.emitted; } diff --git a/src/test/java/org/apache/storm/jms/spout/MockTupleProducer.java b/src/test/java/org/apache/storm/jms/spout/MockTupleProducer.java index f5eda10..ea571fc 100644 --- a/src/test/java/org/apache/storm/jms/spout/MockTupleProducer.java +++ b/src/test/java/org/apache/storm/jms/spout/MockTupleProducer.java @@ -22,9 +22,9 @@ import javax.jms.TextMessage; import org.apache.storm.jms.JmsTupleProducer; -import backtype.storm.topology.OutputFieldsDeclarer; -import backtype.storm.tuple.Fields; -import backtype.storm.tuple.Values; +import org.apache.storm.topology.OutputFieldsDeclarer; +import org.apache.storm.tuple.Fields; +import org.apache.storm.tuple.Values; public class MockTupleProducer implements JmsTupleProducer { private static final long serialVersionUID = 1L;