GH-5333 allow exporting of data directly to the http output stream, but attempt to detect any issues first by exporting up to 1024 statements to a dummy stream and throwing an exception before we set the status.

Author: hmottestad

docker/Dockerfile-jetty

@@ -11,12 +11,12 @@ WORKDIR /tmp
 RUN unzip -q /tmp/rdf4j.zip
 
 # Final workbench
-FROM jetty:9-jre17-eclipse-temurin 
+FROM jetty:9-jre17-eclipse-temurin
 LABEL org.opencontainers.image.authors="Bart Hanssens (bart.hanssens@bosa.fgov.be)"
 
 USER root
 
-ENV JAVA_OPTIONS="-Dorg.eclipse.rdf4j.appdata.basedir=/var/rdf4j -Dorg.eclipse.rdf4j.rio.jsonld_secure_mode=false"
+ENV JAVA_OPTIONS="-Xmx2g -Dorg.eclipse.rdf4j.appdata.basedir=/var/rdf4j -Dorg.eclipse.rdf4j.rio.jsonld_secure_mode=false"
 ENV JETTY_MODULES="server,bytebufferpool,threadpool,security,servlet,webapp,ext,plus,deploy,annotations,http,jsp,jstl"
 
 COPY --from=temp /tmp/eclipse-rdf4j*/war/*.war /var/lib/jetty/webapps/

docker/run.sh

@@ -2,6 +2,8 @@
 set -e
 ./build.sh
 
+export JAVA_OPTIONS="-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=*:5005"
+
 echo "Starting the docker container for ${APP_SERVER}"
 docker compose up --force-recreate -d
 

tools/server-spring/src/main/java/org/eclipse/rdf4j/http/server/repository/statements/ExportStatementsView.java

@@ -12,10 +12,11 @@
 
 import static javax.servlet.http.HttpServletResponse.SC_OK;
 
-import java.io.ByteArrayOutputStream;
+import java.io.IOException;
 import java.io.OutputStream;
 import java.nio.charset.Charset;
 import java.util.Map;
+import java.util.Objects;
 
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
@@ -24,42 +25,40 @@
 import org.eclipse.rdf4j.http.server.repository.RepositoryInterceptor;
 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.RepositoryConnection;
 import org.eclipse.rdf4j.repository.RepositoryException;
 import org.eclipse.rdf4j.rio.RDFFormat;
+import org.eclipse.rdf4j.rio.RDFHandler;
 import org.eclipse.rdf4j.rio.RDFHandlerException;
 import org.eclipse.rdf4j.rio.RDFWriter;
 import org.eclipse.rdf4j.rio.RDFWriterFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.web.servlet.View;
 
 /**
- * View used to export statements. Renders the statements as RDF using a serialization specified using a parameter or
- * Accept header.
+ * Streams statements as RDF in the format requested by the client.
  *
  * @author Herko ter Horst
  */
 public class ExportStatementsView implements View {
 
 	public static final String SUBJECT_KEY = "subject";
-
 	public static final String PREDICATE_KEY = "predicate";
-
 	public static final String OBJECT_KEY = "object";
-
 	public static final String CONTEXTS_KEY = "contexts";
-
 	public static final String USE_INFERENCING_KEY = "useInferencing";
-
 	public static final String CONNECTION_KEY = "connection";
-
 	public static final String TRANSACTION_ID_KEY = "transactionID";
-
 	public static final String FACTORY_KEY = "factory";
-
 	public static final String HEADERS_ONLY = "headersOnly";
 
 	private static final ExportStatementsView INSTANCE = new ExportStatementsView();
+	public static final int MAX_NUMBER_OF_STATEMENTS_WHEN_TESTING_FOR_POSSIBLE_EXCEPTIONS = 1024;
+
+	private static final Logger logger = LoggerFactory.getLogger(ExportStatementsView.class);
 
 	public static ExportStatementsView getInstance() {
 		return INSTANCE;
@@ -70,53 +69,152 @@ private ExportStatementsView() {
 
 	@Override
 	public String getContentType() {
+		// Spring ignores this for View implementations; we set it in render().
 		return null;
 	}
 
-	@SuppressWarnings("rawtypes")
 	@Override
 	public void render(Map model, HttpServletRequest request, HttpServletResponse response) throws Exception {
-		Resource subj = (Resource) model.get(SUBJECT_KEY);
+
+		response.setBufferSize(1024 * 1024); // 1MB
+
+		Resource subj = (Resource) Objects.requireNonNull(model, "model should not be null").get(SUBJECT_KEY);
 		IRI pred = (IRI) model.get(PREDICATE_KEY);
 		Value obj = (Value) model.get(OBJECT_KEY);
 		Resource[] contexts = (Resource[]) model.get(CONTEXTS_KEY);
-		boolean useInferencing = (Boolean) model.get(USE_INFERENCING_KEY);
+		boolean useInferencing = Boolean.TRUE.equals(model.get(USE_INFERENCING_KEY));
+		boolean headersOnly = Boolean.TRUE.equals(model.get(HEADERS_ONLY));
+
+		RDFWriterFactory factory = (RDFWriterFactory) model.get(FACTORY_KEY);
+		RDFFormat rdfFormat = factory.getRDFFormat();
 
-		boolean headersOnly = (Boolean) model.get(HEADERS_ONLY);
+		attemptToDetectExceptions(request, factory, headersOnly, subj, pred, obj, useInferencing, contexts);
+
+		response.setStatus(SC_OK);
+
+		String mimeType = rdfFormat.getDefaultMIMEType();
+		if (rdfFormat.hasCharset()) {
+			Charset charset = rdfFormat.getCharset();
+			mimeType += "; charset=" + charset.name();
+		}
+		response.setContentType(mimeType);
+
+		String filename = "statements";
+		if (rdfFormat.getDefaultFileExtension() != null) {
+			filename += "." + rdfFormat.getDefaultFileExtension();
+		}
+		response.setHeader("Content-Disposition", "attachment; filename=" + filename);
+
+		if (headersOnly) {
+			response.setContentLength(0);
+			response.flushBuffer();
+			return;
+		}
+
+		try (OutputStream out = response.getOutputStream()) {
+			RDFWriter writer = factory.getWriter(out);
+			try (RepositoryConnection conn = RepositoryInterceptor.getRepositoryConnection(request)) {
+				conn.exportStatements(subj, pred, obj, useInferencing, writer, contexts);
+				out.flush();
+				response.flushBuffer();
+			} catch (RDFHandlerException e) {
+				var serverHTTPException = new ServerHTTPException("Serialization error: " + e.getMessage(), e);
+				if (!response.isCommitted()) {
+					response.reset();
+				}
+				throw serverHTTPException;
+			} catch (RepositoryException e) {
+				var serverHTTPException = new ServerHTTPException("Repository error: " + e.getMessage(), e);
+				if (!response.isCommitted()) {
+					response.reset();
+				}
+				throw serverHTTPException;
+			} catch (Throwable e) {
+				if (!response.isCommitted()) {
+					response.reset();
+				}
+				throw e;
+			}
 
-		RDFWriterFactory rdfWriterFactory = (RDFWriterFactory) model.get(FACTORY_KEY);
+		}
 
-		RDFFormat rdfFormat = rdfWriterFactory.getRDFFormat();
+	}
 
-		try (ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
-			RDFWriter rdfWriter = rdfWriterFactory.getWriter(baos);
+	private static void attemptToDetectExceptions(HttpServletRequest request, RDFWriterFactory rdfWriterFactory,
+			boolean headersOnly, Resource subj, IRI pred, Value obj, boolean useInferencing, Resource[] contexts)
+			throws IOException, ServerHTTPException {
+		try (OutputStream out = OutputStream.nullOutputStream()) {
+			RDFHandler rdfWriter = new LimitedSizeRDFHandler(rdfWriterFactory.getWriter(out),
+					MAX_NUMBER_OF_STATEMENTS_WHEN_TESTING_FOR_POSSIBLE_EXCEPTIONS);
 			if (!headersOnly) {
 				try (RepositoryConnection conn = RepositoryInterceptor.getRepositoryConnection(request)) {
 					conn.exportStatements(subj, pred, obj, useInferencing, rdfWriter, contexts);
 				} catch (RDFHandlerException e) {
 					throw new ServerHTTPException("Serialization error: " + e.getMessage(), e);
 				} catch (RepositoryException e) {
 					throw new ServerHTTPException("Repository error: " + e.getMessage(), e);
+				} catch (LimitedSizeReachedException ignored) {
 				}
 			}
-			try (OutputStream out = response.getOutputStream()) {
-				response.setStatus(SC_OK);
+		}
+	}
 
-				String mimeType = rdfFormat.getDefaultMIMEType();
-				if (rdfFormat.hasCharset()) {
-					Charset charset = rdfFormat.getCharset();
-					mimeType += "; charset=" + charset.name();
-				}
-				response.setContentType(mimeType);
+	private static class LimitedSizeRDFHandler implements RDFHandler {
 
-				String filename = "statements";
-				if (rdfFormat.getDefaultFileExtension() != null) {
-					filename += "." + rdfFormat.getDefaultFileExtension();
-				}
-				response.setHeader("Content-Disposition", "attachment; filename=" + filename);
-				out.write(baos.toByteArray());
+		private final RDFHandler delegate;
+		private final long maxSize;
+		private long currentSize = 0;
+
+		public LimitedSizeRDFHandler(RDFHandler delegate, long maxSize) {
+			this.delegate = delegate;
+			this.maxSize = maxSize;
+		}
+
+		@Override
+		public void startRDF() throws RDFHandlerException {
+			delegate.startRDF();
+		}
+
+		@Override
+		public void endRDF() throws RDFHandlerException {
+			delegate.endRDF();
+		}
+
+		@Override
+		public void handleNamespace(String prefix, String uri) throws RDFHandlerException {
+			delegate.handleNamespace(prefix, uri);
+			incrementCurrentSize();
+		}
+
+		@Override
+		public void handleStatement(Statement st) throws RDFHandlerException {
+			delegate.handleStatement(st);
+			incrementCurrentSize();
+		}
+
+		@Override
+		public void handleComment(String comment) throws RDFHandlerException {
+			delegate.handleComment(comment);
+			incrementCurrentSize();
+		}
+
+		private void incrementCurrentSize() {
+			currentSize++;
+			if (currentSize > maxSize) {
+				endRDF();
+				logger.trace(
+						"Limited size reached, throwing LimitedSizeReachedException to signal that we are done testing the export of statements for exceptions.");
+				throw new LimitedSizeReachedException();
 			}
 		}
 	}
 
+	private static class LimitedSizeReachedException extends RuntimeException {
+		@Override
+		public Throwable fillInStackTrace() {
+			// Do not fill in the stack trace to avoid performance overhead
+			return this;
+		}
+	}
+
 }

tools/server/src/test/java/org/eclipse/rdf4j/http/server/ProtocolIT.java

@@ -29,6 +29,7 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Random;
+import java.util.Scanner;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -46,6 +47,7 @@
 import org.apache.http.message.BasicNameValuePair;
 import org.eclipse.rdf4j.common.io.IOUtil;
 import org.eclipse.rdf4j.http.protocol.Protocol;
+import org.eclipse.rdf4j.model.IRI;
 import org.eclipse.rdf4j.model.Namespace;
 import org.eclipse.rdf4j.model.ValueFactory;
 import org.eclipse.rdf4j.model.impl.SimpleNamespace;
@@ -57,6 +59,7 @@
 import org.eclipse.rdf4j.rio.RDFFormat;
 import org.eclipse.rdf4j.rio.Rio;
 import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
@@ -86,6 +89,12 @@ public static void stopServer() throws Exception {
 		server.stop();
 	}
 
+	@AfterEach
+	public void clearRepository() throws Exception {
+		// Clear the repository after each test
+		delete(Protocol.getStatementsLocation(TestServer.REPOSITORY_URL));
+	}
+
 	/**
 	 * Tests the server's methods for updating all data in a repository.
 	 */
@@ -172,7 +181,7 @@ public void testQueryDirect_POST() throws Exception {
 
 		System.out.println("Query Direct POST Status: " + response.getStatusLine());
 		int statusCode = response.getStatusLine().getStatusCode();
-		assertEquals(true, statusCode >= 200 && statusCode < 400);
+		assertTrue(statusCode >= 200 && statusCode < 400);
 	}
 
 	/**
@@ -192,7 +201,7 @@ public void testUpdateDirect_POST() throws Exception {
 
 		System.out.println("Update Direct Post Status: " + response.getStatusLine());
 		int statusCode = response.getStatusLine().getStatusCode();
-		assertEquals(true, statusCode >= 200 && statusCode < 400);
+		assertTrue(statusCode >= 200 && statusCode < 400);
 	}
 
 	/**
@@ -215,7 +224,7 @@ public void testUpdateForm_POST() throws Exception {
 
 		System.out.println("Update Form Post Status: " + response.getStatusLine());
 		int statusCode = response.getStatusLine().getStatusCode();
-		assertEquals(true, statusCode >= 200 && statusCode < 400);
+		assertTrue(statusCode >= 200 && statusCode < 400);
 	}
 
 	/**
@@ -370,6 +379,49 @@ public void testUpdateResponse_HEAD() throws Exception {
 		}
 	}
 
+	@Test
+	public void testUploadAndRetrieveStatements_GET() throws Exception {
+
+		String statementsLocation = Protocol.getStatementsLocation(TestServer.REPOSITORY_URL);
+
+		// PUT the Turtle file into the repository
+		final String baseLocation = Protocol.getStatementsLocation(TestServer.REPOSITORY_URL);
+		final String file = "/testcases/default-graph-2.ttl";
+
+		// 1. PUT the same file multiple times so that we would trigger an OOM error when retrieving it if it were not
+		// directly written to the http output stream
+		for (int i = 1; i <= 20000; i++) {
+			IRI context = vf.createIRI("http://example.org/graph" + i);
+			putFile(baseLocation, file, context);
+		}
+
+		// GET all statements back from the same endpoint
+		URL url = new URL(statementsLocation);
+		HttpURLConnection conn = (HttpURLConnection) url.openConnection();
+		conn.setRequestMethod("GET");
+		conn.setRequestProperty("Accept", RDFFormat.NQUADS.getDefaultMIMEType()); // ask for easy-to-parse format
+		conn.connect();
+
+		try {
+			assertEquals(HttpURLConnection.HTTP_OK, conn.getResponseCode(),
+					"GET /statements should respond 200 OK");
+
+			// Parse the response stream and count triples
+			try (InputStream in = conn.getInputStream()) {
+				Scanner scanner = new Scanner(in);
+				int count = 0;
+				while (scanner.hasNext()) {
+					scanner.nextLine();
+					count++;
+				}
+				assertEquals(1860000, count, "Expected 1860000 triples, but got " + count + " instead.");
+			}
+		} finally {
+			conn.disconnect();
+		}
+
+	}
+
 	/**
 	 * Test for SES-1861
 	 *
@@ -560,9 +612,16 @@ private void deleteNamespace(String location) throws Exception {
 		}
 	}
 
-	private void putFile(String location, String file) throws Exception {
-		System.out.println("Put file to " + location);
+	/**
+	 * PUT a file into a specific named graph (context) by adding the ?context=<encodedIRI> query parameter.
+	 */
+	private void putFile(String repositoryBaseLocation, String file, IRI context) throws Exception {
+		String location = repositoryBaseLocation
+				+ "?" + Protocol.CONTEXT_PARAM_NAME + "=" + Protocol.encodeValue(context);
+		putFile(location, file); // delegate to the existing helper
+	}
 
+	private void putFile(String location, String file) throws Exception {
 		URL url = new URL(location);
 		HttpURLConnection conn = (HttpURLConnection) url.openConnection();
 		conn.setRequestMethod("PUT");