svn commit: r1494301 - /incubator/ambari/trunk/ambari-server/src/test/java/org/apache/ambari/server/agent/TestActionQueue.java

[swagle]

Author: swagle
Date: Tue Jun 18 21:02:02 2013
New Revision: 1494301

URL: http://svn.apache.org/r1494301
Log:
AMBARI-2416. Test Concurrent operations fail intermittently on trunk. (Myroslav 
Papirkovskyy via swagle)

Modified:
    
incubator/ambari/trunk/ambari-server/src/test/java/org/apache/ambari/server/agent/TestActionQueue.java

Modified: 
incubator/ambari/trunk/ambari-server/src/test/java/org/apache/ambari/server/agent/TestActionQueue.java
URL: 
http://svn.apache.org/viewvc/incubator/ambari/trunk/ambari-server/src/test/java/org/apache/ambari/server/agent/TestActionQueue.java?rev=1494301&r1=1494300&r2=1494301&view=diff
==============================================================================
--- 
incubator/ambari/trunk/ambari-server/src/test/java/org/apache/ambari/server/agent/TestActionQueue.java
 (original)
+++ 
incubator/ambari/trunk/ambari-server/src/test/java/org/apache/ambari/server/agent/TestActionQueue.java
 Tue Jun 18 21:02:02 2013
@@ -19,6 +19,7 @@ package org.apache.ambari.server.agent;
 
 import static org.junit.Assert.*;
 
+import java.util.ArrayList;
 import java.util.List;
 
 import org.junit.Test;
@@ -133,10 +134,14 @@ public class TestActionQueue {
     ActionQueueOperation[] dequeOperators = new 
ActionQueueOperation[threadCount];
     ActionQueueOperation[] dequeAllOperators = new 
ActionQueueOperation[threadCount];
 
+    List<Thread> producers = new ArrayList<Thread>();
+    List<Thread> consumers = new ArrayList<Thread>();
+
     for (int i = 0; i < threadCount; i++) {
       dequeOperators[i] = new ActionQueueOperation(aq, hosts,
           ActionQueueOperation.OpType.DEQUEUE);
       Thread t = new Thread(dequeOperators[i]);
+      consumers.add(t);
       t.start();
     }
 
@@ -144,6 +149,7 @@ public class TestActionQueue {
       enqueOperators[i] = new ActionQueueOperation(aq, hosts,
           ActionQueueOperation.OpType.ENQUEUE);
       Thread t = new Thread(enqueOperators[i]);
+      producers.add(t);
       t.start();
     }
 
@@ -151,6 +157,7 @@ public class TestActionQueue {
       dequeAllOperators[i] = new ActionQueueOperation(aq, hosts,
           ActionQueueOperation.OpType.DEQUEUEALL);
       Thread t = new Thread(dequeAllOperators[i]);
+      consumers.add(t);
       t.start();
     }
 
@@ -162,6 +169,10 @@ public class TestActionQueue {
       enqueOperators[i].stop();
     }
 
+    for (Thread producer : producers) {
+      producer.join();
+    }
+
     // Give time to get everything dequeued
     boolean allDequeued = false;
     while (!allDequeued) {
@@ -180,6 +191,10 @@ public class TestActionQueue {
       dequeOperators[i].stop();
       dequeAllOperators[i].stop();
     }
+
+    for (Thread consumer : consumers) {
+      consumer.join();
+    }
     
     for (int h = 0; h<hosts.length; h++) {
       long opsEnqueued = 0;