Shut down leaked executors and Hibernate SessionFactory on close

src/main/java/fi/csc/chipster/filebroker/FileBroker.java

@@ -140,11 +140,26 @@ public static void main(String[] args) throws Exception {
 
 	public void close() {
 		RestUtils.shutdown("file-broker-admin", adminServer);
+
 		try {
-			httpServer.stop();
-			authService.close();
+			if (httpServer != null) {
+				httpServer.stop();
+			}
 		} catch (Exception e) {
 			logger.warn("failed to stop the file-broker", e);
 		}
+
+		try {
+			if (authService != null) {
+				authService.close();
+			}
+		} catch (Exception e) {
+			logger.warn("failed to stop the file-broker auth client", e);
+		}
+
+		// after httpServer.stop(), because requests submit tasks to its executor
+		if (storageDiscovery != null) {
+			storageDiscovery.close();
+		}
 	}
 }

src/main/java/fi/csc/chipster/filestorage/client/FileStorageDiscovery.java

@@ -280,4 +280,8 @@ public Map<String, FileStorage> getStorages() {
 			return storages;
 		}
 	}
+
+	public void close() {
+		updateExecutor.shutdown();
+	}
 }

src/main/java/fi/csc/chipster/sessionworker/ZipSessionServlet.java

@@ -17,6 +17,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.zip.ZipEntry;
@@ -112,6 +113,11 @@ public ZipSessionServlet(ServiceLocatorClient serviceLocator) {
 		this.executor = Executors.newCachedThreadPool();
 	}
 
+	@Override
+	public void destroy() {
+		executor.shutdown();
+	}
+
 	private void packageSession(HttpServletResponse response, StaticCredentials credentials, UUID sessionId)
 			throws IOException {
 
@@ -123,6 +129,9 @@ private void packageSession(HttpServletResponse response, StaticCredentials cred
 
 		ArrayList<InputStreamEntry> entries = new ArrayList<>();
 
+		CountDownLatch latch = new CountDownLatch(1);
+		PipedInputStream in = null;
+
 		try {
 			Session session = sessionDb.getSession(sessionId);
 
@@ -137,8 +146,6 @@ private void packageSession(HttpServletResponse response, StaticCredentials cred
 
 			ArrayList<String> errors = new ArrayList<>();
 
-			CountDownLatch latch = new CountDownLatch(1);
-
 			OutputStream respoonseOutput = response.getOutputStream();
 
 			keepAliveWithSpaces(respoonseOutput, latch);
@@ -181,20 +188,30 @@ private void packageSession(HttpServletResponse response, StaticCredentials cred
 			}
 
 			OutputStream output2 = new PipedOutputStream();
-			PipedInputStream in = new PipedInputStream((PipedOutputStream) output2);
+			in = new PipedInputStream((PipedOutputStream) output2);
 
 			if (errors.isEmpty()) {
 				// start creating the zip stream in background thread (may complete before all
 				// data is uploaded)
-				executor.submit(() -> {
-					try {
-						streamZip(entries, output2);
-					} catch (IOException e) {
-						logger.error("failed to package zip session", e);
-						errors.add("failed to package zip session: " + e.getMessage());
-					}
-				});
+				try {
+					executor.submit(() -> {
+						try {
+							streamZip(entries, output2);
+						} catch (IOException e) {
+							logger.error("failed to package zip session", e);
+							errors.add("failed to package zip session: " + e.getMessage());
+						}
+					});
+				} catch (RejectedExecutionException e) {
+					// destroy() has shut down the executor, i.e. this session-worker is stopping
+					logger.error("failed to start zip packaging, session-worker is stopping", e);
+					errors.add("failed to package zip session: session-worker is stopping");
+				}
+			}
 
+			// skipped when the zip thread wasn't started, because then the upload would
+			// block forever waiting for the zip stream
+			if (errors.isEmpty()) {
 				// upload in the zip stream in this thread, so that we send response to this
 				// servlet request only after the upload has really completed
 				try {
@@ -227,12 +244,23 @@ private void packageSession(HttpServletResponse response, StaticCredentials cred
 
 			logger.info("response: " + RestUtils.asJson(json, true));
 
+			// stop the keep-alive before writing the json. This doesn't wait for a write
+			// that is already in progress, but prevents all the following ones.
 			latch.countDown();
 			respoonseOutput.write(RestUtils.asJson(json).getBytes());
 			respoonseOutput.close();
 
 		} catch (RestException e) {
 			throw ServletUtils.extractRestException(e);
+		} finally {
+			// stop the keep-alive thread also when something unexpected was thrown.
+			// Otherwise it would keep writing spaces to the client forever.
+			latch.countDown();
+
+			// If the upload failed or was skipped, the zip thread may still be writing to
+			// the pipe. Its writes would block forever, because the reader thread stays
+			// alive in the Jetty pool. Closing the read end makes them fail instead.
+			IOUtils.closeQuietly(in);
 		}
 	}
 

src/test/java/fi/csc/chipster/auth/OidcResourceTest.java

@@ -117,6 +117,7 @@ public static void setUp() throws Exception {
 
 	@AfterAll
 	public static void tearDown() throws Exception {
+		hibernate.getSessionFactory().close();
 		launcher.stop();
 	}