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

82.64

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

package org.nanopub.extra.services;

import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.util.EntityUtils;
import org.nanopub.NanopubUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;

/**
 * Second-generation query API call.
 */
public class QueryCall {

    private static int parallelCallCount = 2;
    private static int maxRetryCount = 3;
    private static final Logger logger = LoggerFactory.getLogger(QueryCall.class);

    /**
     * Run a query call with the given query ID and parameters.
     *
     * @param queryRef the reference to the query to run
     * @return the HTTP response from the query API
     * @throws APINotReachableException       if the API is not reachable after retries
     * @throws NotEnoughAPIInstancesException if there are not enough API instances available
     */
    public static HttpResponse run(QueryRef queryRef) throws APINotReachableException, NotEnoughAPIInstancesException {
        int retryCount = 0;
        while (retryCount < maxRetryCount) {
            QueryCall apiCall = new QueryCall(queryRef);
            apiCall.run();
            while (!apiCall.calls.isEmpty() && apiCall.resp == null) {
                try {
                    Thread.sleep(200);
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                }
            }
            if (apiCall.resp != null) {
                return apiCall.resp;
            }
            retryCount = retryCount + 1;
        }
        throw new APINotReachableException("Giving up contacting API: " + queryRef.getQueryId());
    }

    /**
     * List of available query API instances.
     */
    // TODO Available services should be retrieved from a setting, not hard-coded:
    public static String[] queryApiInstances = new String[]{
            "https://query.knowledgepixels.com/",
            "https://query.petapico.org/",
            "https://query.nanodash.net/"
    };

    private static List<String> checkedApiInstances;

    /**
     * Returns the list of available query API instances that are currently accessible.
     *
     * @return a list of accessible query API instance
     */
    public static List<String> getApiInstances() throws NotEnoughAPIInstancesException {
        if (checkedApiInstances != null) return checkedApiInstances;
        checkedApiInstances = new ArrayList<>();
        for (String a : queryApiInstances) {
            try {
                logger.info("Checking API instance: {}", a);
                HttpResponse resp = NanopubUtils.getHttpClient().execute(new HttpGet(a));
                EntityUtils.consumeQuietly(resp.getEntity());
                if (wasSuccessful(resp)) {
                    logger.info("SUCCESS: Nanopub Query instance is accessible: {}", a);
                    checkedApiInstances.add(a);
                } else {
                    logger.error("FAILURE: Nanopub Query instance isn't accessible: {}", a);
                }
            } catch (IOException ex) {
                logger.error("FAILURE: Nanopub Query instance isn't accessible: {}", a);
            }
        }
        logger.info("{} accessible Nanopub Query instances", checkedApiInstances.size());
        if (checkedApiInstances.size() < 2) {
            checkedApiInstances = null;
            throw new NotEnoughAPIInstancesException("Not enough healthy Nanopub Query instances available");
        }
        return checkedApiInstances;
    }

    private QueryRef queryRef;
    private List<String> apisToCall = new ArrayList<>();
    private List<Call> calls = new ArrayList<>();

    private HttpResponse resp;

    private QueryCall(QueryRef queryRef) {
        this.queryRef = queryRef;
        logger.info("Invoking API operation {}", queryRef);
    }

    private void run() throws NotEnoughAPIInstancesException {
        List<String> apiInstancesToTry = new LinkedList<>(getApiInstances());
        while (!apiInstancesToTry.isEmpty() && apisToCall.size() < parallelCallCount) {
            int randomIndex = (int) ((Math.random() * apiInstancesToTry.size()));
            String apiUrl = apiInstancesToTry.get(randomIndex);
            apisToCall.add(apiUrl);
            logger.info("Trying API ({}) {}", apisToCall.size(), apiUrl);
            apiInstancesToTry.remove(randomIndex);
        }
        for (String api : apisToCall) {
            Call call = new Call(api);
            calls.add(call);
            new Thread(call).start();
        }
    }

    private synchronized void finished(Call call, HttpResponse resp, String apiUrl) {
        if (this.resp != null) { // result already in
            EntityUtils.consumeQuietly(resp.getEntity());
            return;
        }
        logger.info("Result in from {}:", apiUrl);
        logger.info("- Request: {}", queryRef);
        logger.info("- Response size: {}", resp.getEntity().getContentLength());
        this.resp = resp;

        for (Call c : calls) {
            if (c != call) c.abort();
        }
    }

    private static boolean wasSuccessful(HttpResponse resp) {
        if (resp == null || resp.getEntity() == null) return false;
        int c = resp.getStatusLine().getStatusCode();
        if (c < 200 || c >= 300) return false;
        return true;
    }

