qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ritch...@apache.org
Subject svn commit: r764079 - in /qpid/trunk/qpid/java: broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java systests/src/main/java/org/apache/qpid/test/unit/close/VerifyAckingOkDuringClose.java
Date Fri, 10 Apr 2009 22:10:31 GMT
Author: ritchiem
Date: Fri Apr 10 22:10:31 2009
New Revision: 764079

URL: http://svn.apache.org/viewvc?rev=764079&view=rev
Log:
QPID-1791 : Replaced the get call before the dequeueAndDelete with a remove. So the remove
call after the delete is now nolonger requried.
Added manual test to assist identifing the problem

Added:
    qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/close/VerifyAckingOkDuringClose.java
Modified:
    qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java

Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java?rev=764079&r1=764078&r2=764079&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java
(original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java
Fri Apr 10 22:10:31 2009
@@ -158,7 +158,7 @@
         else
         {
             QueueEntry queueEntry;
-            queueEntry = unacknowledgedMessageMap.get(deliveryTag);
+            queueEntry = unacknowledgedMessageMap.remove(deliveryTag);
 
             if (debug)
             {
@@ -187,9 +187,6 @@
             // from the transaciton log
             queueEntry.dequeueAndDelete(_storeContext);
 
-            unacknowledgedMessageMap.remove(deliveryTag);
-
-
         }
         if(_inTran)
         {

Added: qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/close/VerifyAckingOkDuringClose.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/close/VerifyAckingOkDuringClose.java?rev=764079&view=auto
==============================================================================
--- qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/close/VerifyAckingOkDuringClose.java
(added)
+++ qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/close/VerifyAckingOkDuringClose.java
Fri Apr 10 22:10:31 2009
@@ -0,0 +1,160 @@
+/*
+ *
+ * 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.qpid.test.unit.close;
+
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.client.AMQConnectionFactory;
+import org.apache.qpid.jndi.PropertiesFileInitialContextFactory;
+
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.naming.InitialContext;
+import java.util.ArrayList;
+
+/**
+ * QPID-1791
+ *
+ * The threading model in the Java broker (at least till 0.5) allows for the
+ * close to be handled immediately even if the broker is still processing state
+ * for that Session.
+ *
+ * This test verifys that QPID-1791 is has been handled.
+ *
+ * The problem was that the whilst the Session is busy processing Acks from the
+ * client the Close frame jumps in and clears the unAcknowledgeMap in an
+ * attempt to start processing them for closing the connection.
+ *
+ * If the session had a consumer consuming from a temporary queue. The closing
+ * thread dequeues and deletes the message that were on the uncknowledgedMap.
+ *
+ * However, the Acking thread currently does:
+ * queuEntry = unackedMap.get(messageID)
+ *
+ * dequeueAndDelete(queueEntry)
+ *
+ * unackedMap.remove(messageID)
+ *
+ * As a result the queueEntry is sitting in the unackedMap whilst it is being
+ * dequeuedAndDeleted which leaves the opportunity for the close thread to
+ * remove contents of the unackedMap for processing. The close thread will then
+ * dequeueAndDelete all these values one of which the acking thread is currently
+ * processing.
+ *
+ *
+ * Test Approach
+ *
+ * Send a lot of persistent messages (5000), the goal of which is to fill the
+ * pretch and to provide the broker with a lot of acks to process
+ *
+ * Using client ack and prefetch buffer of 5000 use receive to get 2500
+ * Use AMQMessage.acknowledgeThis() to send a single ack frame back to the
+ * broker per message so 2500 ack frames.
+ * This will give the broker a lot to process,
+ * Immediately send the consumer close after the acks are all gone.
+ * This will cause the remaining 2500 prefetched messages plus any that have
+ * not yet had their acks processed
+ * to be collected by the requeue() process potentially
+ */
+public class VerifyAckingOkDuringClose
+{
+
+    static final int MESSAGE_SENT = 5000;
+
+    public static void main(String[] args) throws Exception
+    {
+        //Check that we have the InitialContext Configured
+
+        if (System.getProperty(InitialContext.INITIAL_CONTEXT_FACTORY) == null)
+        {
+            System.setProperty(InitialContext.INITIAL_CONTEXT_FACTORY, PropertiesFileInitialContextFactory.class.getName());
+        }
+
+        if (System.getProperty(InitialContext.PROVIDER_URL) == null)
+        {
+            System.err.println(InitialContext.PROVIDER_URL + ": Is not set and is required
to contain a 'default' ConnectionFactory.");
+            System.exit(1);
+        }
+
+        //Retreive the local factory from the properties file
+        // when used with perftest.properties this will be localhost:5672
+        AMQConnectionFactory factory = (AMQConnectionFactory) new InitialContext().lookup("default");
+
+        AMQConnection connection = (AMQConnection) factory.createConnection("guest", "guest");
+
+        //Use the AMQConnection Interface to set the prefetch to the number
+        // we are sending
+        Session session = connection.createSession(false,
+                                                   Session.CLIENT_ACKNOWLEDGE,
+                                                   MESSAGE_SENT);
+
+        Queue queue = session.createTemporaryQueue();
+
+        MessageConsumer consumer = session.createConsumer(queue);
+        connection.start();
+       
+        MessageProducer producer = session.createProducer(queue);
+
+        Message message = session.createTextMessage("Close");
+
+        for (int i = 0; i < MESSAGE_SENT; i++)
+        {
+            message.setIntProperty("SequenceNumber", i);
+
+            producer.send(message);
+        }
+
+        // Put a reasonable about of data on the queue.
+
+        //Receive all the messags
+        ArrayList<Message> received = new ArrayList<Message>();
+
+        message = consumer.receive(2000);
+
+        while (message != null)
+        {
+            received.add(message);
+            message = consumer.receive(2000);
+        }
+
+        //Check we have all the messages
+        if (received.size() != MESSAGE_SENT)
+        {
+            System.err.println("Test Failed Not all the messages received:" + received.size());
+            System.exit(1);
+        }
+
+        //individually ack the first half then close
+        for (int i = 0; i < MESSAGE_SENT / 2; i++)
+        {
+            ((org.apache.qpid.jms.Message) received.get(i)).acknowledgeThis();
+        }
+
+        // Close the Session to force a requeue on the server of the unackedMsgs
+
+        System.out.println("Killing client to force requeue on broker");
+
+        System.exit(1);
+    }
+
+}



---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project:      http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org


Mime
View raw message