GH-4673 Use ConcurrentCleaner to auto-shutdown SPARQLRepository's dependent session manager on GC; add test

Author: hmottestad

core/repository/sparql/src/main/java/org/eclipse/rdf4j/repository/sparql/SPARQLRepository.java

@@ -11,10 +11,12 @@
 package org.eclipse.rdf4j.repository.sparql;
 
 import java.io.File;
+import java.lang.ref.Cleaner;
 import java.util.Collections;
 import java.util.Map;
 
 import org.apache.http.client.HttpClient;
+import org.eclipse.rdf4j.common.concurrent.locks.diagnostics.ConcurrentCleaner;
 import org.eclipse.rdf4j.http.client.HttpClientDependent;
 import org.eclipse.rdf4j.http.client.HttpClientSessionManager;
 import org.eclipse.rdf4j.http.client.SPARQLProtocolSession;
@@ -33,6 +35,8 @@
  */
 public class SPARQLRepository extends AbstractRepository implements HttpClientDependent, SessionManagerDependent {
 
+	private static final ConcurrentCleaner CLEANER = new ConcurrentCleaner();
+
 	/**
 	 * Flag indicating if quad mode is enabled in newly created {@link SPARQLConnection}s.
 	 *
@@ -49,6 +53,11 @@ public class SPARQLRepository extends AbstractRepository implements HttpClientDe
 	 */
 	private volatile SharedHttpClientSessionManager dependentClient;
 
+	/**
+	 * Cleanable registration to auto-invoke cleanup when this repository becomes unreachable.
+	 */
+	private volatile Cleaner.Cleanable cleanable;
+
 	private String username;
 
 	private String password;
@@ -93,7 +102,17 @@ public HttpClientSessionManager getHttpClientSessionManager() {
 			synchronized (this) {
 				result = client;
 				if (result == null) {
-					result = client = dependentClient = new SharedHttpClientSessionManager();
+					SharedHttpClientSessionManager created = new SharedHttpClientSessionManager();
+					result = client = dependentClient = created;
+					// Register a cleaner that shuts down the dependent client if this repository is GC'ed without
+					// explicit shutdown
+					cleanable = CLEANER.register(this, () -> {
+						try {
+							created.shutDown();
+						} catch (Throwable t) {
+							// ignore
+						}
+					});
 				}
 			}
 		}
@@ -110,6 +129,11 @@ public void setHttpClientSessionManager(HttpClientSessionManager client) {
 			if (toCloseDependentClient != null) {
 				toCloseDependentClient.shutDown();
 			}
+			Cleaner.Cleanable toClean = cleanable;
+			cleanable = null;
+			if (toClean != null) {
+				toClean.clean();
+			}
 		}
 	}
 
@@ -213,6 +237,11 @@ protected void shutDownInternal() throws RepositoryException {
 				toCloseDependentClient.shutDown();
 			}
 		} finally {
+			Cleaner.Cleanable toClean = cleanable;
+			cleanable = null;
+			if (toClean != null) {
+				toClean.clean();
+			}
 			// remove reference but do not shut down, client may be shared by
 			// other repos.
 			client = null;

core/repository/sparql/src/test/java/org/eclipse/rdf4j/repository/sparql/SPARQLRepositoryCleanerTest.java

@@ -10,13 +10,14 @@
  *******************************************************************************/
 package org.eclipse.rdf4j.repository.sparql;
 
-import static org.assertj.core.api.Assertions.fail;
+import static org.assertj.core.api.Assertions.assertThat;
 
-import java.util.concurrent.CountDownLatch;
+import java.lang.reflect.Field;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 
+import org.eclipse.rdf4j.http.client.SharedHttpClientSessionManager;
 import org.eclipse.rdf4j.repository.RepositoryConnection;
-import org.eclipse.rdf4j.repository.RepositoryException;
 import org.junit.jupiter.api.Test;
 
 /**
@@ -31,42 +32,33 @@ public class SPARQLRepositoryCleanerTest {
 
 	@Test
 	void autoShutdownOnUnreachable() throws Exception {
-		CountDownLatch shutdownInvoked = new CountDownLatch(1);
+		SPARQLRepository repo = new SPARQLRepository("http://example.org/sparql");
 
-		runRepo(shutdownInvoked);
-
-		// Encourage GC and wait briefly for cleaner to run
-		boolean observed = false;
-		for (int i = 0; i < 20 && !observed; i++) {
-			System.gc();
-			System.runFinalization();
-			observed = shutdownInvoked.await(250, TimeUnit.MILLISECONDS);
+		// Ensure dependent client is created
+		try (RepositoryConnection conn = repo.getConnection()) {
+			// no-op
 		}
 
-		if (!observed) {
-			fail("Expected shutDownInternal() to be invoked when SPARQLRepository became unreachable");
-		}
-	}
+		SharedHttpClientSessionManager mgr = (SharedHttpClientSessionManager) repo.getHttpClientSessionManager();
 
-	private static void runRepo(CountDownLatch shutdownInvoked) {
-		// Create a repository instance that signals when shutDownInternal() is invoked
-		SPARQLRepository repo = new SPARQLRepository("http://example.org/sparql") {
-			@Override
-			protected void shutDownInternal() throws RepositoryException {
-				try {
-					super.shutDownInternal();
-				} finally {
-					shutdownInvoked.countDown();
-				}
-			}
-		};
+		// Access internal executor to verify shutdown state
+		Field f = SharedHttpClientSessionManager.class.getDeclaredField("executor");
+		f.setAccessible(true);
+		ExecutorService exec = (ExecutorService) f.get(mgr);
 
-		// Exercise minimal usage without hitting the network
-		try (RepositoryConnection conn = repo.getConnection()) {
-			// no-op
+		// Drop strong reference and encourage GC to trigger Cleaner
+		repo = null;
+
+		boolean cleaned = false;
+		for (int i = 0; i < 40 && !cleaned; i++) {
+			System.gc();
+			System.runFinalization();
+			TimeUnit.MILLISECONDS.sleep(100);
+			cleaned = exec.isShutdown() || exec.isTerminated();
 		}
 
-		// Drop strong reference and ask GC to collect
-		repo = null;
+		assertThat(cleaned)
+				.as("dependent session manager executor should be shut down by Cleaner")
+				.isTrue();
 	}
 }