GH-5347 make the ShaclSail more robust during shutdown() (#5348)

Author: hmottestad

compliance/repository/src/test/java/org/eclipse/rdf4j/repository/sparql/SPARQLStoreConnectionTest.java

@@ -524,4 +524,12 @@ public void testRemoveStatementsFromContextSingleTransaction(IsolationLevel leve
 	public void testClearStatementsFromContextSingleTransaction(IsolationLevel level) throws Exception {
 		super.testClearStatementsFromContextSingleTransaction(level);
 	}
+
+	@ParameterizedTest
+	@MethodSource("parameters")
+	@Disabled("relies on pending updates being visible in own connection")
+	@Override
+	public void testRollbackAfterInterrupt(IsolationLevel level) {
+		super.testRollbackAfterInterrupt(level);
+	}
 }

core/common/iterator/src/main/java/org/eclipse/rdf4j/common/iteration/FilterIteration.java

@@ -102,17 +102,18 @@ private void findNextElement() {
 				}
 			}
 
-			// We know that nextElement has to be null and that wrappedIter.hasNext() must be true, based on the code
-			// above. To be sure that these invariants don't change we also assert them below.
-			assert nextElement == null && wrappedIter.hasNext();
-
 			do {
 				E result;
 				if (Thread.currentThread().isInterrupted()) {
 					close();
 					return;
 				}
 				try {
+					// We know that nextElement has to be null and that wrappedIter.hasNext() must be true, based on the
+					// code
+					// above. To be sure that these invariants don't change we also assert them below.
+					assert nextElement == null && wrappedIter.hasNext();
+
 					result = wrappedIter.next();
 				} catch (NoSuchElementException e) {
 					close();

core/common/iterator/src/main/java/org/eclipse/rdf4j/common/iteration/LookAheadIteration.java

@@ -13,12 +13,16 @@
 
 import java.util.NoSuchElementException;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 /**
  * An Iteration that looks one element ahead, if necessary, to handle calls to {@link #hasNext}. This is a convenient
  * super class for Iterations that have no easy way to tell if there are any more results, but still should implement
  * the <var>java.util.Iteration</var> interface.
  */
 public abstract class LookAheadIteration<E> extends AbstractCloseableIteration<E> {
+	private static final Logger log = LoggerFactory.getLogger(LookAheadIteration.class);
 
 	/*-----------*
 	 * Variables *
@@ -50,7 +54,14 @@ public final boolean hasNext() {
 			return false;
 		}
 
-		return lookAhead() != null;
+		try {
+			return lookAhead() != null;
+		} catch (NoSuchElementException logged) {
+			// The lookAhead() method shouldn't throw a NoSuchElementException since it should return null when there
+			// are no more elements.
+			log.trace("LookAheadIteration threw NoSuchElementException:", logged);
+			return false;
+		}
 	}
 
 	@Override

core/queryalgebra/evaluation/src/main/java/org/eclipse/rdf4j/query/algebra/evaluation/impl/DefaultEvaluationStrategy.java

@@ -217,7 +217,7 @@ protected static CloseableIteration<BindingSet> evaluate(TupleFunction func,
 			throws QueryEvaluationException {
 		final CloseableIteration<? extends List<? extends Value>> iter = func
 				.evaluate(valueFactory, argValues);
-		return new LookAheadIteration<BindingSet>() {
+		return new LookAheadIteration<>() {
 
 			@Override
 			public BindingSet getNextElement() throws QueryEvaluationException {
@@ -712,7 +712,7 @@ protected QueryEvaluationStep prepare(Distinct node, QueryEvaluationContext cont
 		final CollectionFactory cf = this.getCollectionFactory().get();
 		return bindings -> {
 			final CloseableIteration<BindingSet> evaluate = child.evaluate(bindings);
-			return new DistinctIteration<BindingSet>(evaluate, cf.createSetOfBindingSets()) {
+			return new DistinctIteration<>(evaluate, cf.createSetOfBindingSets()) {
 
 				@Override
 				protected void handleClose() throws QueryEvaluationException {

core/repository/sail/src/main/java/org/eclipse/rdf4j/repository/sail/SailRepositoryConnection.java

@@ -231,10 +231,17 @@ public void commit() throws RepositoryException {
 
 	@Override
 	public void rollback() throws RepositoryException {
+		// We still want to attempt to rollback the transaction even if the thread is interrupted. We restore the
+		// interrupted status when we are done.
+		boolean interrupted = Thread.interrupted();
 		try {
 			sailConnection.rollback();
 		} catch (SailException e) {
 			throw new RepositoryException(e);
+		} finally {
+			if (interrupted) {
+				Thread.currentThread().interrupt();
+			}
 		}
 	}
 

core/sail/api/src/main/java/org/eclipse/rdf4j/common/concurrent/locks/ExclusiveReentrantLockManager.java

@@ -45,14 +45,22 @@ public ExclusiveReentrantLockManager(boolean trackLocks) {
 	}
 
 	public ExclusiveReentrantLockManager(boolean trackLocks, int collectionFrequency) {
+		this(trackLocks, collectionFrequency, "ExclusiveReentrantLockManager");
+	}
+
+	public ExclusiveReentrantLockManager(String lockName) {
+		this(false, LockMonitoring.INITIAL_WAIT_TO_COLLECT, lockName);
+	}
+
+	public ExclusiveReentrantLockManager(boolean trackLocks, int collectionFrequency, String lockName) {
 
 		this.waitToCollect = collectionFrequency;
 
 		if (trackLocks || Properties.lockTrackingEnabled()) {
 
 			lockMonitoring = new LockTracking(
 					true,
-					"ExclusiveReentrantLockManager",
+					lockName,
 					LoggerFactory.getLogger(this.getClass()),
 					waitToCollect,
 					Lock.ExtendedSupplier.wrap(this::getExclusiveLockInner, this::tryExclusiveLockInner)
@@ -61,7 +69,7 @@ public ExclusiveReentrantLockManager(boolean trackLocks, int collectionFrequency
 		} else {
 			lockMonitoring = new LockCleaner(
 					false,
-					"ExclusiveReentrantLockManager",
+					lockName,
 					LoggerFactory.getLogger(this.getClass()),
 					Lock.ExtendedSupplier.wrap(this::getExclusiveLockInner, this::tryExclusiveLockInner)
 			);
@@ -101,6 +109,16 @@ private Lock getExclusiveLockInner() throws InterruptedException {
 						return lock;
 					} else {
 						lockMonitoring.runCleanup();
+						lock = tryExclusiveLockInner();
+						if (lock != null) {
+							return lock;
+						}
+
+						Thread thread = owner.get();
+						if (thread != null && !thread.isAlive()) {
+							logger.debug("Lock is held by dead thread: " + thread);
+						}
+
 						owner.wait(waitToCollect);
 					}
 				} while (true);

core/sail/api/src/main/java/org/eclipse/rdf4j/sail/InterruptedSailException.java

@@ -0,0 +1,35 @@
+/*******************************************************************************
+ * Copyright (c) 2025 Eclipse RDF4J contributors.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Distribution License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * SPDX-License-Identifier: BSD-3-Clause
+ ******************************************************************************/
+
+package org.eclipse.rdf4j.sail;
+
+public class InterruptedSailException extends SailException {
+
+	public InterruptedSailException(String s) {
+		super(s);
+		assert Thread.currentThread().isInterrupted();
+	}
+
+	public InterruptedSailException() {
+		super();
+		assert Thread.currentThread().isInterrupted();
+	}
+
+	public InterruptedSailException(Throwable t) {
+		super(t);
+		assert Thread.currentThread().isInterrupted();
+	}
+
+	public InterruptedSailException(String msg, Throwable t) {
+		super(msg, t);
+		assert Thread.currentThread().isInterrupted();
+	}
+}

core/sail/api/src/main/java/org/eclipse/rdf4j/sail/helpers/AbstractSail.java

@@ -40,7 +40,7 @@
 public abstract class AbstractSail implements Sail {
 
 	/**
-	 * Default connection timeout on shutdown: 20,000 milliseconds.
+	 * Default connection timeout on shutdown: 20,000 milliseconds (20 seconds).
 	 */
 	protected final static long DEFAULT_CONNECTION_TIMEOUT = 20000L;
 
@@ -227,17 +227,19 @@ public void shutDown() throws SailException {
 				try {
 					SailConnection con = entry.getKey();
 
-					if (con instanceof AbstractSailConnection) {
-						AbstractSailConnection sailCon = (AbstractSailConnection) con;
-						Thread owner = sailCon.getOwner();
-						if (owner != Thread.currentThread()) {
-							owner.interrupt();
-							// wait up to 1 second for the owner thread to die
-							owner.join(1000);
-							if (owner.isAlive()) {
-								logger.error(
-										"Closing active connection due to shut down and interrupted the owning thread of the connection {} but thread is still alive after 1000 ms!",
-										owner);
+					if (con.isOpen()) {
+						if (con instanceof AbstractSailConnection) {
+							AbstractSailConnection sailCon = (AbstractSailConnection) con;
+							Thread owner = sailCon.getOwner();
+							if (owner != Thread.currentThread()) {
+								owner.interrupt();
+								// wait up to 1 second for the owner thread to die
+								owner.join(1000);
+								if (owner.isAlive() && con.isOpen()) {
+									logger.error(
+											"Closing active connection due to shut down and interrupted the owning thread of the connection {} but thread is still alive after 1000 ms!",
+											owner);
+								}
 							}
 						}
 					}
@@ -257,6 +259,9 @@ public void shutDown() throws SailException {
 			// Forcefully close any connections that are still open
 			for (Map.Entry<SailConnection, Throwable> entry : activeConnectionsCopy.entrySet()) {
 				SailConnection con = entry.getKey();
+				if (!con.isOpen()) {
+					continue;
+				}
 				Throwable stackTrace = entry.getValue();
 
 				if (stackTrace == null) {