HADOOP-12605. Fix intermittent failure of TestIPC.testIpcWithReaderQueuing (iwasakims)

hadoop-common-project/hadoop-common/CHANGES.txt

@@ -676,6 +676,9 @@ Release 2.9.0 - UNRELEASED
 
   BUG FIXES
 
+    HADOOP-12605. Fix intermittent failure of TestIPC.testIpcWithReaderQueuing
+    (iwasakims)
+
     HADOOP-12655. TestHttpServer.testBindAddress bind port range is wider
     than expected. (Wei-Chiu Chuang via stevel)
 

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java

@@ -27,6 +27,8 @@
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
 
 import java.io.ByteArrayOutputStream;
 import java.io.DataInput;
@@ -71,6 +73,7 @@
 import org.apache.hadoop.io.retry.RetryProxy;
 import org.apache.hadoop.ipc.Client.ConnectionId;
 import org.apache.hadoop.ipc.RPC.RpcKind;
+import org.apache.hadoop.ipc.Server.Call;
 import org.apache.hadoop.ipc.Server.Connection;
 import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto;
 import org.apache.hadoop.net.ConnectTimeoutException;
@@ -84,6 +87,7 @@
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mockito;
+import org.mockito.internal.util.reflection.Whitebox;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 
@@ -703,6 +707,7 @@ public void testIpcWithReaderQueuing() throws Exception {
   // goal is to jam a handler with a connection, fill the callq with
   // connections, in turn jamming the readers - then flood the server and
   // ensure that the listener blocks when the reader connection queues fill
+  @SuppressWarnings("unchecked")
   private void checkBlocking(int readers, int readerQ, int callQ) throws Exception {
     int handlers = 1; // makes it easier
     
@@ -722,6 +727,9 @@ private void checkBlocking(int readers, int readerQ, int callQ) throws Exception
     // start server
     final TestServerQueue server =
         new TestServerQueue(clients, readers, callQ, handlers, conf);
+    CallQueueManager<Call> spy = spy(
+        (CallQueueManager<Call>)Whitebox.getInternalState(server, "callQueue"));
+    Whitebox.setInternalState(server, "callQueue", spy);
     final InetSocketAddress addr = NetUtils.getConnectAddress(server);
     server.start();
 
@@ -757,12 +765,10 @@ public void run() {
       if (i==0) {
         // let first reader block in a call
         server.firstCallLatch.await();
-      } else if (i <= callQ) {
+      }
-        // let subsequent readers jam the callq, will happen immediately 
+      // wait until reader put a call to callQueue, to make sure all readers
-        while (server.getCallQueueLen() != i) {
+      // are blocking on the queue after initialClients threads are started.
-          Thread.sleep(1);
+      verify(spy, timeout(100).times(i + 1)).put(Mockito.<Call>anyObject());
-        }
-      } // additional threads block the readers trying to add to the callq
     }
 
     try {