[jboss-cvs] JBoss Messaging SVN: r6165 - in branches/Branch_1_4: tests/src/org/jboss/test/messaging/jms/clustering and 1 other directory.

[jboss-cvs-commits at lists.jboss.org]

Author: gaohoward
Date: 2009-03-25 12:28:10 -0400 (Wed, 25 Mar 2009)
New Revision: 6165

Modified:
   branches/Branch_1_4/src/main/org/jboss/messaging/core/impl/clusterconnection/MessageSucker.java
   branches/Branch_1_4/tests/src/org/jboss/test/messaging/jms/clustering/FailoverTest.java
Log:
JBMESSAGING-1547


Modified: branches/Branch_1_4/src/main/org/jboss/messaging/core/impl/clusterconnection/MessageSucker.java
===================================================================
--- branches/Branch_1_4/src/main/org/jboss/messaging/core/impl/clusterconnection/MessageSucker.java	2009-03-25 15:41:58 UTC (rev 6164)
+++ branches/Branch_1_4/src/main/org/jboss/messaging/core/impl/clusterconnection/MessageSucker.java	2009-03-25 16:28:10 UTC (rev 6165)
@@ -129,66 +129,68 @@
 								
 		consumer.setMessageListener(this);		
 		
+		started = true;
 		//Register ourselves with the local queue - this queue will handle flow control for us
 		
 		if (trace) { log.trace(this + " Registering sucker"); }
 		
 		localQueue.registerSucker(this);
 		
-		started = true;
-		
 		if (trace) { log.trace(this + " Registered sucker"); }
 	}
 	
-	synchronized void stop()
-	{
-		if (!started)
-		{
-			return;
-		}
-		
-		setConsuming(false);
-				
-		localQueue.unregisterSucker(this);
-		
-		try
-		{
-			consumer.closing(-1);
-		}
-		catch (Throwable t)
-		{
-			// Ignore
-		}
-		try
-		{
-			consumer.close();
-		}
-		catch (Throwable t)
-		{
-			//Ignore
-		}
-		
-		try
-		{
-			producer.close();
-		}
-		catch (Throwable t)
-		{
-			//Ignore
-		}
+   void stop()
+   {
+      localQueue.unregisterSucker(this);
 
-		sourceSession = null;
-		
-		localSession = null;
-		
-		consumer = null;
-		
-		clientConsumer = null;
-		
-		producer = null;
-		
-		started = false;
-	}
+      synchronized (this)
+      {
+         if (!started)
+         {
+            return;
+         }
+
+         setConsuming(false);
+
+         try
+         {
+            consumer.closing(-1);
+         }
+         catch (Throwable t)
+         {
+            // Ignore
+         }
+         try
+         {
+            consumer.close();
+         }
+         catch (Throwable t)
+         {
+            // Ignore
+         }
+
+         try
+         {
+            producer.close();
+         }
+         catch (Throwable t)
+         {
+            // Ignore
+         }
+
+         sourceSession = null;
+
+         localSession = null;
+
+         consumer = null;
+
+         clientConsumer = null;
+
+         producer = null;
+
+         started = false;
+      }
+   }
 	
 	public String getQueueName()
 	{
@@ -198,6 +200,11 @@
 	public synchronized void setConsuming(boolean consume)
 	{
 		if (trace) { log.trace(this + " setConsuming " + consume); }
+      
+		if (!started)
+      {
+         return;
+      }
 		
 		try
 		{

Modified: branches/Branch_1_4/tests/src/org/jboss/test/messaging/jms/clustering/FailoverTest.java
===================================================================
--- branches/Branch_1_4/tests/src/org/jboss/test/messaging/jms/clustering/FailoverTest.java	2009-03-25 15:41:58 UTC (rev 6164)
+++ branches/Branch_1_4/tests/src/org/jboss/test/messaging/jms/clustering/FailoverTest.java	2009-03-25 16:28:10 UTC (rev 6165)
@@ -6,13 +6,16 @@
  */
 package org.jboss.test.messaging.jms.clustering;
 
+import java.util.ArrayList;
 import java.util.Enumeration;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Set;
 
 import javax.jms.Connection;
 import javax.jms.ConnectionFactory;
 import javax.jms.DeliveryMode;
+import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.MessageConsumer;
 import javax.jms.MessageProducer;
@@ -32,6 +35,7 @@
 
 /**
  * @author <a href="mailto:ovidiu at feodorov.com">Ovidiu Feodorov</a>
+ * @author <a href="mailto:hgao at redhat.com">Howard Gao</a>
  * @version <tt>$Revision$</tt>
  *
  * $Id$
@@ -53,6 +57,54 @@
 
    // Public ---------------------------------------------------------------------------------------
 
+   //https://jira.jboss.org/jira/browse/JBMESSAGING-1547
+   //the dead lock happens when node0 is trying to deliver while node1 
+   //was shutdown.
+   public void testMessageSuckerStopDeadLock() throws Exception
+   {
+      Connection conn = null;
+
+      try 
+      {
+         conn = createConnectionOnServer(cf, 0);
+         conn.start();
+
+         Session session0 = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+         //killing node 1
+         ServerManagement.kill(1);
+
+         //this will cause the MessagingQueue.informSuckers() called.
+         //there is a chance when nodeLeft is detected, MessageSucker.stop() is being called
+         //just to collide with the informSuckers() call.
+         for (int i = 0; i < 10000; i++)
+         {
+            MessageConsumer consumer1 = session0.createConsumer(queue[0]);
+            consumer1.close();
+            Thread.yield();
+         }
+         
+         //if the deak lock happens, server1 won't be delivering any messages.
+         MessageProducer prod = session0.createProducer(queue[0]);
+         TextMessage msg = session0.createTextMessage("deadlock");
+         prod.send(msg);
+         
+         MessageConsumer cons = session0.createConsumer(queue[0]);
+         msg = (TextMessage)cons.receive(5000);
+         
+         assertNotNull(msg);
+         assertEquals("deadlock", msg.getText());
+         
+      }
+      finally
+      {
+         if (conn != null)
+         {
+            conn.close();
+         }
+      }
+   }
+   
    public void testSimpleConnectionFailover() throws Exception
    {
    	//We need to sleep and relookup the connection factory due to http://jira.jboss.com/jira/browse/JBMESSAGING-1038
@@ -2343,7 +2395,5 @@
    }
 
 
-
    // Inner classes --------------------------------------------------------------------------------
-
 }