[ML] Avoid 5s wait in AbstractNativeProcessTests (elastic#74916)

Calling close() before counting down the latch was causing a 5 second wait and timeout in the tests

Author: davidkyle

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/process/AbstractNativeProcessTests.java

@@ -86,19 +86,23 @@ public void terminateExecutorService() {
     public void testStart_DoNotDetectCrashWhenNoInputPipeProvided() throws Exception {
         when(processPipes.getProcessInStream()).thenReturn(Optional.empty());
         try (AbstractNativeProcess process = new TestNativeProcess()) {
-            process.start(executorService);
-        } finally {
-            mockNativeProcessLoggingStreamEnds.countDown();
-            // Not detecting a crash is confirmed in terminateExecutorService()
+            try {
+                process.start(executorService);
+            } finally {
+                mockNativeProcessLoggingStreamEnds.countDown();
+                // Not detecting a crash is confirmed in terminateExecutorService()
+            }
         }
     }
 
     public void testStart_DoNotDetectCrashWhenProcessIsBeingClosed() throws Exception {
         try (AbstractNativeProcess process = new TestNativeProcess()) {
-            process.start(executorService);
-        } finally {
-            mockNativeProcessLoggingStreamEnds.countDown();
-            // Not detecting a crash is confirmed in terminateExecutorService()
+            try {
+                process.start(executorService);
+            } finally {
+                mockNativeProcessLoggingStreamEnds.countDown();
+                // Not detecting a crash is confirmed in terminateExecutorService()
+            }
         }
     }
 
@@ -142,65 +146,74 @@ public void testCrashReporting() throws Exception {
 
     public void testWriteRecord() throws Exception {
         try (AbstractNativeProcess process = new TestNativeProcess()) {
-            process.start(executorService);
-            process.writeRecord(new String[] {"a", "b", "c"});
-            process.flushStream();
-
-            verify(inputStream).write(any(), anyInt(), anyInt());
-
-        } finally {
-            mockNativeProcessLoggingStreamEnds.countDown();
+            try {
+                process.start(executorService);
+                process.writeRecord(new String[]{"a", "b", "c"});
+                process.flushStream();
+                verify(inputStream).write(any(), anyInt(), anyInt());
+            } finally {
+                mockNativeProcessLoggingStreamEnds.countDown();
+            }
         }
     }
 
     public void testWriteRecord_FailWhenNoInputPipeProvided() throws Exception {
         when(processPipes.getProcessInStream()).thenReturn(Optional.empty());
         try (AbstractNativeProcess process = new TestNativeProcess()) {
-            process.start(executorService);
-            expectThrows(NullPointerException.class, () -> process.writeRecord(new String[] {"a", "b", "c"}));
-        } finally {
-            mockNativeProcessLoggingStreamEnds.countDown();
+            try {
+                process.start(executorService);
+                expectThrows(NullPointerException.class, () -> process.writeRecord(new String[]{"a", "b", "c"}));
+            } finally {
+                mockNativeProcessLoggingStreamEnds.countDown();
+            }
         }
     }
 
     public void testFlush() throws Exception {
         try (AbstractNativeProcess process = new TestNativeProcess()) {
-            process.start(executorService);
-            process.flushStream();
-
-            verify(inputStream).flush();
-        } finally {
-            mockNativeProcessLoggingStreamEnds.countDown();
+            try {
+                process.start(executorService);
+                process.flushStream();
+                verify(inputStream).flush();
+            } finally {
+                mockNativeProcessLoggingStreamEnds.countDown();
+            }
         }
     }
 
     public void testFlush_FailWhenNoInputPipeProvided() throws Exception {
         when(processPipes.getProcessInStream()).thenReturn(Optional.empty());
         try (AbstractNativeProcess process = new TestNativeProcess()) {
-            process.start(executorService);
-            expectThrows(NullPointerException.class, process::flushStream);
-        } finally {
-            mockNativeProcessLoggingStreamEnds.countDown();
+            try {
+                process.start(executorService);
+                expectThrows(NullPointerException.class, process::flushStream);
+            } finally {
+                mockNativeProcessLoggingStreamEnds.countDown();
+            }
         }
     }
 
     public void testIsReady() throws Exception {
         try (AbstractNativeProcess process = new TestNativeProcess()) {
-            process.start(executorService);
-            assertThat(process.isReady(), is(false));
-            process.setReady();
-            assertThat(process.isReady(), is(true));
-        } finally {
-            mockNativeProcessLoggingStreamEnds.countDown();
+            try {
+                process.start(executorService);
+                assertThat(process.isReady(), is(false));
+                process.setReady();
+                assertThat(process.isReady(), is(true));
+            } finally {
+                mockNativeProcessLoggingStreamEnds.countDown();
+            }
         }
     }
 
     public void testConsumeAndCloseOutputStream_GivenNoOutputStream() throws Exception {
         when(processPipes.getProcessOutStream()).thenReturn(Optional.empty());
         try (AbstractNativeProcess process = new TestNativeProcess()) {
-            process.consumeAndCloseOutputStream();
-        } finally {
-            mockNativeProcessLoggingStreamEnds.countDown();
+            try {
+                process.consumeAndCloseOutputStream();
+            } finally {
+                mockNativeProcessLoggingStreamEnds.countDown();
+            }
         }
     }