Github user arunmahadevan commented on a diff in the pull request:

    https://github.com/apache/storm/pull/2639#discussion_r183539214
  
    --- Diff: 
external/storm-jms/src/main/java/org/apache/storm/jms/spout/JmsSpout.java ---
    @@ -339,26 +339,26 @@ public void nextTuple() {
          */
         @Override
         public void ack(Object msgId) {
    -
             Message msg = this.pendingMessages.remove(msgId);
    -        JmsMessageID oldest = this.toCommit.first();
    -        if (msgId.equals(oldest)) {
    -            if (msg != null) {
    -                try {
    -                    LOG.debug("Committing...");
    -                    msg.acknowledge();
    -                    LOG.debug("JMS Message acked: " + msgId);
    -                    this.toCommit.remove(msgId);
    -                } catch (JMSException e) {
    -                    LOG.warn("Error acknowldging JMS message: " + msgId, 
e);
    +        if (!toCommit.isEmpty()) {
    +            JmsMessageID oldest = this.toCommit.first();
    +            if (msgId.equals(oldest)) {
    +                if (msg != null) {
    +                    try {
    +                        LOG.debug("Committing...");
    +                        msg.acknowledge();
    --- End diff --
    
    I am not sure acking the oldest message in JMS is correct even for 
`CLIENT_ACKNOWLEDGE`. This would ack the new messages that have been consumed 
in the session (and possibly emitted) even before the spout received the ACK 
for the message. I guess we should keep removing the message from `toCommit` 
and invoke the JMS ack when its the last message in `toCommit`. (assuming we 
dont consume any other message in the meanwhile).


---

Reply via email to