23729181399

Committed 30 Mar 2026 05:19AM UTC coverage: 51.353% (+0.01%) from 51.341%

Build # 23729181399

Build Type

push

github

Committed by

tkuhn

Commit Message

fix: prevent HTTP connection pool exhaustion under concurrent load

Increase connectionRequestTimeout from 500ms to 10s to match the other
timeouts — 500ms was too aggressive for a shared pool under concurrent
load, causing "Timeout waiting for connection from pool" errors
(knowledgepixels/nanodash#416).

Also fix two connection leak paths: consume response entities in
QueryCall.getApiInstances() health checks, and ensure QueryAccess.call()
always releases connections via try-with-resources and a finally block.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Coverage Stats

1030 of 2990 branches covered (34.45%)

Branch coverage included in aggregate %.

5253 of 9245 relevant lines covered (56.82%)

7.98 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.33

src/main/java/org/nanopub/extra/services/QueryAccess.java

package org.nanopub.extra.services;

import com.opencsv.CSVParserBuilder;
import com.opencsv.CSVReader;
import com.opencsv.CSVReaderBuilder;
import com.opencsv.CSVWriterBuilder;
import com.opencsv.ICSVParser;
import com.opencsv.ICSVWriter;
import com.opencsv.exceptions.CsvValidationException;
import org.apache.http.HttpResponse;
import org.apache.http.util.EntityUtils;
import org.eclipse.rdf4j.model.Model;
import org.eclipse.rdf4j.rio.RDFFormat;
import org.eclipse.rdf4j.rio.Rio;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Writer;

/**
 * Second-generation query API access
 */
public abstract class QueryAccess {

    /**
     * Process the header.
     *
     * @param line the header line from the CSV response
     */
    protected abstract void processHeader(String[] line);

    /**
     * Process a line of data from the CSV response.
     *
     * @param line the line of data from the CSV response
     */
    protected abstract void processLine(String[] line);

    /**
     * Process the RDF content from a SPARQL CONSTRUCT query response.
     * Override this method to handle RDF responses. Default implementation is a no-op.
     *
     * @param model the parsed RDF model
     */
    protected void processRdfContent(Model model) {
    }

    /**
     * Call a query with the given queryId and parameters.
     *
     * @param queryRef the query reference
     * @throws FailedApiCallException         if the API call fails
     * @throws APINotReachableException       if the API is not reachable
     * @throws NotEnoughAPIInstancesException if there are not enough API instances available
     */
    public void call(QueryRef queryRef) throws FailedApiCallException, APINotReachableException, NotEnoughAPIInstancesException {
        HttpResponse resp = QueryCall.run(queryRef);
        try {
            String contentType = resp.getFirstHeader("Content-Type") != null
                    ? resp.getFirstHeader("Content-Type").getValue() : "";
            String mimeType = contentType.contains(";") ? contentType.split(";")[0].trim() : contentType.trim();
            if (mimeType.equals("text/csv") || mimeType.isEmpty()) {
                try (CSVReader csvReader = new CSVReaderBuilder(new BufferedReader(new InputStreamReader(resp.getEntity().getContent())))
                        .withCSVParser(new CSVParserBuilder().withEscapeChar(ICSVParser.NULL_CHARACTER).build())
                        .build()) {
                    String[] line = null;
                    int n = 0;
                    while ((line = csvReader.readNext()) != null) {
                        n++;
                        if (n == 1) {
                            processHeader(line);
                        } else {
                            processLine(line);
                        }
                    }
                } catch (IOException | CsvValidationException ex) {
                    throw new FailedApiCallException(ex);
                }
            } else {
                RDFFormat format = Rio.getParserFormatForMIMEType(mimeType).orElse(RDFFormat.TURTLE);
                try (InputStream in = resp.getEntity().getContent()) {
                    Model model = Rio.parse(in, format);
                    processRdfContent(model);
                } catch (IOException ex) {
                    throw new FailedApiCallException(ex);
                }
            }
        } finally {
            EntityUtils.consumeQuietly(resp.getEntity());
        }
    }

    /**
     * Print the response of a query in CSV format to the given writer.
     *
     * @param queryRef the query reference
     * @param writer   the writer to print the CSV response to
     * @throws FailedApiCallException         if the API call fails
     * @throws APINotReachableException       if the API is not reachable
     * @throws NotEnoughAPIInstancesException if there are not enough API instances available
     */
    public static void printCvsResponse(QueryRef queryRef, Writer writer) throws FailedApiCallException, APINotReachableException, NotEnoughAPIInstancesException {
        ICSVWriter icsvWriter = new CSVWriterBuilder(writer).withSeparator(',').build();
        QueryAccess a = new QueryAccess() {

            @Override
            protected void processHeader(String[] line) {
                icsvWriter.writeNext(line);
            }

            @Override
            protected void processLine(String[] line) {
                icsvWriter.writeNext(line);
            }

        };
        a.call(queryRef);
        icsvWriter.flushQuietly();
    }

    /**
     * Get the response of a query as an ApiResponse object.
     *
     * @param queryRef the query reference
     * @return an ApiResponse object containing the response data
     * @throws FailedApiCallException         if the API call fails
     * @throws APINotReachableException       if the API is not reachable
     * @throws NotEnoughAPIInstancesException if there are not enough API instances available
     */
    public static ApiResponse get(QueryRef queryRef) throws FailedApiCallException, APINotReachableException, NotEnoughAPIInstancesException {
        final ApiResponse response = new ApiResponse();
        QueryAccess a = new QueryAccess() {

            @Override
            protected void processHeader(String[] line) {
                response.setHeader(line);
            }

            @Override
            protected void processLine(String[] line) {
                response.add(line);
            }

            @Override
            protected void processRdfContent(Model model) {
                response.setRdfContent(model);
            }

        };
        a.call(queryRef);
        return response;
    }

}

1	package org.nanopub.extra.services;
2
3	import com.opencsv.CSVParserBuilder;
4	import com.opencsv.CSVReader;
5	import com.opencsv.CSVReaderBuilder;
6	import com.opencsv.CSVWriterBuilder;
7	import com.opencsv.ICSVParser;
8	import com.opencsv.ICSVWriter;
9	import com.opencsv.exceptions.CsvValidationException;
10	import org.apache.http.HttpResponse;
11	import org.apache.http.util.EntityUtils;
12	import org.eclipse.rdf4j.model.Model;
13	import org.eclipse.rdf4j.rio.RDFFormat;
14	import org.eclipse.rdf4j.rio.Rio;
15
16	import java.io.BufferedReader;
17	import java.io.IOException;
18	import java.io.InputStream;
19	import java.io.InputStreamReader;
20	import java.io.Writer;
21
22	/**
23	* Second-generation query API access
24	*/
25	public abstract class QueryAccess {	9✔
26
27	/**
28	* Process the header.
29	*
30	* @param line the header line from the CSV response
31	*/
32	protected abstract void processHeader(String[] line);
33
34	/**
35	* Process a line of data from the CSV response.
36	*
37	* @param line the line of data from the CSV response
38	*/
39	protected abstract void processLine(String[] line);
40
41	/**
42	* Process the RDF content from a SPARQL CONSTRUCT query response.
43	* Override this method to handle RDF responses. Default implementation is a no-op.
44	*
45	* @param model the parsed RDF model
46	*/
47	protected void processRdfContent(Model model) {
48	}	×
49
50	/**
51	* Call a query with the given queryId and parameters.
52	*
53	* @param queryRef the query reference
54	* @throws FailedApiCallException if the API call fails
55	* @throws APINotReachableException if the API is not reachable
56	* @throws NotEnoughAPIInstancesException if there are not enough API instances available
57	*/
58	public void call(QueryRef queryRef) throws FailedApiCallException, APINotReachableException, NotEnoughAPIInstancesException {
59	HttpResponse resp = QueryCall.run(queryRef);	9✔
60	try {
61	String contentType = resp.getFirstHeader("Content-Type") != null	12!
62	? resp.getFirstHeader("Content-Type").getValue() : "";	18✔
63	String mimeType = contentType.contains(";") ? contentType.split(";")[0].trim() : contentType.trim();	42✔
64	if (mimeType.equals("text/csv") \|\| mimeType.isEmpty()) {	21!
65	try (CSVReader csvReader = new CSVReaderBuilder(new BufferedReader(new InputStreamReader(resp.getEntity().getContent())))	48✔
66	.withCSVParser(new CSVParserBuilder().withEscapeChar(ICSVParser.NULL_CHARACTER).build())	9✔
67	.build()) {	6✔
68	String[] line = null;	6✔
69	int n = 0;	6✔
70	while ((line = csvReader.readNext()) != null) {	15✔
71	n++;	3✔
72	if (n == 1) {	9✔
73	processHeader(line);	12✔
74	} else {
75	processLine(line);	12✔
76	}
77	}
78	} catch (IOException \| CsvValidationException ex) {	×
79	throw new FailedApiCallException(ex);	×
80	}	3✔
81	} else {
82	RDFFormat format = Rio.getParserFormatForMIMEType(mimeType).orElse(RDFFormat.TURTLE);	18✔
83	try (InputStream in = resp.getEntity().getContent()) {	12✔
84	Model model = Rio.parse(in, format);	18✔
85	processRdfContent(model);	9✔
86	} catch (IOException ex) {	×
87	throw new FailedApiCallException(ex);	×
88	}	3✔
89	}
90	} finally {
91	EntityUtils.consumeQuietly(resp.getEntity());	9✔
92	}
93	}	3✔
94
95	/**
96	* Print the response of a query in CSV format to the given writer.
97	*
98	* @param queryRef the query reference
99	* @param writer the writer to print the CSV response to
100	* @throws FailedApiCallException if the API call fails
101	* @throws APINotReachableException if the API is not reachable
102	* @throws NotEnoughAPIInstancesException if there are not enough API instances available
103	*/
104	public static void printCvsResponse(QueryRef queryRef, Writer writer) throws FailedApiCallException, APINotReachableException, NotEnoughAPIInstancesException {
105	ICSVWriter icsvWriter = new CSVWriterBuilder(writer).withSeparator(',').build();	×
106	QueryAccess a = new QueryAccess() {	×
107
108	@Override
109	protected void processHeader(String[] line) {
110	icsvWriter.writeNext(line);	×
111	}	×
112
113	@Override
114	protected void processLine(String[] line) {
115	icsvWriter.writeNext(line);	×
116	}	×
117
118	};
119	a.call(queryRef);	×
120	icsvWriter.flushQuietly();	×
121	}	×
122
123	/**
124	* Get the response of a query as an ApiResponse object.
125	*
126	* @param queryRef the query reference
127	* @return an ApiResponse object containing the response data
128	* @throws FailedApiCallException if the API call fails
129	* @throws APINotReachableException if the API is not reachable
130	* @throws NotEnoughAPIInstancesException if there are not enough API instances available
131	*/
132	public static ApiResponse get(QueryRef queryRef) throws FailedApiCallException, APINotReachableException, NotEnoughAPIInstancesException {
133	final ApiResponse response = new ApiResponse();	12✔
134	QueryAccess a = new QueryAccess() {	33✔
135
136	@Override
137	protected void processHeader(String[] line) {
138	response.setHeader(line);	12✔
139	}	3✔
140
141	@Override
142	protected void processLine(String[] line) {
143	response.add(line);	12✔
144	}	3✔
145
146	@Override
147	protected void processRdfContent(Model model) {
148	response.setRdfContent(model);	12✔
149	}	3✔
150
151	};
152	a.call(queryRef);	9✔
153	return response;	6✔
154	}
155
156	}

Nanopublication / nanopub-java / 23729181399

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous