You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by gt...@apache.org on 2009/11/23 19:43:02 UTC
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
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());
}