Merge main into develop (#4721)

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

@@ -34,11 +34,13 @@ public class Iterations {
 	 * @return a List containing all elements obtained from the specified iteration.
 	 */
 	public static <E, X extends Exception> List<E> asList(CloseableIteration<? extends E> iter) {
-		// stream.collect is slightly slower than addAll for lists
-		List<E> list = new ArrayList<>();
+		try (iter) {
+			// stream.collect is slightly slower than addAll for lists
+			List<E> list = new ArrayList<>();
 
-		// addAll closes the iteration
-		return addAll(iter, list);
+			// addAll closes the iteration
+			return addAll(iter, list);
+		}
 	}
 
 	/**
@@ -107,9 +109,11 @@ public static <T> Stream<T> stream(CloseableIteration<T> iteration) {
 	 * @return A String representation of the objects provided by the supplied iteration.
 	 */
 	public static <X extends Exception> String toString(CloseableIteration<?> iteration, String separator) {
-		StringBuilder sb = new StringBuilder();
-		toString(iteration, separator, sb);
-		return sb.toString();
+		try (iteration) {
+			StringBuilder sb = new StringBuilder();
+			toString(iteration, separator, sb);
+			return sb.toString();
+		}
 	}
 
 	/**
@@ -124,11 +128,13 @@ public static <X extends Exception> String toString(CloseableIteration<?> iterat
 	public static <X extends Exception> void toString(CloseableIteration<?> iteration, String separator,
 			StringBuilder sb)
 			throws X {
-		while (iteration.hasNext()) {
-			sb.append(iteration.next());
+		try (iteration) {
+			while (iteration.hasNext()) {
+				sb.append(iteration.next());
 
-			if (iteration.hasNext()) {
-				sb.append(separator);
+				if (iteration.hasNext()) {
+					sb.append(separator);
+				}
 			}
 		}
 
@@ -149,4 +155,5 @@ public static <E, X extends Exception> Set<E> asSet(CloseableIteration<? extends
 		}
 		return set;
 	}
+
 }

core/sail/shacl/src/main/java/org/eclipse/rdf4j/sail/shacl/results/ValidationResult.java

@@ -38,6 +38,8 @@
 import org.eclipse.rdf4j.sail.shacl.ast.constraintcomponents.ConstraintComponent;
 import org.eclipse.rdf4j.sail.shacl.ast.constraintcomponents.SparqlConstraintComponent;
 import org.eclipse.rdf4j.sail.shacl.ast.paths.Path;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * The ValidationResult represents the results from a SHACL validation in an easy-to-use Java API.
@@ -48,6 +50,8 @@
 @Deprecated
 public class ValidationResult {
 
+	private static final Logger logger = LoggerFactory.getLogger(ValidationResult.class);
+
 	private Resource id;
 	private final Optional<Value> value;
 	private final Shape shape;
@@ -73,7 +77,18 @@ public ValidationResult(Value focusNode, Value value, Shape shape,
 
 		if (sourceConstraintComponent.producesValidationResultValue()) {
 			assert value != null;
-			this.value = Optional.of(value);
+
+			// value could be null if assertions are disabled
+			// noinspection ConstantValue
+			if (value == null) {
+				logger.error(
+						"Source constraint component {} was expected to produce a value, but value is null! Shape: {}",
+						sourceConstraintComponent, shape);
+			}
+
+			// value could be null if assertions are disabled
+			// noinspection OptionalOfNullableMisuse
+			this.value = Optional.ofNullable(value);
 		} else {
 			assert scope != ConstraintComponent.Scope.propertyShape || value == null;
 			this.value = Optional.empty();

tools/federation/src/main/java/org/eclipse/rdf4j/federated/evaluation/concurrent/ParallelServiceExecutor.java

@@ -166,7 +166,9 @@ protected CloseableIteration<BindingSet> performTaskInternal() throws Exception
 			// This is basically required to avoid processing background tuple
 			// request (i.e. HTTP slots) in the correct order.
 			Service service1 = service.getService();
-			return new CollectionIteration<>(Iterations.asList(strategy.precompile(service1).evaluate(bindings)));
+			try (CloseableIteration<BindingSet> evaluate = strategy.precompile(service1).evaluate(bindings)) {
+				return new CollectionIteration<>(Iterations.asList(evaluate));
+			}
 		}
 
 		@Override

tools/federation/src/main/java/org/eclipse/rdf4j/federated/evaluation/concurrent/ParallelTaskBase.java

@@ -107,7 +107,23 @@ public void close() {
 						logger.debug("Attempting to cancel task {}", this);
 						boolean successfullyCanceled = scheduledFuture.cancel(true);
 						if (!successfullyCanceled) {
-							logger.debug("Task {} could not be cancelled properly.", this);
+							logger.debug("Task {} could not be cancelled properly. Maybe it has already completed.",
+									this);
+						}
+
+						int timeout = 100;
+						for (int i = 0; i < timeout && !scheduledFuture.isDone(); i++) {
+							try {
+								Thread.sleep(1);
+							} catch (InterruptedException e) {
+								Thread.currentThread().interrupt();
+								break;
+							}
+						}
+
+						if (!scheduledFuture.isDone()) {
+							logger.error("Timeout while waiting for task {} to terminate after it was cancelled.",
+									this);
 						}
 					}
 				}

tools/federation/src/main/java/org/eclipse/rdf4j/federated/evaluation/iterator/CloseDependentConnectionIteration.java

@@ -38,6 +38,12 @@ public CloseDependentConnectionIteration(CloseableIteration<T> inner,
 	@Override
 	public boolean hasNext() throws QueryEvaluationException {
 		try {
+			if (Thread.interrupted()) {
+				Thread.currentThread().interrupt();
+				close();
+				return false;
+			}
+
 			boolean res = inner.hasNext();
 			if (!res) {
 				close();

tools/federation/src/main/java/org/eclipse/rdf4j/federated/evaluation/iterator/ConsumingIteration.java

@@ -55,21 +55,32 @@ public ConsumingIteration(CloseableIteration<BindingSet> iter, int max)
 		try {
 			while (consumed.size() < max && iter.hasNext()) {
 				consumed.add(iter.next());
+				if (Thread.interrupted()) {
+					Thread.currentThread().interrupt();
+					close();
+					return;
+				}
 			}
 			if (!iter.hasNext()) {
 				iter.close();
 			}
 			completed = true;
 		} finally {
 			if (!completed) {
-				iter.close();
+				close();
 			}
 		}
 
 	}
 
 	@Override
 	public boolean hasNext() throws QueryEvaluationException {
+		if (Thread.interrupted()) {
+			Thread.currentThread().interrupt();
+			close();
+			return false;
+		}
+
 		return currentIndex < consumed.size() || innerIter.hasNext();
 	}
 

tools/federation/src/main/java/org/eclipse/rdf4j/federated/evaluation/join/ParallelServiceJoinTask.java

@@ -48,7 +48,9 @@ protected CloseableIteration<BindingSet> performTaskInternal() throws Exception
 		// Note: in order two avoid deadlocks we consume the SERVICE result.
 		// This is basically required to avoid processing background tuple
 		// request (i.e. HTTP slots) in the correct order.
-		return new CollectionIteration<>(Iterations.asList(strategy.evaluateService(expr, bindings)));
+		try (var iter = strategy.evaluateService(expr, bindings)) {
+			return new CollectionIteration<>(Iterations.asList(iter));
+		}
 	}
 
 	@Override

tools/federation/src/main/java/org/eclipse/rdf4j/federated/optimizer/SourceSelection.java

@@ -320,8 +320,8 @@ public ParallelCheckTask(Endpoint endpoint, StatementPattern stmt, QueryInfo que
 		protected CloseableIteration<BindingSet> performTaskInternal() throws Exception {
 			try {
 				TripleSource t = endpoint.getTripleSource();
-				boolean hasResults;
-				hasResults = t.hasStatements(stmt, EmptyBindingSet.getInstance(), queryInfo, queryInfo.getDataset());
+				boolean hasResults = t.hasStatements(stmt, EmptyBindingSet.getInstance(), queryInfo,
+						queryInfo.getDataset());
 
 				SourceSelection sourceSelection = control.sourceSelection;
 				sourceSelection.cache.updateInformation(new SubQuery(stmt, queryInfo.getDataset()), endpoint,