Set a timeout on kube pod process integration tests to prevent a hang

airbyte-workers/src/test-integration/java/io/airbyte/workers/process/KubePodProcessIntegrationTest.java

@@ -27,13 +27,17 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang.RandomStringUtils;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
 
 // requires kube running locally to run. If using Minikube it requires MINIKUBE=true
+// Must have a timeout on this class because it tests child processes that may misbehave; otherwise this can hang forever during failures.
+@Timeout(value = 5, unit = TimeUnit.MINUTES)
 public class KubePodProcessIntegrationTest {
 
   private static final boolean IS_MINIKUBE = Boolean.parseBoolean(Optional.ofNullable(System.getenv("IS_MINIKUBE")).orElse("false"));
@@ -74,7 +78,7 @@ public void teardown() throws Exception {
   @Test
   public void testSuccessfulSpawning() throws Exception {
     // start a finite process
-    var availablePortsBefore = KubePortManagerSingleton.getInstance().getNumAvailablePorts();
+    final var availablePortsBefore = KubePortManagerSingleton.getInstance().getNumAvailablePorts();
     final Process process = getProcess("echo hi; sleep 1; echo hi2");
     process.waitFor();
 
@@ -87,9 +91,9 @@ public void testSuccessfulSpawning() throws Exception {
   @Test
   public void testSuccessfulSpawningWithQuotes() throws Exception {
     // start a finite process
-    var availablePortsBefore = KubePortManagerSingleton.getInstance().getNumAvailablePorts();
+    final var availablePortsBefore = KubePortManagerSingleton.getInstance().getNumAvailablePorts();
     final Process process = getProcess("echo \"h\\\"i\"; sleep 1; echo hi2");
-    var output = new String(process.getInputStream().readAllBytes());
+    final var output = new String(process.getInputStream().readAllBytes());
     assertEquals("h\"i\nhi2\n", output);
     process.waitFor();
 
@@ -102,7 +106,7 @@ public void testSuccessfulSpawningWithQuotes() throws Exception {
   @Test
   public void testPipeInEntrypoint() throws Exception {
     // start a process that has a pipe in the entrypoint
-    var availablePortsBefore = KubePortManagerSingleton.getInstance().getNumAvailablePorts();
+    final var availablePortsBefore = KubePortManagerSingleton.getInstance().getNumAvailablePorts();
     final Process process = getProcess("echo hi | cat");
     process.waitFor();
 
@@ -115,7 +119,7 @@ public void testPipeInEntrypoint() throws Exception {
   @Test
   public void testExitCodeRetrieval() throws Exception {
     // start a process that requests
-    var availablePortsBefore = KubePortManagerSingleton.getInstance().getNumAvailablePorts();
+    final var availablePortsBefore = KubePortManagerSingleton.getInstance().getNumAvailablePorts();
     final Process process = getProcess("exit 10");
     process.waitFor();
 
@@ -128,7 +132,7 @@ public void testExitCodeRetrieval() throws Exception {
   @Test
   public void testMissingEntrypoint() throws WorkerException, InterruptedException {
     // start a process with an entrypoint that doesn't exist
-    var availablePortsBefore = KubePortManagerSingleton.getInstance().getNumAvailablePorts();
+    final var availablePortsBefore = KubePortManagerSingleton.getInstance().getNumAvailablePorts();
     final Process process = getProcess(null);
     process.waitFor();
 
@@ -141,7 +145,7 @@ public void testMissingEntrypoint() throws WorkerException, InterruptedException
   @Test
   public void testKillingWithoutHeartbeat() throws Exception {
     // start an infinite process
-    var availablePortsBefore = KubePortManagerSingleton.getInstance().getNumAvailablePorts();
+    final var availablePortsBefore = KubePortManagerSingleton.getInstance().getNumAvailablePorts();
     final Process process = getProcess("while true; do echo hi; sleep 1; done");
 
     // kill the heartbeat server
@@ -156,18 +160,18 @@ public void testKillingWithoutHeartbeat() throws Exception {
     assertNotEquals(0, process.exitValue());
   }
 
-  private static String getRandomFile(int lines) {
-    var sb = new StringBuilder();
+  private static String getRandomFile(final int lines) {
+    final var sb = new StringBuilder();
     for (int i = 0; i < lines; i++) {
       sb.append(RandomStringUtils.randomAlphabetic(100));
       sb.append("\n");
     }
     return sb.toString();
   }
 
-  private Process getProcess(String entrypoint) throws WorkerException {
+  private Process getProcess(final String entrypoint) throws WorkerException {
     // these files aren't used for anything, it's just to check for exceptions when uploading
-    var files = ImmutableMap.of(
+    final var files = ImmutableMap.of(
         "file0", "fixed str",
         "file1", getRandomFile(1),
         "file2", getRandomFile(100),
@@ -183,20 +187,20 @@ private Process getProcess(String entrypoint) throws WorkerException {
         entrypoint, WorkerUtils.DEFAULT_RESOURCE_REQUIREMENTS, Map.of());
   }
 
-  private static Set<Integer> getOpenPorts(int count) {
+  private static Set<Integer> getOpenPorts(final int count) {
     final Set<ServerSocket> servers = new HashSet<>();
     final Set<Integer> ports = new HashSet<>();
 
     try {
       for (int i = 0; i < count; i++) {
-        var server = new ServerSocket(0);
+        final var server = new ServerSocket(0);
         servers.add(server);
         ports.add(server.getLocalPort());
       }
-    } catch (IOException e) {
+    } catch (final IOException e) {
       throw new RuntimeException(e);
     } finally {
-      for (ServerSocket server : servers) {
+      for (final ServerSocket server : servers) {
         Exceptions.swallow(server::close);
       }
     }
@@ -207,9 +211,8 @@ private static Set<Integer> getOpenPorts(int count) {
   private static String getHost() {
     try {
       return (IS_MINIKUBE ? Inet4Address.getLocalHost().getHostAddress() : "host.docker.internal");
-    } catch (UnknownHostException e) {
+    } catch (final UnknownHostException e) {
       throw new RuntimeException(e);
     }
-  };
-
+  }
 }