GH-5787 Fix SHACL shutdown interrupt race (#5788)

Author: hmottestad

core/sail/shacl/src/main/java/org/eclipse/rdf4j/sail/shacl/ConnectionHelper.java

@@ -11,12 +11,15 @@
 
 package org.eclipse.rdf4j.sail.shacl;
 
+import java.util.NoSuchElementException;
+
 import org.eclipse.rdf4j.common.iteration.CloseableIteration;
 import org.eclipse.rdf4j.model.IRI;
 import org.eclipse.rdf4j.model.Resource;
 import org.eclipse.rdf4j.model.Statement;
 import org.eclipse.rdf4j.model.Value;
 import org.eclipse.rdf4j.repository.RepositoryResult;
+import org.eclipse.rdf4j.sail.InterruptedSailException;
 import org.eclipse.rdf4j.sail.SailConnection;
 import org.eclipse.rdf4j.sail.SailException;
 
@@ -55,8 +58,23 @@ static void transferStatements(SailConnection from, TransferStatement transfer)
 		try (CloseableIteration<? extends Statement> statements = from
 				.getStatements(null, null, null, false)) {
 
-			while (statements.hasNext()) {
-				Statement next = statements.next();
+			while (true) {
+				throwIfInterrupted();
+				boolean hasNext = statements.hasNext();
+				throwIfInterrupted();
+				if (!hasNext) {
+					break;
+				}
+
+				Statement next;
+				try {
+					next = statements.next();
+				} catch (NoSuchElementException e) {
+					if (Thread.currentThread().isInterrupted()) {
+						throw new InterruptedSailException("Thread was interrupted while transferring statements.", e);
+					}
+					throw e;
+				}
 
 				transfer.transfer(next.getSubject(), next.getPredicate(),
 						next.getObject(), next.getContext());
@@ -66,6 +84,12 @@ static void transferStatements(SailConnection from, TransferStatement transfer)
 
 	}
 
+	private static void throwIfInterrupted() {
+		if (Thread.currentThread().isInterrupted()) {
+			throw new InterruptedSailException("Thread was interrupted while transferring statements.");
+		}
+	}
+
 	static boolean isEmpty(SailConnection connection) {
 		return !connection.hasStatement(null, null, null, false);
 	}

core/sail/shacl/src/test/java/org/eclipse/rdf4j/sail/shacl/ConnectionHelperTest.java

@@ -0,0 +1,78 @@
+/*******************************************************************************
+ * Copyright (c) 2026 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
+ *******************************************************************************/
+// Some portions generated by Codex
+
+package org.eclipse.rdf4j.sail.shacl;
+
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.NoSuchElementException;
+
+import org.eclipse.rdf4j.common.iteration.CloseableIteration;
+import org.eclipse.rdf4j.model.Statement;
+import org.eclipse.rdf4j.sail.InterruptedSailException;
+import org.eclipse.rdf4j.sail.SailConnection;
+import org.junit.jupiter.api.Test;
+
+class ConnectionHelperTest {
+
+	@Test
+	void transferStatementsConvertsInterruptedIterationToInterruptedSailException() {
+		SailConnection connection = mock(SailConnection.class);
+		@SuppressWarnings("unchecked")
+		CloseableIteration<? extends Statement> statements = mock(CloseableIteration.class);
+		doReturn(statements).when(connection).getStatements(isNull(), isNull(), isNull(), anyBoolean());
+		when(statements.hasNext()).thenReturn(true);
+		when(statements.next()).thenAnswer(invocation -> {
+			Thread.currentThread().interrupt();
+			throw new NoSuchElementException("The iteration has been interrupted.");
+		});
+
+		try {
+			assertThatExceptionOfType(InterruptedSailException.class)
+					.isThrownBy(() -> ConnectionHelper.transferStatements(connection, (subject, predicate, object,
+							context) -> {
+						throw new AssertionError("No statement should be transferred after interrupt");
+					}))
+					.withCauseInstanceOf(NoSuchElementException.class);
+			verify(statements).close();
+		} finally {
+			Thread.interrupted();
+		}
+	}
+
+	@Test
+	void transferStatementsKeepsNonInterruptNoSuchElementException() {
+		SailConnection connection = mock(SailConnection.class);
+		@SuppressWarnings("unchecked")
+		CloseableIteration<? extends Statement> statements = mock(CloseableIteration.class);
+		NoSuchElementException exception = new NoSuchElementException("empty");
+		doReturn(statements).when(connection).getStatements(isNull(), isNull(), isNull(), anyBoolean());
+		when(statements.hasNext()).thenReturn(true);
+		when(statements.next()).thenThrow(exception);
+
+		assertThatExceptionOfType(NoSuchElementException.class)
+				.isThrownBy(() -> ConnectionHelper.transferStatements(connection, (subject, predicate, object,
+						context) -> {
+					throw new AssertionError("No statement should be transferred");
+				}))
+				.isSameAs(exception);
+		verify(statements).close();
+		verify(statements, never()).remove();
+	}
+}