You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by omerhadari <gi...@git.apache.org> on 2017/07/21 15:09:42 UTC
[GitHub] storm pull request #2234: Bugfix/fix configuration cast exception
GitHub user omerhadari opened a pull request:
https://github.com/apache/storm/pull/2234
Bugfix/fix configuration cast exception
I open this PR since I was delighted to see that there is an existing JMS spout implementation, but encountered a bug when trying to use it. The bug caused an exception (described below) to be thrown when trying to set up (`open`) the spout, since the value of `Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS` corresponds to a `Long`, and not to an `Integer`.
This is a minor change, hope it's ok :)
```
7317 [Thread-18-a-executor[2 2]] ERROR o.a.s.util - Async loop died!
java.lang.ClassCastException: java.lang.Long cannot be cast to java.lang.Integer
at org.apache.storm.jms.spout.JmsSpout.open(JmsSpout.java:175) ~[storm-jms-1.1.0.jar:1.1.0]
at org.apache.storm.daemon.executor$fn__4976$fn__4991.invoke(executor.clj:600) ~[storm-core-1.1.0.jar:1.1.0]
at org.apache.storm.util$async_loop$fn__557.invoke(util.clj:482) [storm-core-1.1.0.jar:1.1.0]
at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_131]
7320 [Thread-18-a-executor[2 2]] ERROR o.a.s.d.executor -
java.lang.ClassCastException: java.lang.Long cannot be cast to java.lang.Integer
at org.apache.storm.jms.spout.JmsSpout.open(JmsSpout.java:175) ~[storm-jms-1.1.0.jar:1.1.0]
at org.apache.storm.daemon.executor$fn__4976$fn__4991.invoke(executor.clj:600) ~[storm-core-1.1.0.jar:1.1.0]
at org.apache.storm.util$async_loop$fn__557.invoke(util.clj:482) [storm-core-1.1.0.jar:1.1.0]
at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_131]
```
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/omerhadari/storm bugfix/fix-configuration-cast-exception
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/2234.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2234
----
commit 34d6fdcb5b85c06cb6d365f60e8f6a77006eec75
Author: Omer Hadari <ha...@gmail.com>
Date: 2017-07-21T07:21:23Z
Cast message timeout conf to `Number`
I changed the casting to `Number` instead of `Integer`
since it is actually a `Long`, and it caused a `ClassCastException`
to be thrown. Now it is referred to as a `Number` so that it won't
happen again.
commit fac2080cb3042e0bc41a0447f47298d3e4e792a1
Author: Omer Hadari <ha...@gmail.com>
Date: 2017-07-21T07:31:54Z
Test behaviour for different Number types
commit afb8adc157f4a7f2f7338cfffe6f4d175df5db5d
Author: Omer Hadari <ha...@gmail.com>
Date: 2017-07-21T15:06:24Z
Remove accidental license edit
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2234#discussion_r128852007
--- Diff: external/storm-jms/src/main/java/org/apache/storm/jms/spout/JmsSpout.java ---
@@ -352,27 +478,46 @@ private static final String toDeliveryModeString(int deliveryMode) {
}
}
+ /**
+ * @return The currently active session.
+ */
protected Session getSession() {
return this.session;
}
+ /**
+ * Check if there is a pending messages state.
--- End diff --
Nit: Rephrase as Check if the subscription requires messages to be acked.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2234#discussion_r128850445
--- Diff: external/storm-jms/src/main/java/org/apache/storm/jms/spout/JmsSpout.java ---
@@ -30,65 +35,110 @@
import javax.jms.MessageListener;
import javax.jms.Session;
-import org.apache.storm.topology.base.BaseRichSpout;
-import org.apache.storm.utils.Utils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
+import org.apache.storm.Config;
import org.apache.storm.jms.JmsProvider;
import org.apache.storm.jms.JmsTupleProducer;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
/**
- * A Storm <code>Spout</code> implementation that listens to a JMS topic or queue
- * and outputs tuples based on the messages it receives.
- * <p>
- * <code>JmsSpout</code> instances rely on <code>JmsProducer</code> implementations
- * to obtain the JMS <code>ConnectionFactory</code> and <code>Destination</code> objects
- * necessary to connect to a JMS topic/queue.
- * <p>
- * When a <code>JmsSpout</code> receives a JMS message, it delegates to an
- * internal <code>JmsTupleProducer</code> instance to create a Storm tuple from the
- * incoming message.
- * <p>
- * Typically, developers will supply a custom <code>JmsTupleProducer</code> implementation
- * appropriate for the expected message content.
+ * A Storm <code>Spout</code> implementation that listens to a JMS topic or
+ * queue and outputs tuples based on the messages it receives.
+ *
+ * <p><code>JmsSpout</code> instances rely on <code>JmsProducer</code>
+ * implementations to obtain the JMS
+ * <code>ConnectionFactory</code> and <code>Destination</code> objects necessary
+ * to connect to a JMS topic/queue.
+ *
+ * <p>When a <code>JmsSpout</code> receives a JMS message, it delegates to an
+ * internal <code>JmsTupleProducer</code> instance to create a Storm tuple from
+ * the incoming message.
+ *
+ * <p>Typically, developers will supply a custom <code>JmsTupleProducer</code>
+ * implementation appropriate for the expected message content.
*/
@SuppressWarnings("serial")
public class JmsSpout extends BaseRichSpout implements MessageListener {
+
+ /** The logger object instance for this class. */
private static final Logger LOG = LoggerFactory.getLogger(JmsSpout.class);
- // JMS options
+ /** The logger of the recovery task. */
+ private static final Logger RECOVERY_TASK_LOG =
+ LoggerFactory.getLogger(RecoveryTask.class);
+
+ /** Time to sleep between queue polling attempts. */
+ private static final int POLL_INTERVAL = 50;
+
+ /**
+ * The default value the {@link Config#TOPOLOGY_MESSAGE_TIMEOUT_SECS}.
+ */
+ private static final int DEFAULT_MESSAGE_TIMEOUT = 30;
+
+ /** Time to wait before queuing the first recovery task (in seconds). */
+ private static final int RECOVERY_DELAY = 10;
--- End diff --
Nit: RECOVERY_DELAY_SECS
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2234: STORM-2652: fix error in open method of JmsSpout
Posted by omerhadari <gi...@git.apache.org>.
Github user omerhadari commented on a diff in the pull request:
https://github.com/apache/storm/pull/2234#discussion_r128855869
--- Diff: external/storm-jms/src/main/java/org/apache/storm/jms/spout/JmsSpout.java ---
@@ -72,19 +72,18 @@
private static final Logger LOG = LoggerFactory.getLogger(JmsSpout.class);
/** The logger of the recovery task. */
- private static final Logger RECOVERY_TASK_LOG =
- LoggerFactory.getLogger(RecoveryTask.class);
+ private static final Logger RECOVERY_TASK_LOG = LoggerFactory.getLogger(RecoveryTask.class);
/** Time to sleep between queue polling attempts. */
- private static final int POLL_INTERVAL = 50;
+ private static final int POLL_INTERVAL_MS = 50;
/**
- * The default value the {@link Config#TOPOLOGY_MESSAGE_TIMEOUT_SECS}.
+ * The default value for {@link Config#TOPOLOGY_MESSAGE_TIMEOUT_SECS}.
*/
- private static final int DEFAULT_MESSAGE_TIMEOUT = 30;
+ private static final int DEFAULT_MESSAGE_TIMEOUT_SECS = 30;
- /** Time to wait before queuing the first recovery task (in seconds). */
- private static final int RECOVERY_DELAY = 10;
+ /** Time to wait before queuing the first recovery task. */
+ private static final int RECOVERY_DELAY_SECS = 10;
--- End diff --
Right, nice catch
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2234
Thanks @vesense
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by omerhadari <gi...@git.apache.org>.
Github user omerhadari commented on the issue:
https://github.com/apache/storm/pull/2234
Done
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2234
Sorry, I didn't realize you had to be given rights in order to be assigned an issue. @ptgoetz can you help out?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by omerhadari <gi...@git.apache.org>.
Github user omerhadari commented on the issue:
https://github.com/apache/storm/pull/2234
Well I have fixed all violations for the files I touched, but it still over many style violations in other files. I'd be more than happy to fix them in a separate PR, but for the scope of this one I pushed and changed my fixes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2234#discussion_r128849355
--- Diff: external/storm-jms/src/main/java/org/apache/storm/jms/spout/JmsSpout.java ---
@@ -205,6 +277,12 @@ public void open(Map<String, Object> conf, TopologyContext context,
}
+ /**
+ * Close the {@link #session} and {@link #connection}.
+ *
+ * <p>When overridden, should always call {@code super} for handling the
--- End diff --
Nit: call `{@code super}` to finalize the active connections
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2234#discussion_r128854811
--- Diff: external/storm-jms/src/main/java/org/apache/storm/jms/spout/JmsSpout.java ---
@@ -30,65 +35,110 @@
import javax.jms.MessageListener;
import javax.jms.Session;
-import org.apache.storm.topology.base.BaseRichSpout;
-import org.apache.storm.utils.Utils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
+import org.apache.storm.Config;
import org.apache.storm.jms.JmsProvider;
import org.apache.storm.jms.JmsTupleProducer;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
/**
- * A Storm <code>Spout</code> implementation that listens to a JMS topic or queue
- * and outputs tuples based on the messages it receives.
- * <p>
- * <code>JmsSpout</code> instances rely on <code>JmsProducer</code> implementations
- * to obtain the JMS <code>ConnectionFactory</code> and <code>Destination</code> objects
- * necessary to connect to a JMS topic/queue.
- * <p>
- * When a <code>JmsSpout</code> receives a JMS message, it delegates to an
- * internal <code>JmsTupleProducer</code> instance to create a Storm tuple from the
- * incoming message.
- * <p>
- * Typically, developers will supply a custom <code>JmsTupleProducer</code> implementation
- * appropriate for the expected message content.
+ * A Storm <code>Spout</code> implementation that listens to a JMS topic or
+ * queue and outputs tuples based on the messages it receives.
+ *
+ * <p><code>JmsSpout</code> instances rely on <code>JmsProducer</code>
+ * implementations to obtain the JMS
+ * <code>ConnectionFactory</code> and <code>Destination</code> objects necessary
+ * to connect to a JMS topic/queue.
+ *
+ * <p>When a <code>JmsSpout</code> receives a JMS message, it delegates to an
+ * internal <code>JmsTupleProducer</code> instance to create a Storm tuple from
+ * the incoming message.
+ *
+ * <p>Typically, developers will supply a custom <code>JmsTupleProducer</code>
+ * implementation appropriate for the expected message content.
*/
@SuppressWarnings("serial")
public class JmsSpout extends BaseRichSpout implements MessageListener {
+
+ /** The logger object instance for this class. */
private static final Logger LOG = LoggerFactory.getLogger(JmsSpout.class);
- // JMS options
+ /** The logger of the recovery task. */
+ private static final Logger RECOVERY_TASK_LOG =
+ LoggerFactory.getLogger(RecoveryTask.class);
+
+ /** Time to sleep between queue polling attempts. */
+ private static final int POLL_INTERVAL = 50;
--- End diff --
Yes, that sounds good. I just wanted to remark on it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2234
@omerhadari There are a few conflicts and a use of a Java 8 API when cherry picking this on to 1.x-branch. Would you mind opening another PR for this for 1.x?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: Bugfix/fix configuration cast exception
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2234
+1. The integration test failure looks unrelated, looks like it couldn't install Zookeeper for some reason.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2234#discussion_r128847982
--- Diff: external/storm-jms/src/main/java/org/apache/storm/jms/spout/JmsSpout.java ---
@@ -30,65 +35,110 @@
import javax.jms.MessageListener;
import javax.jms.Session;
-import org.apache.storm.topology.base.BaseRichSpout;
-import org.apache.storm.utils.Utils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
+import org.apache.storm.Config;
import org.apache.storm.jms.JmsProvider;
import org.apache.storm.jms.JmsTupleProducer;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
/**
- * A Storm <code>Spout</code> implementation that listens to a JMS topic or queue
- * and outputs tuples based on the messages it receives.
- * <p>
- * <code>JmsSpout</code> instances rely on <code>JmsProducer</code> implementations
- * to obtain the JMS <code>ConnectionFactory</code> and <code>Destination</code> objects
- * necessary to connect to a JMS topic/queue.
- * <p>
- * When a <code>JmsSpout</code> receives a JMS message, it delegates to an
- * internal <code>JmsTupleProducer</code> instance to create a Storm tuple from the
- * incoming message.
- * <p>
- * Typically, developers will supply a custom <code>JmsTupleProducer</code> implementation
- * appropriate for the expected message content.
+ * A Storm <code>Spout</code> implementation that listens to a JMS topic or
+ * queue and outputs tuples based on the messages it receives.
+ *
+ * <p><code>JmsSpout</code> instances rely on <code>JmsProducer</code>
+ * implementations to obtain the JMS
+ * <code>ConnectionFactory</code> and <code>Destination</code> objects necessary
+ * to connect to a JMS topic/queue.
+ *
+ * <p>When a <code>JmsSpout</code> receives a JMS message, it delegates to an
+ * internal <code>JmsTupleProducer</code> instance to create a Storm tuple from
+ * the incoming message.
+ *
+ * <p>Typically, developers will supply a custom <code>JmsTupleProducer</code>
+ * implementation appropriate for the expected message content.
*/
@SuppressWarnings("serial")
public class JmsSpout extends BaseRichSpout implements MessageListener {
+
+ /** The logger object instance for this class. */
private static final Logger LOG = LoggerFactory.getLogger(JmsSpout.class);
- // JMS options
+ /** The logger of the recovery task. */
+ private static final Logger RECOVERY_TASK_LOG =
+ LoggerFactory.getLogger(RecoveryTask.class);
+
+ /** Time to sleep between queue polling attempts. */
+ private static final int POLL_INTERVAL = 50;
+
+ /**
+ * The default value the {@link Config#TOPOLOGY_MESSAGE_TIMEOUT_SECS}.
+ */
+ private static final int DEFAULT_MESSAGE_TIMEOUT = 30;
+
+ /** Time to wait before queuing the first recovery task (in seconds). */
+ private static final int RECOVERY_DELAY = 10;
+
+ /**
+ * The acknowledgment mode used for this instance.
+ *
+ * @see Session
+ */
private int jmsAcknowledgeMode = Session.AUTO_ACKNOWLEDGE;
+ /** Indicates whether or not this spout should run as a singleton. */
private boolean distributed = true;
+ /** Used to generate tuples from incoming messages. */
private JmsTupleProducer tupleProducer;
+ /** Encapsulates jms related classes needed to communicate with the mq. */
private JmsProvider jmsProvider;
+ /** Stores incoming messages for later sending. */
private LinkedBlockingQueue<Message> queue;
+
+ /** Contains all message ids of messages that were not yet acked. */
private TreeSet<JmsMessageID> toCommit;
+
+ /** Maps between message ids of not-yet acked messages, and the messages. */
private HashMap<JmsMessageID, Message> pendingMessages;
+
+ /** Counter of handled messages. */
private long messageSequence = 0;
+ /** The collector used to emit tuples. */
private SpoutOutputCollector collector;
+ /** Connection to the jms queue. */
private transient Connection connection;
+
+ /** The active jms session. */
private transient Session session;
+ /** Indicates whether or not a message failed to be processed. */
private boolean hasFailures = false;
- public final Serializable recoveryMutex = "RECOVERY_MUTEX";
+
+ /** Used to safely recover failed JMS sessions across instances. */
+ private final Serializable recoveryMutex = "RECOVERY_MUTEX";
+
+ /** Schedules recovery tasks periodically. */
private Timer recoveryTimer = null;
+
+ /** To to wait between recovery attempts. */
--- End diff --
Nit: Time to wait. Also could you rename recoveryPeriod to recoveryPeriodMs?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by omerhadari <gi...@git.apache.org>.
Github user omerhadari commented on the issue:
https://github.com/apache/storm/pull/2234
This still fails due to stylecheck because of numerous error regarding lines I have not touched.
I do not mind fixing them, but I don't know if it is within the scope of this PR. What do you think?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2234#discussion_r128849546
--- Diff: external/storm-jms/src/main/java/org/apache/storm/jms/spout/JmsSpout.java ---
@@ -30,65 +35,110 @@
import javax.jms.MessageListener;
import javax.jms.Session;
-import org.apache.storm.topology.base.BaseRichSpout;
-import org.apache.storm.utils.Utils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
+import org.apache.storm.Config;
import org.apache.storm.jms.JmsProvider;
import org.apache.storm.jms.JmsTupleProducer;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
/**
- * A Storm <code>Spout</code> implementation that listens to a JMS topic or queue
- * and outputs tuples based on the messages it receives.
- * <p>
- * <code>JmsSpout</code> instances rely on <code>JmsProducer</code> implementations
- * to obtain the JMS <code>ConnectionFactory</code> and <code>Destination</code> objects
- * necessary to connect to a JMS topic/queue.
- * <p>
- * When a <code>JmsSpout</code> receives a JMS message, it delegates to an
- * internal <code>JmsTupleProducer</code> instance to create a Storm tuple from the
- * incoming message.
- * <p>
- * Typically, developers will supply a custom <code>JmsTupleProducer</code> implementation
- * appropriate for the expected message content.
+ * A Storm <code>Spout</code> implementation that listens to a JMS topic or
+ * queue and outputs tuples based on the messages it receives.
+ *
+ * <p><code>JmsSpout</code> instances rely on <code>JmsProducer</code>
+ * implementations to obtain the JMS
+ * <code>ConnectionFactory</code> and <code>Destination</code> objects necessary
+ * to connect to a JMS topic/queue.
+ *
+ * <p>When a <code>JmsSpout</code> receives a JMS message, it delegates to an
+ * internal <code>JmsTupleProducer</code> instance to create a Storm tuple from
+ * the incoming message.
+ *
+ * <p>Typically, developers will supply a custom <code>JmsTupleProducer</code>
+ * implementation appropriate for the expected message content.
*/
@SuppressWarnings("serial")
public class JmsSpout extends BaseRichSpout implements MessageListener {
+
+ /** The logger object instance for this class. */
private static final Logger LOG = LoggerFactory.getLogger(JmsSpout.class);
- // JMS options
+ /** The logger of the recovery task. */
+ private static final Logger RECOVERY_TASK_LOG =
+ LoggerFactory.getLogger(RecoveryTask.class);
+
+ /** Time to sleep between queue polling attempts. */
+ private static final int POLL_INTERVAL = 50;
--- End diff --
Nit: POLL_INTERVAL_MS. I'm also wondering why this spout is sleeping manually instead of using the built in functionality for this https://github.com/apache/storm/blob/64168037fbf383a350c5be7e220ae137d4317411/conf/defaults.yaml#L247 (you don't need to do anything about this, I'm just asking if we should file an issue for it?).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by omerhadari <gi...@git.apache.org>.
Github user omerhadari commented on the issue:
https://github.com/apache/storm/pull/2234
I see, Fixed that as well.
See if you think someone needs to take a look and verify that my style fixes are the correct ones :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2234
@omerhadari I don't think you need to fix all violations for the module, but of course wouldn't complain if you did ;) If you do I think it should go in another PR though.
Check the pom.xml for storm-jms. It has a count of ignored violations If the build fails now it's because the diff here adds some new checkstyle violations. If you build storm-jms locally, you should get a checkstyle-violations.xml in storm-jms/target. I think it would be good to either fix the new violations (check out master and diff checkstyle-violations with yours), or just fix all violations for the two modified files. If you're up for fixing the violations for the two files, please reduce the number of ignored errors in pom.xml to match.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by omerhadari <gi...@git.apache.org>.
Github user omerhadari commented on the issue:
https://github.com/apache/storm/pull/2234
No problem, I'll open a separate PR tomorrow evening.
Do let me know if this one needs any further work.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by omerhadari <gi...@git.apache.org>.
Github user omerhadari commented on the issue:
https://github.com/apache/storm/pull/2234
Done, and done
https://issues.apache.org/jira/browse/STORM-2652
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2234#discussion_r128850697
--- Diff: external/storm-jms/src/main/java/org/apache/storm/jms/spout/JmsSpout.java ---
@@ -268,18 +366,25 @@ public void ack(Object msgId) {
LOG.warn("Error acknowldging JMS message: " + msgId, e);
}
} else {
- LOG.warn("Couldn't acknowledge unknown JMS message ID: " + msgId);
+ LOG.warn("Couldn't acknowledge unknown JMS message ID: "
+ + msgId);
}
} else {
this.toCommit.remove(msgId);
}
}
- /*
- * Will only be called if we're transactional or not AUTO_ACKNOWLEDGE
+ /**
+ * Fail an unsuccessfully handled message by the its {@link JmsMessageID}.
--- End diff --
Nit: the its
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: Bugfix/fix configuration cast exception
Posted by ptgoetz <gi...@git.apache.org>.
Github user ptgoetz commented on the issue:
https://github.com/apache/storm/pull/2234
@omerhadari Could you remove the wildcard imports? I'm +1 once that's done.
Also do you mind creating a corresponding JIRA ticket for this and updating the title of this pull request to be prefixed with the JIRA ID?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2234
I think the changes look good. Could you squash the commits down so this PR only contains 1 commit where the message matches the PR title?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2234: STORM-2652: fix error in open method of JmsSpout
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/storm/pull/2234
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by ptgoetz <gi...@git.apache.org>.
Github user ptgoetz commented on the issue:
https://github.com/apache/storm/pull/2234
Thanks @vesense!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2234#discussion_r128851000
--- Diff: external/storm-jms/src/main/java/org/apache/storm/jms/spout/JmsSpout.java ---
@@ -288,19 +393,28 @@ public void fail(Object msgId) {
}
}
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ /**
+ * {@inheritDoc}
--- End diff --
Nit: I think javadoc does this automagically. I think inheritDoc is in case you want to add some extra javadoc and then inherit the rest
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by omerhadari <gi...@git.apache.org>.
Github user omerhadari commented on the issue:
https://github.com/apache/storm/pull/2234
I tried looking into the failed integration test, and I believe it has nothing to do with the classes I touched (especially since it looks to have passed a few commits earlier, all of which contained minor documentation and naming changes).
Let me know if there is anything more I can improve within the scope of this PR before finalizing it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2234
@omerhadari It is showing 232 violations for me now, and no violations in JmsSpout. I believe that running checkstyle:check will run checkstyle with the default configuration (and ruleset) rather than the one we have set in the pom, which is why the count is inconsistent. If you want to see how many errors there are, you can set the max violations to 0 and run mvn clean validate.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2234: STORM-2652: fix error in open method of JmsSpout
Posted by omerhadari <gi...@git.apache.org>.
Github user omerhadari commented on a diff in the pull request:
https://github.com/apache/storm/pull/2234#discussion_r128853255
--- Diff: external/storm-jms/src/main/java/org/apache/storm/jms/spout/JmsSpout.java ---
@@ -30,65 +35,110 @@
import javax.jms.MessageListener;
import javax.jms.Session;
-import org.apache.storm.topology.base.BaseRichSpout;
-import org.apache.storm.utils.Utils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
+import org.apache.storm.Config;
import org.apache.storm.jms.JmsProvider;
import org.apache.storm.jms.JmsTupleProducer;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
/**
- * A Storm <code>Spout</code> implementation that listens to a JMS topic or queue
- * and outputs tuples based on the messages it receives.
- * <p>
- * <code>JmsSpout</code> instances rely on <code>JmsProducer</code> implementations
- * to obtain the JMS <code>ConnectionFactory</code> and <code>Destination</code> objects
- * necessary to connect to a JMS topic/queue.
- * <p>
- * When a <code>JmsSpout</code> receives a JMS message, it delegates to an
- * internal <code>JmsTupleProducer</code> instance to create a Storm tuple from the
- * incoming message.
- * <p>
- * Typically, developers will supply a custom <code>JmsTupleProducer</code> implementation
- * appropriate for the expected message content.
+ * A Storm <code>Spout</code> implementation that listens to a JMS topic or
+ * queue and outputs tuples based on the messages it receives.
+ *
+ * <p><code>JmsSpout</code> instances rely on <code>JmsProducer</code>
+ * implementations to obtain the JMS
+ * <code>ConnectionFactory</code> and <code>Destination</code> objects necessary
+ * to connect to a JMS topic/queue.
+ *
+ * <p>When a <code>JmsSpout</code> receives a JMS message, it delegates to an
+ * internal <code>JmsTupleProducer</code> instance to create a Storm tuple from
+ * the incoming message.
+ *
+ * <p>Typically, developers will supply a custom <code>JmsTupleProducer</code>
+ * implementation appropriate for the expected message content.
*/
@SuppressWarnings("serial")
public class JmsSpout extends BaseRichSpout implements MessageListener {
+
+ /** The logger object instance for this class. */
private static final Logger LOG = LoggerFactory.getLogger(JmsSpout.class);
- // JMS options
+ /** The logger of the recovery task. */
+ private static final Logger RECOVERY_TASK_LOG =
+ LoggerFactory.getLogger(RecoveryTask.class);
+
+ /** Time to sleep between queue polling attempts. */
+ private static final int POLL_INTERVAL = 50;
--- End diff --
It probably should use this. There is generally a lot to do in this package, regarding this and also the handling of acked and failed messages in a better way. I intend to go over it and open issues when I get to it (if that's ok?)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2234
@omerhadari For some reason I can't set you as the assignee on the jira issue. Please assign the issue to yourself, that way it is easy to tell who provided the fix, and it won't show up when people search for unassigned issues.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2234#discussion_r128850623
--- Diff: external/storm-jms/src/main/java/org/apache/storm/jms/spout/JmsSpout.java ---
@@ -250,10 +341,17 @@ public void nextTuple() {
}
- /*
- * Will only be called if we're transactional or not AUTO_ACKNOWLEDGE
+ /**
+ * Ack a successfully handled message by the matching {@link JmsMessageID}.
+ *
+ * <p>Acking means removing the message from the pending messages
+ * collections,and if it was the oldest pending message -
--- End diff --
Nit: Missing space at the comma
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by omerhadari <gi...@git.apache.org>.
Github user omerhadari commented on the issue:
https://github.com/apache/storm/pull/2234
I have no permissions to do that. I reopened the issue so that someone who can assigns it to me.
Will work on the 1.x PR ASAP. Thanks for the merge.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on the issue:
https://github.com/apache/storm/pull/2234
@omerhadari I assigned the Jira to you and added you to the Jira contributors.
@srdo I added you to the Jira committers/PMCs.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2234
No, I think this is good. Thanks for the fixes :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2234#discussion_r128855363
--- Diff: external/storm-jms/src/main/java/org/apache/storm/jms/spout/JmsSpout.java ---
@@ -72,19 +72,18 @@
private static final Logger LOG = LoggerFactory.getLogger(JmsSpout.class);
/** The logger of the recovery task. */
- private static final Logger RECOVERY_TASK_LOG =
- LoggerFactory.getLogger(RecoveryTask.class);
+ private static final Logger RECOVERY_TASK_LOG = LoggerFactory.getLogger(RecoveryTask.class);
/** Time to sleep between queue polling attempts. */
- private static final int POLL_INTERVAL = 50;
+ private static final int POLL_INTERVAL_MS = 50;
/**
- * The default value the {@link Config#TOPOLOGY_MESSAGE_TIMEOUT_SECS}.
+ * The default value for {@link Config#TOPOLOGY_MESSAGE_TIMEOUT_SECS}.
*/
- private static final int DEFAULT_MESSAGE_TIMEOUT = 30;
+ private static final int DEFAULT_MESSAGE_TIMEOUT_SECS = 30;
- /** Time to wait before queuing the first recovery task (in seconds). */
- private static final int RECOVERY_DELAY = 10;
+ /** Time to wait before queuing the first recovery task. */
+ private static final int RECOVERY_DELAY_SECS = 10;
--- End diff --
Sorry, I think this is actually milliseconds going by the usage in L262
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #2234: STORM-2652: fix error in open method of JmsSpout
Posted by omerhadari <gi...@git.apache.org>.
Github user omerhadari commented on the issue:
https://github.com/apache/storm/pull/2234
I fixed all style issues reported in `checkstyle-violations.xml` regarding the files I touched, and it still failed. I check out master and ran `mvn checkout:check`, it failed there. Too - with even more errors.
I would be happy to improve the style of other classes in a separate PR. In the meantime, I fixed all style issues regarding `JmsSpout` and `JmsSpoutTest` and updated `pom.xml` to a working allowed violations value.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #2234: STORM-2652: fix error in open method of JmsSpout
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2234#discussion_r128847943
--- Diff: external/storm-jms/src/main/java/org/apache/storm/jms/spout/JmsSpout.java ---
@@ -30,65 +35,110 @@
import javax.jms.MessageListener;
import javax.jms.Session;
-import org.apache.storm.topology.base.BaseRichSpout;
-import org.apache.storm.utils.Utils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
+import org.apache.storm.Config;
import org.apache.storm.jms.JmsProvider;
import org.apache.storm.jms.JmsTupleProducer;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
/**
- * A Storm <code>Spout</code> implementation that listens to a JMS topic or queue
- * and outputs tuples based on the messages it receives.
- * <p>
- * <code>JmsSpout</code> instances rely on <code>JmsProducer</code> implementations
- * to obtain the JMS <code>ConnectionFactory</code> and <code>Destination</code> objects
- * necessary to connect to a JMS topic/queue.
- * <p>
- * When a <code>JmsSpout</code> receives a JMS message, it delegates to an
- * internal <code>JmsTupleProducer</code> instance to create a Storm tuple from the
- * incoming message.
- * <p>
- * Typically, developers will supply a custom <code>JmsTupleProducer</code> implementation
- * appropriate for the expected message content.
+ * A Storm <code>Spout</code> implementation that listens to a JMS topic or
+ * queue and outputs tuples based on the messages it receives.
+ *
+ * <p><code>JmsSpout</code> instances rely on <code>JmsProducer</code>
+ * implementations to obtain the JMS
+ * <code>ConnectionFactory</code> and <code>Destination</code> objects necessary
+ * to connect to a JMS topic/queue.
+ *
+ * <p>When a <code>JmsSpout</code> receives a JMS message, it delegates to an
+ * internal <code>JmsTupleProducer</code> instance to create a Storm tuple from
+ * the incoming message.
+ *
+ * <p>Typically, developers will supply a custom <code>JmsTupleProducer</code>
+ * implementation appropriate for the expected message content.
*/
@SuppressWarnings("serial")
public class JmsSpout extends BaseRichSpout implements MessageListener {
+
+ /** The logger object instance for this class. */
private static final Logger LOG = LoggerFactory.getLogger(JmsSpout.class);
- // JMS options
+ /** The logger of the recovery task. */
+ private static final Logger RECOVERY_TASK_LOG =
+ LoggerFactory.getLogger(RecoveryTask.class);
+
+ /** Time to sleep between queue polling attempts. */
+ private static final int POLL_INTERVAL = 50;
+
+ /**
+ * The default value the {@link Config#TOPOLOGY_MESSAGE_TIMEOUT_SECS}.
--- End diff --
Nit: "The default value for {@link Config#TOPOLOGY_MESSAGE_TIMEOUT_SECS}", and it would be good to rename to DEFAULT_MESSAGE_TIMEOUT_SECS
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---