activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gtu...@apache.org
Subject svn commit: r883458 - in /activemq/trunk/activemq-core/src: main/java/org/apache/activemq/ActiveMQMessageConsumer.java test/java/org/apache/activemq/bugs/AMQ2489Test.java test/java/org/apache/activemq/usecases/ExpiredMessagesWithNoConsumerTest.java
Date Mon, 23 Nov 2009 18:43:02 GMT
Author: gtully
Date: Mon Nov 23 18:43:01 2009
New Revision: 883458

URL: http://svn.apache.org/viewvc?rev=883458&view=rev
Log:
resolve https://issues.apache.org/activemq/browse/AMQ-2489 - duplicate delivery acks resulted
in broker exceptions with client or inividual ack - delivery acks now only for unacked messages
- down side is pending messages in broker remain on expiry awaiting ack from ackLaer that
dependes on prefetch value - but this is reasonable and to be expected. they will be removed
on close or subsequent acks in any event

Added:
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2489Test.java 
 (with props)
Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/ExpiredMessagesWithNoConsumerTest.java

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java?rev=883458&r1=883457&r2=883458&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
Mon Nov 23 18:43:01 2009
@@ -831,7 +831,13 @@
             } else if (isAutoAcknowledgeBatch()) {
                 ackLater(md, MessageAck.STANDARD_ACK_TYPE);
             } else if (session.isClientAcknowledge()||session.isIndividualAcknowledge())
{
-                ackLater(md, MessageAck.DELIVERED_ACK_TYPE);
+                boolean messageUnackedByConsumer = false;
+                synchronized (deliveredMessages) {
+                    messageUnackedByConsumer = deliveredMessages.contains(md);
+                }
+                if (messageUnackedByConsumer) {
+                    ackLater(md, MessageAck.DELIVERED_ACK_TYPE);
+                }
             } 
             else {
                 throw new IllegalStateException("Invalid session state.");

Added: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2489Test.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2489Test.java?rev=883458&view=auto
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2489Test.java (added)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2489Test.java Mon
Nov 23 18:43:01 2009
@@ -0,0 +1,226 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.bugs;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import javax.jms.Connection;
+import javax.jms.DeliveryMode;
+import javax.jms.ExceptionListener;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+
+import org.apache.activemq.ActiveMQSession;
+import org.apache.activemq.TestSupport;
+import org.apache.activemq.command.ActiveMQQueue;
+
+/**
+ * In CLIENT_ACKNOWLEDGE and INDIVIDUAL_ACKNOWLEDGE modes following exception
+ * occurs when ASYNCH consumers acknowledges messages in not in order they
+ * received the messages.
+ * <p>
+ * Exception thrown on broker side:
+ * <p>
+ * {@code javax.jms.JMSException: Could not correlate acknowledgment with
+ * dispatched message: MessageAck}
+ * 
+ * @author daroo
+ */
+public class AMQ2489Test extends TestSupport {
+    private final static String SEQ_NUM_PROPERTY = "seqNum";
+
+    private final static int TOTAL_MESSAGES_CNT = 2;
+    private final static int CONSUMERS_CNT = 2;
+
+    private final CountDownLatch LATCH = new CountDownLatch(TOTAL_MESSAGES_CNT);
+
+    private Connection connection;
+
+    protected void setUp() throws Exception {
+        super.setUp();
+        connection = createConnection();
+    }
+
+    protected void tearDown() throws Exception {
+        if (connection != null) {
+            connection.close();
+            connection = null;
+        }
+        super.tearDown();
+    }
+
+    public void testUnorderedClientAcknowledge() throws Exception {
+        doUnorderedAck(Session.CLIENT_ACKNOWLEDGE);
+    }
+
+    public void testUnorderedIndividualAcknowledge() throws Exception {
+        doUnorderedAck(ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE);
+    }
+
+    /**
+     * Main test method
+     * 
+     * @param acknowledgmentMode
+     *            - ACK mode to be used by consumers
+     * @throws Exception
+     */
+    protected void doUnorderedAck(int acknowledgmentMode) throws Exception {
+        List<Consumer> consumers = null;
+        Session producerSession = null;
+
+        connection.start();
+        // Because exception is thrown on broker side only, let's set up
+        // exception listener to get it
+        final TestExceptionListener exceptionListener = new TestExceptionListener();
+        connection.setExceptionListener(exceptionListener);
+        try {
+            consumers = new ArrayList<Consumer>();
+            // start customers
+            for (int i = 0; i < CONSUMERS_CNT; i++) {
+                consumers.add(new Consumer(acknowledgmentMode));
+            }
+
+            // produce few test messages
+            producerSession = connection.createSession(false,
+                    Session.AUTO_ACKNOWLEDGE);
+            final MessageProducer producer = producerSession
+                    .createProducer(new ActiveMQQueue(getQueueName()));
+            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+            for (int i = 0; i < TOTAL_MESSAGES_CNT; i++) {
+                final Message message = producerSession
+                        .createTextMessage("test");
+                // assign each message sequence number
+                message.setIntProperty(SEQ_NUM_PROPERTY, i);
+                producer.send(message);
+            }
+
+            // during each onMessage() calls consumers decreases the LATCH
+            // counter.
+            // 
+            // so, let's wait till all messages are consumed.
+            //
+            LATCH.await();
+
+            // wait a bit more to give exception listener a chance be populated
+            // with
+            // broker's error
+            TimeUnit.SECONDS.sleep(1);
+
+            assertFalse(exceptionListener.getStatusText(), exceptionListener.hasExceptions());
+
+        } finally {
+            if (producerSession != null)
+                producerSession.close();
+
+            if (consumers != null) {
+                for (Consumer c : consumers) {
+                    c.close();
+                }
+            }
+        }
+    }
+
+    protected String getQueueName() {
+        return getClass().getName() + "." + getName();
+    }
+
+    public final class Consumer implements MessageListener {
+        final Session session;
+
+        private Consumer(int acknowledgmentMode) {
+            try {
+                session = connection.createSession(false, acknowledgmentMode);
+                final Queue queue = session.createQueue(getQueueName()
+                        + "?consumer.prefetchSize=1");
+                final MessageConsumer consumer = session.createConsumer(queue);
+                consumer.setMessageListener(this);
+            } catch (JMSException e) {
+                e.printStackTrace();
+                throw new RuntimeException(e);
+            }
+        }
+
+        public void onMessage(Message message) {
+            try {
+                // retrieve sequence number assigned by producer...
+                final int seqNum = message.getIntProperty(SEQ_NUM_PROPERTY);
+
+                // ...and let's delay every second message a little bit before
+                // acknowledgment
+                if ((seqNum % 2) == 0) {
+                    System.out.println("Delayed message sequence numeber: "
+                            + seqNum);
+                    try {
+                        TimeUnit.SECONDS.sleep(1);
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                    }
+                }
+
+                message.acknowledge();
+            } catch (JMSException e) {
+                e.printStackTrace();
+                throw new RuntimeException(e);
+            } finally {
+                // decrease LATCH counter in the main test method.
+                LATCH.countDown();
+            }
+        }
+
+        private void close() {
+            if (session != null) {
+                try {
+                    session.close();
+                } catch (JMSException e) {
+                    e.printStackTrace();
+                    throw new RuntimeException(e);
+                }
+            }
+        }
+    }
+
+    public final class TestExceptionListener implements ExceptionListener {
+        private final java.util.Queue<Exception> exceptions = new ConcurrentLinkedQueue<Exception>();
+
+        public void onException(JMSException e) {
+            exceptions.add(e);
+        }
+
+        public boolean hasExceptions() {
+            return exceptions.isEmpty() == false;
+        }
+
+        public String getStatusText() {
+            final StringBuilder str = new StringBuilder();
+            str.append("Exceptions count on broker side: " + exceptions.size()
+                    + ".\nMessages:\n");
+            for (Exception e : exceptions) {
+                str.append(e.getMessage() + "\n\n");
+            }
+            return str.toString();
+        }
+    }
+}

Propchange: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2489Test.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2489Test.java
------------------------------------------------------------------------------
    svn:executable = *

Propchange: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2489Test.java
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/ExpiredMessagesWithNoConsumerTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/ExpiredMessagesWithNoConsumerTest.java?rev=883458&r1=883457&r2=883458&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/ExpiredMessagesWithNoConsumerTest.java
(original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/ExpiredMessagesWithNoConsumerTest.java
Mon Nov 23 18:43:01 2009
@@ -159,7 +159,8 @@
 	// first ack delivered after expiry
     public void testExpiredMessagesWithVerySlowConsumer() throws Exception {
         createBroker();  
-        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
+        final long queuePrefetch = 600;
+        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616?jms.prefetchPolicy.queuePrefetch="
+ queuePrefetch);
         connection = factory.createConnection();
         session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
         producer = session.createProducer(destination);
@@ -222,7 +223,7 @@
             
         assertTrue("all dispatched up to default prefetch ", Wait.waitFor(new Wait.Condition()
{
             public boolean isSatisified() throws Exception {
-                return 1000 == view.getDispatchCount();
+                return queuePrefetch == view.getDispatchCount();
             }
         }));
         assertTrue("All sent have expired ", Wait.waitFor(new Wait.Condition() {
@@ -240,17 +241,29 @@
         
         Wait.waitFor(new Wait.Condition() {
             public boolean isSatisified() throws Exception {
-                return 0 == view.getInFlightCount();
+                // consumer ackLater(delivery ack for expired messages) is based on half
the prefetch value
+                // which will leave half of the prefetch pending till consumer close
+                return (queuePrefetch/2) -1 == view.getInFlightCount();
             }
         });
         LOG.info("enqueue=" + view.getEnqueueCount() + ", dequeue=" + view.getDequeueCount()
                 + ", inflight=" + view.getInFlightCount() + ", expired= " + view.getExpiredCount()
                 + ", size= " + view.getQueueSize());
-        assertEquals("prefetch gets back to 0 ", 0, view.getInFlightCount());
+        
+        
+        assertEquals("inflight reduces to half prefetch minus single delivered message",
(queuePrefetch/2) -1, view.getInFlightCount());
         assertEquals("size gets back to 0 ", 0, view.getQueueSize());
         assertEquals("dequeues match sent/expired ", sendCount, view.getDequeueCount());
         
         consumer.close();
+        
+        Wait.waitFor(new Wait.Condition() {
+            public boolean isSatisified() throws Exception {
+                return 0 == view.getInFlightCount();
+            }
+        });
+        assertEquals("inflight goes to zeor on close", 0, view.getInFlightCount());
+      
         LOG.info("done: " + getName());
     }
 



Mime
View raw message