    private static boolean wasSuccessfulNonempty(HttpResponse resp) {
        if (!wasSuccessful(resp)) return false;
        // TODO Make sure we always return proper error codes, and then this shouldn't be necessary:
        if (resp.getHeaders("Content-Length").length > 0 && resp.getEntity().getContentLength() < 0) return false;
        return true;
    }


    private class Call implements Runnable {

        private String apiUrl;
        private HttpGet get;

        public Call(String apiUrl) {
            this.apiUrl = apiUrl;
        }

        public void run() {
            get = new HttpGet(apiUrl + "api/" + queryRef.getAsUrlString());
            get.setHeader("Accept", "text/csv, text/turtle;q=0.9, application/ld+json;q=0.8");
            HttpResponse resp = null;
            try {
                resp = NanopubUtils.getHttpClient().execute(get);
                if (!wasSuccessfulNonempty(resp)) {
                    throw new IOException(resp.getStatusLine().toString());
                }
                finished(this, resp, apiUrl);
            } catch (Exception ex) {
                if (resp != null) EntityUtils.consumeQuietly(resp.getEntity());
                logger.error("Request to {} was not successful: {}", apiUrl, ex.getMessage());
            }
            calls.remove(this);
        }

        private void abort() {
            if (get == null) return;
            if (get.isAborted()) return;
            get.abort();
        }

    }

}

