Surface server errors for RDF GSP exports (#129)

Previously, GSP based RDF exports were unable to extract any error
messages from Neptune in the trailing headers. The changes here will
catch parsing exceptions (which is a symptom of a server side error)
and attempt to extract additional server failure details from the trailers.

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Amazon Neptune Export CHANGELOG
 
+## Neptune Export v1.1.7 (Release Date: TBD):
+
+- Surface server errors for RDF GSP exports
+
 ## Neptune Export v1.1.6 (Release Date: April 17, 2024):
 
 ### Bug Fixes:

src/main/java/com/amazonaws/services/neptune/rdf/NeptuneSparqlClient.java

@@ -12,6 +12,7 @@
 
 package com.amazonaws.services.neptune.rdf;
 
+import com.amazonaws.AmazonServiceException;
 import com.amazonaws.auth.AWSCredentialsProvider;
 import com.amazonaws.neptune.auth.NeptuneSigV4SignerException;
 import com.amazonaws.services.neptune.cluster.ConnectionConfig;
@@ -21,9 +22,13 @@
 import com.amazonaws.services.neptune.rdf.io.NeptuneExportSparqlRepository;
 import com.amazonaws.services.neptune.rdf.io.RdfTargetConfig;
 import com.amazonaws.services.neptune.util.EnvironmentVariableUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.http.Header;
 import org.apache.http.client.HttpClient;
 import org.apache.http.client.methods.HttpGet;
 import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.impl.io.ChunkedInputStream;
+import org.apache.http.util.EntityUtils;
 import org.eclipse.rdf4j.http.client.HttpClientSessionManager;
 import org.eclipse.rdf4j.http.client.RDF4JProtocolSession;
 import org.eclipse.rdf4j.http.client.SPARQLProtocolSession;
@@ -34,6 +39,7 @@
 import org.eclipse.rdf4j.repository.sparql.SPARQLRepository;
 import org.eclipse.rdf4j.rio.ParserConfig;
 import org.eclipse.rdf4j.rio.RDFFormat;
+import org.eclipse.rdf4j.rio.RDFParseException;
 import org.eclipse.rdf4j.rio.RDFParser;
 import org.eclipse.rdf4j.rio.RDFWriter;
 import org.eclipse.rdf4j.rio.Rio;
@@ -42,6 +48,9 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.UnsupportedEncodingException;
+import java.lang.reflect.Field;
+import java.net.URLDecoder;
 import java.util.List;
 import java.util.Random;
 import java.util.stream.Collectors;
@@ -133,7 +142,7 @@ public void executeTupleQuery(String sparql, RdfTargetConfig targetConfig) throw
 
         } catch (Exception e) {
             if (repository instanceof NeptuneExportSparqlRepository) {
-                throw new RuntimeException(((NeptuneExportSparqlRepository) repository).getErrorMessageFromTrailers(), e);
+                throw new AmazonServiceException(((NeptuneExportSparqlRepository) repository).getErrorMessageFromTrailers(), e);
             }
             else {
                 throw new RuntimeException(e);
@@ -153,7 +162,7 @@ public void executeGraphQuery(String sparql, RdfTargetConfig targetConfig) throw
 
         } catch (Exception e) {
             if (repository instanceof NeptuneExportSparqlRepository) {
-                throw new RuntimeException(((NeptuneExportSparqlRepository) repository).getErrorMessageFromTrailers(), e);
+                throw new AmazonServiceException(((NeptuneExportSparqlRepository) repository).getErrorMessageFromTrailers(), e);
             }
             else {
                 throw new RuntimeException(e);
@@ -181,6 +190,7 @@ void executeGSPExport(RdfTargetConfig targetConfig, String graph) throws IOExcep
         HttpClient httpClient = chooseRepository().getHttpClient();
         HttpUriRequest request = new HttpGet(getGSPEndpoint(graph));
         request.addHeader("Accept", "application/n-triples");
+        request.addHeader("te", "trailers");
 
         org.apache.http.HttpResponse response = httpClient.execute(request);
         InputStream responseBody = response.getEntity().getContent();
@@ -192,6 +202,14 @@ void executeGSPExport(RdfTargetConfig targetConfig, String graph) throws IOExcep
 
             try {
                 rdfParser.parse(responseBody);
+            } catch(RDFParseException e) {
+                String serverErrorMessage = getErrorMessageFromTrailers(response);
+                if (StringUtils.isNotEmpty(serverErrorMessage)) {
+                    throw new AmazonServiceException(getErrorMessageFromTrailers(response), e);
+                } else {
+                    throw new RuntimeException("Failed to parse RDF response. This may indicate malformed data in the " +
+                            "results, or a corrupted response from the server potentially due to query execution failure.", e);
+                }
             } finally {
                 responseBody.close();
             }
@@ -217,4 +235,47 @@ private String getGSPEndpoint(String graphName) {
                 graphName);
     }
 
+    /**
+     * Attempts to extract error messages from trailing headers from the most recent response received by 'repository'.
+     * If no trailers are found an empty String is returned.
+     */
+    static String getErrorMessageFromTrailers(org.apache.http.HttpResponse response) {
+        if (response == null) {
+            return "";
+        }
+        InputStream responseInStream;
+        try {
+            responseInStream = response.getEntity().getContent();
+        } catch (IOException e) {
+            return "";
+        }
+        // HTTPClient 4.5.13 provides no methods for accessing trailers from a wrapped stream requiring the use of
+        // reflection to break encapsulation. This bug is being tracked in https://issues.apache.org/jira/browse/HTTPCLIENT-2263.
+        while (!(responseInStream instanceof ChunkedInputStream)){
+            Field wrappedStreamField;
+            try {
+                wrappedStreamField = responseInStream.getClass().getDeclaredField("wrappedStream");
+                wrappedStreamField.setAccessible(true);
+                responseInStream = (InputStream) wrappedStreamField.get(responseInStream);
+                wrappedStreamField.setAccessible(false);
+            } catch (Exception e) {
+                return "";
+            }
+        }
+        // Consume remaining response, ensures trailers have been consumed and are available
+        EntityUtils.consumeQuietly(response.getEntity());
+
+        Header[] trailers = ((ChunkedInputStream) responseInStream).getFooters();
+        StringBuilder messageBuilder = new StringBuilder();
+        for (Header trailer : trailers) {
+            try {
+                messageBuilder.append(URLDecoder.decode(trailer.toString(), "UTF-8"));
+            } catch (UnsupportedEncodingException e) {
+                messageBuilder.append(trailer);
+            }
+            messageBuilder.append('\n');
+        }
+        return messageBuilder.toString();
+    }
+
 }