activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gtu...@apache.org
Subject svn commit: r762464 - in /activemq/trunk/activemq-core/src: main/java/org/apache/activemq/ test/java/org/apache/activemq/bugs/
Date Mon, 06 Apr 2009 19:20:12 GMT
Author: gtully
Date: Mon Apr  6 19:20:12 2009
New Revision: 762464

URL: http://svn.apache.org/viewvc?rev=762464&view=rev
Log:
ensure duplicate acks are send immediatly and supppress outstanding delivery acks on transport
resumption, AMQ-2149|https://issues.apache.org/activemq/browse/AMQ-2149

Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2149Test.java

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java?rev=762464&r1=762463&r2=762464&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
Mon Apr  6 19:20:12 2009
@@ -1836,12 +1836,6 @@
             TransportListener listener = iter.next();
             listener.transportResumed();
         }
-        for (Iterator<ActiveMQSession> i = this.sessions.iterator(); i.hasNext();)
{
-            ActiveMQSession s = i.next();
-            // deliverAcks at this point is too early as acks can arrive at the broker
-            // before redispatch of messages and hence be out or order
-            s.transportResumed();
-        }
     }
 
     /**

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=762464&r1=762463&r2=762464&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 Apr  6 19:20:12 2009
@@ -1054,6 +1054,14 @@
                             session.connection.rollbackDuplicate(this, old.getMessage());
                         }
                     }
+                    if (pendingAck != null && pendingAck.isDeliveredAck()) {
+                        // on resumption a pending delivered ack will be out of sync with
+                        // re deliveries.
+                        if (LOG.isDebugEnabled()) {
+                            LOG.debug("removing pending delivered ack on transport interupt:
" + pendingAck);
+                        }   
+                        pendingAck = null;
+                    }
                 }
                 if (!unconsumedMessages.isClosed()) {
                     if (this.info.isBrowser() || !session.connection.isDuplicate(this, md.getMessage()))
{
@@ -1085,12 +1093,15 @@
                     } else {
                         // ignore duplicate
                         if (LOG.isDebugEnabled()) {
-                            LOG.debug(getConsumerId() + " ignoring duplicate: " + md.getMessage());
+                            LOG.debug(getConsumerId() + " ignoring(auto acking) duplicate:
" + md.getMessage());
                         }
                         // in a transaction ack delivery of duplicates to ensure prefetch
extension kicks in.
                         // the normal ack will happen in the transaction.
-                        ackLater(md, session.isTransacted() ? 
-                                MessageAck.DELIVERED_ACK_TYPE : MessageAck.STANDARD_ACK_TYPE);
+                        if (session.isTransacted()) {
+                            ackLater(md, MessageAck.DELIVERED_ACK_TYPE);
+                        } else {
+                            acknowledge(md);
+                        }
                     }
                 }
             }
@@ -1159,13 +1170,6 @@
         return lastDeliveredSequenceId;
     }
 
-    // on resumption re deliveries will percolate acks in their own good time
-    public void transportResumed() {
-        pendingAck = null; 
-        additionalWindowSize = 0;
-        deliveredCounter = 0;
-    }
-
 	public IOException getFailureError() {
 		return failureError;
 	}

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java?rev=762464&r1=762463&r2=762464&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java Mon
Apr  6 19:20:12 2009
@@ -1962,12 +1962,4 @@
             syncSendPacket(ack);
         }
     }
-
-    public void transportResumed() {
-        for (Iterator<ActiveMQMessageConsumer> iter = consumers.iterator(); iter.hasNext();)
{
-            ActiveMQMessageConsumer consumer = iter.next();
-            consumer.transportResumed();
-        }
-    }
-
 }

Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2149Test.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2149Test.java?rev=762464&r1=762463&r2=762464&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2149Test.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2149Test.java Mon
Apr  6 19:20:12 2009
@@ -39,8 +39,6 @@
 import org.apache.activemq.broker.region.Destination;
 import org.apache.activemq.broker.region.DestinationStatistics;
 import org.apache.activemq.broker.region.RegionBroker;
-import org.apache.activemq.broker.region.policy.PolicyEntry;
-import org.apache.activemq.broker.region.policy.PolicyMap;
 import org.apache.activemq.broker.util.LoggingBrokerPlugin;
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.store.amq.AMQPersistenceAdapterFactory;
@@ -64,19 +62,20 @@
     private final String SEQ_NUM_PROPERTY = "seqNum";
 
     final int MESSAGE_LENGTH_BYTES = 75 * 1024;
-    final long SLEEP_BETWEEN_SEND_MS = 3;
+    final long SLEEP_BETWEEN_SEND_MS = 25;
     final int NUM_SENDERS_AND_RECEIVERS = 10;
     final Object brokerLock = new Object();
     
     private static final long DEFAULT_BROKER_STOP_PERIOD = 20 * 1000;
-    private static final long DEFAULT_NUM_TO_SEND = 1500;
+    private static final long DEFAULT_NUM_TO_SEND = 1400;
     
     long brokerStopPeriod = DEFAULT_BROKER_STOP_PERIOD;
     long numtoSend = DEFAULT_NUM_TO_SEND;
+    long sleepBetweenSend = SLEEP_BETWEEN_SEND_MS;
     String brokerURL = DEFAULT_BROKER_URL;
     
     int numBrokerRestarts = 0;
-    final static int MAX_BROKER_RESTARTS = 4;
+    final static int MAX_BROKER_RESTARTS = 5;
     BrokerService broker;
     Vector<Throwable> exceptions = new Vector<Throwable>();
 
@@ -110,6 +109,7 @@
         dataDirFile = new File("target/"+ getName());
         numtoSend = DEFAULT_NUM_TO_SEND;
         brokerStopPeriod = DEFAULT_BROKER_STOP_PERIOD;
+        sleepBetweenSend = SLEEP_BETWEEN_SEND_MS;
         brokerURL = DEFAULT_BROKER_URL;
     }
     
@@ -244,9 +244,9 @@
                     LOG.error(dest + " send error", e);
                     exceptions.add(e);
                 }
-                if (SLEEP_BETWEEN_SEND_MS > 0) {
+                if (sleepBetweenSend > 0) {
                     try {
-                        Thread.sleep(SLEEP_BETWEEN_SEND_MS);
+                        Thread.sleep(sleepBetweenSend);
                     } catch (InterruptedException e) {
                         LOG.warn(dest + " sleep interrupted", e);
                     }
@@ -301,7 +301,7 @@
     }
 
 
-    public void x_testOrderWithRestart() throws Exception {
+    public void testOrderWithRestart() throws Exception {
         createBroker(new Configurer() {
             public void configure(BrokerService broker) throws Exception {
                 broker.deleteAllMessages();     
@@ -323,7 +323,7 @@
         verifyStats(true);
     }
         
-    public void x_testTopicOrderWithRestart() throws Exception {
+    public void testTopicOrderWithRestart() throws Exception {
         createBroker(new Configurer() {
             public void configure(BrokerService broker) throws Exception {
                 broker.deleteAllMessages();
@@ -351,7 +351,8 @@
     }
     
     public void doTestTransactionalOrderWithRestart(byte destinationType) throws Exception
{
-        numtoSend = 15000;
+        numtoSend = 10000;
+        sleepBetweenSend = 3;
         brokerStopPeriod = 30 * 1000;
               
         createBroker(new Configurer() {



Mime
View raw message