fix Intermittent test failures in ProxyPublishConsumeTest.socketTest (#253)

pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyAuthenticationTest.java

@@ -16,10 +16,12 @@
  */
 package com.yahoo.pulsar.websocket.proxy;
 
+import static java.util.concurrent.Executors.newFixedThreadPool;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.spy;
 
 import java.net.URI;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
@@ -110,12 +112,21 @@ public class ProxyAuthenticationTest extends ProducerConsumerBase {
         } catch (Throwable t) {
             log.error(t.getMessage());
         } finally {
+            ExecutorService executor = newFixedThreadPool(1);
+            try {
+                executor.submit(() -> {
                     try {
                         consumeClient.stop();
                         produceClient.stop();
+                        log.info("proxy clients are stopped successfully");
                     } catch (Exception e) {
                         log.error(e.getMessage());
                     }
+                }).get(2, TimeUnit.SECONDS);
+            } catch (Exception e) {
+                log.error("failed to close clients ", e);
+            }
+            executor.shutdownNow();
         }
     }
 

pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyPublishConsumeTest.java

@@ -19,7 +19,11 @@ import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.spy;
 
 import java.net.URI;
+import static java.util.concurrent.Executors.newFixedThreadPool;
+
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.bookkeeper.test.PortManager;
 import org.eclipse.jetty.websocket.api.Session;
@@ -71,7 +75,7 @@ public class ProxyPublishConsumeTest extends ProducerConsumerBase {
         log.info("Finished Cleaning Up Test setup");
     }
 
-    @Test(timeOut=30000)
+    @Test(timeOut=10000)
     public void socketTest() throws Exception {
         URI consumeUri = URI.create(CONSUME_URI);
         URI produceUri = URI.create(PRODUCE_URI);
@@ -101,12 +105,21 @@ public class ProxyPublishConsumeTest extends ProducerConsumerBase {
             Assert.assertTrue(produceSocket.getBuffer().size() > 0);
             Assert.assertEquals(produceSocket.getBuffer(), consumeSocket.getBuffer());
         } finally {
+            ExecutorService executor = newFixedThreadPool(1);
+            try {
+                executor.submit(() -> {
                     try {
                         consumeClient.stop();
                         produceClient.stop();
+                        log.info("proxy clients are stopped successfully");
                     } catch (Exception e) {
                         log.error(e.getMessage());
                     }
+                }).get(2, TimeUnit.SECONDS);
+            } catch (Exception e) {
+                log.error("failed to close clients ", e);
+            }
+            executor.shutdownNow();
         }
     }
 

pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyPublishConsumeTls.java

@@ -15,6 +15,7 @@
  */
 package com.yahoo.pulsar.websocket.proxy;
 
+import static java.util.concurrent.Executors.newFixedThreadPool;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.spy;
 
@@ -22,6 +23,7 @@ import java.net.URI;
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
 import java.security.SecureRandom;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
@@ -129,12 +131,21 @@ public class ProxyPublishConsumeTls extends ProducerConsumerBase {
         } catch (Throwable t) {
             log.error(t.getMessage());
         } finally {
+            ExecutorService executor = newFixedThreadPool(1);
+            try {
+                executor.submit(() -> {
                     try {
                         consumeClient.stop();
                         produceClient.stop();
+                        log.info("proxy clients are stopped successfully");
                     } catch (Exception e) {
                         log.error(e.getMessage());
                     }
+                }).get(2, TimeUnit.SECONDS);
+            } catch (Exception e) {
+                log.error("failed to close clients ", e);
+            }
+            executor.shutdownNow();
         }
     }
 

pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyPublishConsumeWithoutZKTest.java

@@ -15,11 +15,14 @@
  */
 package com.yahoo.pulsar.websocket.proxy;
 
+import static java.util.concurrent.Executors.newFixedThreadPool;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.spy;
 
 import java.net.URI;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.bookkeeper.test.PortManager;
 import org.eclipse.jetty.websocket.api.Session;
@@ -100,12 +103,21 @@ public class ProxyPublishConsumeWithoutZKTest extends ProducerConsumerBase {
             Assert.assertTrue(produceSocket.getBuffer().size() > 0);
             Assert.assertEquals(produceSocket.getBuffer(), consumeSocket.getBuffer());
         } finally {
+            ExecutorService executor = newFixedThreadPool(1);
+            try {
+                executor.submit(() -> {
                     try {
                         consumeClient.stop();
                         produceClient.stop();
+                        log.info("proxy clients are stopped successfully");
                     } catch (Exception e) {
                         log.error(e.getMessage());
                     }
+                }).get(2, TimeUnit.SECONDS);
+            } catch (Exception e) {
+                log.error("failed to close clients ", e);
+            }
+            executor.shutdownNow();
         }
     }