[MINOR][CI] Fix leaked threads hanging Java test forks (#2488)

Several Java test suites (most visibly **.component.c** and data.misc/lineage) intermittently ran until the GitHub Actions job timeout even though the tests themselves had completed. The cause was leaked non-daemon threads keeping the surefire fork JVM alive, so the fork never exited and the job stalled until cancelled. There were two sources: in-JVM federated workers (FederatedWorker's Netty event loops and the test-side worker wrapper threads were non-daemon), and CommonThreadPool's fallback pools — when called off the main thread, it returned Executors.newFixedThreadPool/newCachedThreadPool, which default to non-daemon threads, while only the ForkJoinPool-backed variants were already daemon.

This PR makes those threads daemon at the source: FederatedWorker now creates its Netty event-loop groups with a daemon thread factory, and CommonThreadPool routes its fixed/cached fallbacks through one too, so daemon behavior is uniform across all pool variants. On the test side, AutomatedTestBase marks spawned worker threads as daemon, TestUtils.shutdownThread bounds its join (30s, warns on stragglers, restores the interrupt flag), and the lineage tests (LineageFedReuseAlg, FedFullReuseTest, FedUDFReuseTest) now shut workers down in a finally block so failures no longer leak workers (the large line counts there are just reindentation from the try/finally wrap). The javaTests.yml job cap stays at 30 minutes, with a comment documenting why it sits above the 600s per-fork surefire timeout, which remains the backstop for genuine hangs.

Author: Baunsgaard

.github/workflows/javaTests.yml

@@ -50,6 +50,8 @@ concurrency:
 jobs:
   java_tests:
     runs-on: ubuntu-24.04
+    # Job cap kept above the per-fork surefire timeout (test-forkedProcessTimeout,
+    # 600s) so surefire can kill a hung fork before GitHub Actions cancels the job.
     timeout-minutes: 30
     strategy:
       fail-fast: false

dev/release/src/test/java/org/apache/sysds/validation/Utility.java

@@ -185,6 +185,7 @@ public static int runCommand(String [] command, String strCurDir, String strOutp
 		try {
 			exitValue = process.waitFor();
 		} catch (InterruptedException ie) {
+			Thread.currentThread().interrupt();
 			debugPrint(Constants.DEBUG_ERROR, "Program interrunpted: " + ie);
 		}
 		debugPrint(Constants.DEBUG_CODE, "Program '" + String.join(" ", command) + "' exited with exit status " + exitValue, strOutputFile);

src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorker.java

@@ -65,6 +65,7 @@
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
 import io.netty.handler.ssl.util.SelfSignedCertificate;
+import io.netty.util.concurrent.DefaultThreadFactory;
 
 @SuppressWarnings("deprecation")
 public class FederatedWorker {
@@ -99,9 +100,11 @@ private void run() {
 		LOG.info("Setting up Federated Worker on port " + _port);
 		int par_conn = ConfigurationManager.getDMLConfig().getIntValue(DMLConfig.FEDERATED_PAR_CONN);
 		final int EVENT_LOOP_THREADS = (par_conn > 0) ? par_conn : InfrastructureAnalyzer.getLocalParallelism();
-		NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
+		// Daemon event loops so a leaked in-JVM (test) worker cannot block JVM exit.
+		NioEventLoopGroup bossGroup = new NioEventLoopGroup(1,
+			new DefaultThreadFactory("fed-worker-boss", true));
 		ThreadPoolExecutor workerTPE = new ThreadPoolExecutor(1, Integer.MAX_VALUE, 10, TimeUnit.SECONDS,
-			new SynchronousQueue<Runnable>(true));
+			new SynchronousQueue<Runnable>(true), new DefaultThreadFactory("fed-worker-pool", true));
 		NioEventLoopGroup workerGroup = new NioEventLoopGroup(EVENT_LOOP_THREADS, workerTPE);
 
 		final boolean ssl = ConfigurationManager.isFederatedSSL();

src/main/java/org/apache/sysds/runtime/ooc/cache/OOCMatrixIOHandler.java

@@ -148,7 +148,8 @@ public void shutdown() {
 					_q[i].close();
 				}
 			}
-			catch(InterruptedException ignored) {
+			catch(InterruptedException e) {
+				Thread.currentThread().interrupt();
 			}
 		}
 		_writeExec.getQueue().clear();
@@ -174,7 +175,8 @@ public CompletableFuture<Void> scheduleEviction(BlockEntry block) {
 			int i = (int)(q % WRITER_SIZE);
 			_q[i].enqueueIfOpen(new Tuple2<>(block, future));
 		}
-		catch(InterruptedException ignored) {
+		catch(InterruptedException e) {
+			Thread.currentThread().interrupt();
 		}
 
 		return future;

src/main/java/org/apache/sysds/runtime/util/CommonThreadPool.java

@@ -29,6 +29,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.Future;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
@@ -141,11 +142,26 @@ else if(mainThread || threadName.contains("PARFOR") || threadName.contains("FedE
 				incorrectPoolUse = true;
 			}
 
-			return Executors.newFixedThreadPool(k);
+			return Executors.newFixedThreadPool(k, daemonThreadFactory());
 
 		}
 	}
 
+	/**
+	 * Thread factory that produces daemon threads. The ForkJoinPool-backed pools already use daemon
+	 * threads; the fallback {@link Executors#newFixedThreadPool} and {@link Executors#newCachedThreadPool}
+	 * pools default to non-daemon threads, which can keep the JVM (e.g. a surefire test fork) alive
+	 * if a caller forgets to shut the pool down. Making them daemon keeps that behavior uniform.
+	 */
+	private static ThreadFactory daemonThreadFactory() {
+		final ThreadFactory base = Executors.defaultThreadFactory();
+		return r -> {
+			Thread t = base.newThread(r);
+			t.setDaemon(true);
+			return t;
+		};
+	}
+
 	/**
 	 * Invoke the collection of tasks and shutdown the pool upon job termination.
 	 * 
@@ -180,7 +196,7 @@ public synchronized static ExecutorService getDynamicPool() {
 			// It is guaranteed not to be shut down because of the synchronized barrier
 			return asyncPool;
 		else {
-			asyncPool = Executors.newCachedThreadPool();
+			asyncPool = Executors.newCachedThreadPool(daemonThreadFactory());
 			return asyncPool;
 		}
 	}

src/test/java/org/apache/sysds/performance/generators/GenMatrices.java

@@ -72,7 +72,7 @@ public void generate(int N) throws InterruptedException {
 				}
 			}
 			catch(InterruptedException e) {
-				e.printStackTrace();
+				Thread.currentThread().interrupt();
 			}
 		});
 	}

src/test/java/org/apache/sysds/test/AutomatedTestBase.java

@@ -1939,6 +1939,9 @@ private static Thread spawnLocalFedWorkerThread(int port, String[] otherArgs) {
 				LOG.error("Exception in startup of federated worker", e);
 			}
 		});
+		// Daemon so a worker left running by a failed/forgetful test cannot keep the
+		// surefire fork JVM alive and stall CI until the job-level timeout.
+		t.setDaemon(true);
 		t.start();
 		return t;
 	}
@@ -1979,6 +1982,9 @@ public static Thread startLocalFedWorkerWithArgs(String[] args) {
 				LOG.error("Exception in startup of federated worker on port " + port, e);
 			}
 		});
+		// Daemon so a worker left running by a failed/forgetful test cannot keep the
+		// surefire fork JVM alive and stall CI until the job-level timeout.
+		t.setDaemon(true);
 		t.start();
 		FederatedWorkerUtils.waitForWorker(t, port, FED_WORKER_WAIT);
 		return t;

src/test/java/org/apache/sysds/test/TestUtils.java

@@ -3489,15 +3489,23 @@ public static void shutdownThreads(Process... ts) {
 		}
 	}
 
+	/** Upper bound (ms) on how long {@link #shutdownThread(Thread)} waits for a worker to stop. */
+	private static final long THREAD_SHUTDOWN_JOIN_MS = 30_000;
+
 	public static void shutdownThread(Thread t) {
 		// kill the worker
 		if( t != null ) {
 			t.interrupt();
 			try {
-				t.join();
+				// Bounded join: workers are daemon threads, so even if one ignores the interrupt
+				// we must not block cleanup (and the JVM) indefinitely waiting for it.
+				t.join(THREAD_SHUTDOWN_JOIN_MS);
+				if( t.isAlive() )
+					LOG.warn("Federated worker thread " + t.getName()
+						+ " did not stop within " + THREAD_SHUTDOWN_JOIN_MS + "ms; leaving it as a daemon.");
 			}
 			catch (InterruptedException e) {
-				e.printStackTrace();
+				Thread.currentThread().interrupt();
 			}
 		}
 	}
@@ -3514,7 +3522,8 @@ public static void shutdownThread(Process t) {
 					forciblyDestroyed.waitFor(); // Wait until it's definitely terminated
 				}
 			} catch (InterruptedException e) {
-				e.printStackTrace();
+				LOG.warn("Interrupted while shutting down federated worker process", e);
+				Thread.currentThread().interrupt();
 			}
 		}
 	}

src/test/java/org/apache/sysds/test/functions/federated/monitoring/FederatedBackendPerformanceTest.java

@@ -91,8 +91,11 @@ public void testBackendPerformance() throws InterruptedException {
 			taskFutures.forEach(res -> {
 				try {
 					Assert.assertEquals("Stats parsed correctly", res.get().statusCode(), 200);
-				} catch (InterruptedException | ExecutionException e) {
-					e.printStackTrace();
+				} catch (InterruptedException e) {
+					Thread.currentThread().interrupt();
+					Assert.fail("Interrupted while fetching statistics: " + e.getMessage());
+				} catch (ExecutionException e) {
+					Assert.fail("Failed to fetch statistics: " + e.getMessage());
 				}
 			});
 

src/test/java/org/apache/sysds/test/functions/federated/primitives/part5/FederatedMatrixScalarOperationsTest.java

@@ -209,7 +209,7 @@ private void runGenericTest(String dmlFile, int scalar) {
 			compareResults();
 		}
 		catch(InterruptedException e) {
-			e.printStackTrace();
+			Thread.currentThread().interrupt();
 			assert (false);
 		}
 		finally {