1	package org.nanopub.extra.services;
2
3	import org.apache.http.HttpResponse;
4	import org.apache.http.client.methods.HttpGet;
5	import org.apache.http.util.EntityUtils;
6	import org.nanopub.NanopubUtils;
7	import org.slf4j.Logger;
8	import org.slf4j.LoggerFactory;
9
10	import java.io.IOException;
11	import java.util.ArrayList;
12	import java.util.LinkedList;
13	import java.util.List;
14
15	/**
16	* Second-generation query API call.
17	*/
18	public class QueryCall {
19
20	private static int parallelCallCount = 2;	6✔
21	private static int maxRetryCount = 3;	6✔
22	private static final Logger logger = LoggerFactory.getLogger(QueryCall.class);	9✔
23
24	/**
25	* Run a query call with the given query ID and parameters.
26	*
27	* @param queryRef the reference to the query to run
28	* @return the HTTP response from the query API
29	* @throws APINotReachableException if the API is not reachable after retries
30	* @throws NotEnoughAPIInstancesException if there are not enough API instances available
31	*/
32	public static HttpResponse run(QueryRef queryRef) throws APINotReachableException, NotEnoughAPIInstancesException {
33	int retryCount = 0;	6✔
34	while (retryCount < maxRetryCount) {	9!
35	QueryCall apiCall = new QueryCall(queryRef);	15✔
36	apiCall.run();	6✔
37	while (!apiCall.calls.isEmpty() && apiCall.resp == null) {	21!
38	try {
39	Thread.sleep(200);	6✔
40	} catch (InterruptedException ex) {	×
41	Thread.currentThread().interrupt();	×
42	}	3✔
43	}
44	if (apiCall.resp != null) {	9!
45	return apiCall.resp;	9✔
46	}
47	retryCount = retryCount + 1;	×
48	}	×
49	throw new APINotReachableException("Giving up contacting API: " + queryRef.getQueryId());	×
50	}
51
52	/**
53	* List of available query API instances.
54	*/
55	// TODO Available services should be retrieved from a setting, not hard-coded:
56	public static String[] queryApiInstances = new String[]{	48✔
57	"https://query.knowledgepixels.com/",
58	"https://query.petapico.org/",
59	"https://query.nanodash.net/"
60	};
61
62	private static List<String> checkedApiInstances;
63
64	/**
65	* Returns the list of available query API instances that are currently accessible.
66	*
67	* @return a list of accessible query API instance
68	*/
69	public static List<String> getApiInstances() throws NotEnoughAPIInstancesException {
70	if (checkedApiInstances != null) return checkedApiInstances;	12✔
71	checkedApiInstances = new ArrayList<>();	12✔
72	for (String a : queryApiInstances) {	48✔
73	try {
74	logger.info("Checking API instance: {}", a);	12✔
75	HttpResponse resp = NanopubUtils.getHttpClient().execute(new HttpGet(a));	21✔
76	EntityUtils.consumeQuietly(resp.getEntity());	9✔
77	if (wasSuccessful(resp)) {	9✔
78	logger.info("SUCCESS: Nanopub Query instance is accessible: {}", a);	12✔
79	checkedApiInstances.add(a);	15✔
80	} else {
81	logger.error("FAILURE: Nanopub Query instance isn't accessible: {}", a);	12✔
82	}
83	} catch (IOException ex) {	×
84	logger.error("FAILURE: Nanopub Query instance isn't accessible: {}", a);	×
85	}	3✔
86	}
87	logger.info("{} accessible Nanopub Query instances", checkedApiInstances.size());	18✔
88	if (checkedApiInstances.size() < 2) {	12✔
89	checkedApiInstances = null;	6✔
90	throw new NotEnoughAPIInstancesException("Not enough healthy Nanopub Query instances available");	15✔
91	}
92	return checkedApiInstances;	6✔
93	}
94
95	private QueryRef queryRef;
96	private List<String> apisToCall = new ArrayList<>();	15✔
97	private List<Call> calls = new ArrayList<>();	15✔
98
99	private HttpResponse resp;
100
101	private QueryCall(QueryRef queryRef) {	6✔
102	this.queryRef = queryRef;	9✔
103	logger.info("Invoking API operation {}", queryRef);	12✔
104	}	3✔
105
106	private void run() throws NotEnoughAPIInstancesException {
107	List<String> apiInstancesToTry = new LinkedList<>(getApiInstances());	15✔
108	while (!apiInstancesToTry.isEmpty() && apisToCall.size() < parallelCallCount) {	24!
109	int randomIndex = (int) ((Math.random() * apiInstancesToTry.size()));	21✔
110	String apiUrl = apiInstancesToTry.get(randomIndex);	15✔
111	apisToCall.add(apiUrl);	15✔
112	logger.info("Trying API ({}) {}", apisToCall.size(), apiUrl);	24✔
113	apiInstancesToTry.remove(randomIndex);	12✔
114	}	3✔
115	for (String api : apisToCall) {	33✔
116	Call call = new Call(api);	18✔
117	calls.add(call);	15✔
118	new Thread(call).start();	15✔
119	}	3✔
120	}	3✔
121
122	private synchronized void finished(Call call, HttpResponse resp, String apiUrl) {
123	if (this.resp != null) { // result already in	9!
124	EntityUtils.consumeQuietly(resp.getEntity());	×
125	return;	×
126	}
127	logger.info("Result in from {}:", apiUrl);	12✔
128	logger.info("- Request: {}", queryRef);	15✔
129	logger.info("- Response size: {}", resp.getEntity().getContentLength());	21✔
130	this.resp = resp;	9✔
131
132	for (Call c : calls) {	33✔
133	if (c != call) c.abort();	15✔
134	}	3✔
135	}	3✔
136
137	private static boolean wasSuccessful(HttpResponse resp) {
138	if (resp == null \|\| resp.getEntity() == null) return false;	15!
139	int c = resp.getStatusLine().getStatusCode();	12✔
140	if (c < 200 \|\| c >= 300) return false;	24!
141	return true;	6✔
142	}
143
144	private static boolean wasSuccessfulNonempty(HttpResponse resp) {
145	if (!wasSuccessful(resp)) return false;	9!
146	// TODO Make sure we always return proper error codes, and then this shouldn't be necessary:
147	if (resp.getHeaders("Content-Length").length > 0 && resp.getEntity().getContentLength() < 0) return false;	33!
148	return true;	6✔
149	}
150
151
152	private class Call implements Runnable {
153
154	private String apiUrl;
155	private HttpGet get;
156
157	public Call(String apiUrl) {	15✔
158	this.apiUrl = apiUrl;	9✔
159	}	3✔
160
161	public void run() {
162	get = new HttpGet(apiUrl + "api/" + queryRef.getAsUrlString());	36✔
163	get.setHeader("Accept", "text/csv, text/turtle;q=0.9, application/ld+json;q=0.8");	15✔
164	HttpResponse resp = null;	6✔
165	try {
166	resp = NanopubUtils.getHttpClient().execute(get);	15✔
167	if (!wasSuccessfulNonempty(resp)) {	9!
168	throw new IOException(resp.getStatusLine().toString());	×
169	}
170	finished(this, resp, apiUrl);	21✔
171	} catch (Exception ex) {	3✔
172	if (resp != null) EntityUtils.consumeQuietly(resp.getEntity());	6!
173	logger.error("Request to {} was not successful: {}", apiUrl, ex.getMessage());	21✔
174	}	3✔
175	calls.remove(this);	18✔
176	}	3✔
177
178	private void abort() {
179	if (get == null) return;	9!
180	if (get.isAborted()) return;	12!
181	get.abort();	9✔
182	}	3✔
183
184	}
185
186	}

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