fix acknowledgeMode bug in trident spout
Project: http://git-wip-us.apache.org/repos/asf/storm/repo Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/3fc7f30e Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/3fc7f30e Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/3fc7f30e Branch: refs/heads/master Commit: 3fc7f30ed693354f014f00948c527e0f2273c66c Parents: 6212947 Author: P. Taylor Goetz <ptgo...@gmail.com> Authored: Mon Oct 27 15:51:16 2014 -0400 Committer: P. Taylor Goetz <ptgo...@gmail.com> Committed: Mon Oct 27 15:51:16 2014 -0400 ---------------------------------------------------------------------- examples/src/main/resources/jms-activemq.xml | 2 +- src/main/java/backtype/storm/contrib/jms/spout/JmsSpout.java | 3 +-- .../java/backtype/storm/contrib/jms/trident/TridentJmsSpout.java | 2 +- 3 files changed, 3 insertions(+), 4 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/storm/blob/3fc7f30e/examples/src/main/resources/jms-activemq.xml ---------------------------------------------------------------------- diff --git a/examples/src/main/resources/jms-activemq.xml b/examples/src/main/resources/jms-activemq.xml index b720ae3..1a845b8 100644 --- a/examples/src/main/resources/jms-activemq.xml +++ b/examples/src/main/resources/jms-activemq.xml @@ -1,4 +1,4 @@ -<?xml version="1.0"?> +<?xml version="1.0" encoding="UTF-8"?> <!-- Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with http://git-wip-us.apache.org/repos/asf/storm/blob/3fc7f30e/src/main/java/backtype/storm/contrib/jms/spout/JmsSpout.java ---------------------------------------------------------------------- diff --git a/src/main/java/backtype/storm/contrib/jms/spout/JmsSpout.java b/src/main/java/backtype/storm/contrib/jms/spout/JmsSpout.java index a2bdb4f..424708d 100644 --- a/src/main/java/backtype/storm/contrib/jms/spout/JmsSpout.java +++ b/src/main/java/backtype/storm/contrib/jms/spout/JmsSpout.java @@ -214,8 +214,7 @@ public class JmsSpout extends BaseRichSpout implements MessageListener { // ack if we're not in AUTO_ACKNOWLEDGE mode, or the message requests ACKNOWLEDGE LOG.debug("Requested deliveryMode: " + toDeliveryModeString(msg.getJMSDeliveryMode())); LOG.debug("Our deliveryMode: " + toDeliveryModeString(this.jmsAcknowledgeMode)); - if (this.isDurableSubscription() - || (msg.getJMSDeliveryMode() != Session.AUTO_ACKNOWLEDGE)) { + if (this.isDurableSubscription()) { LOG.debug("Requesting acks."); JmsMessageID messageId = new JmsMessageID(this.messageSequence++, msg.getJMSMessageID()); this.collector.emit(vals, messageId); http://git-wip-us.apache.org/repos/asf/storm/blob/3fc7f30e/src/main/java/backtype/storm/contrib/jms/trident/TridentJmsSpout.java ---------------------------------------------------------------------- diff --git a/src/main/java/backtype/storm/contrib/jms/trident/TridentJmsSpout.java b/src/main/java/backtype/storm/contrib/jms/trident/TridentJmsSpout.java index ef292ba..7b87959 100644 --- a/src/main/java/backtype/storm/contrib/jms/trident/TridentJmsSpout.java +++ b/src/main/java/backtype/storm/contrib/jms/trident/TridentJmsSpout.java @@ -321,7 +321,7 @@ public class TridentJmsSpout implements ITridentSpout<JmsBatch> { } try { - if (msg.getJMSDeliveryMode() != Session.AUTO_ACKNOWLEDGE) { + if (TridentJmsSpout.this.jmsAcknowledgeMode != Session.AUTO_ACKNOWLEDGE) { batchMessages.add(msg); } Values tuple = tupleProducer.toTuple(msg);
