jdyer1 commented on code in PR #2402:
URL: https://github.com/apache/solr/pull/2402#discussion_r1562689501
##########
solr/solrj/src/test/org/apache/solr/client/solrj/impl/LBHttp2SolrClientTest.java:
##########
@@ -51,4 +67,249 @@ public void
testLBHttp2SolrClientWithTheseParamNamesInTheUrl() {
http2SolrClient.getUrlParamNames().toArray());
}
}
+
+ @Test
+ public void testAsyncDeprecated() {
+ testAsync(true);
+ }
+
+ @Test
+ public void testAsync() {
+ testAsync(false);
+ }
+
+ @Test
+ public void testAsyncWithFailures() {
+
+ // TODO: This demonstrates that the failing endpoint always gets retried,
but
+ // I would expect it to be labelled as a "zombie" and be skipped with
additional iterations.
+
+ LBSolrClient.Endpoint ep1 = new
LBSolrClient.Endpoint("http://endpoint.one";);
+ LBSolrClient.Endpoint ep2 = new
LBSolrClient.Endpoint("http://endpoint.two";);
+ List<LBSolrClient.Endpoint> endpointList = List.of(ep1, ep2);
+
+ Http2SolrClient.Builder b = new Http2SolrClient.Builder("http://base.url";);
+ try (MockHttp2SolrClient client = new
MockHttp2SolrClient("http://base.url";, b);
+ LBHttp2SolrClient testClient = new LBHttp2SolrClient.Builder(client,
ep1, ep2).build()) {
+
+ for (int j = 0; j < 2; j++) {
+ // j: first time Endpoint One will retrun error code 500.
+ // second time Endpoint One will be healthy
+
+ String basePathToSucceed;
+ if (j == 0) {
+ client.basePathToFail = ep1.getBaseUrl();
+ basePathToSucceed = ep2.getBaseUrl();
+ } else {
+ client.basePathToFail = ep2.getBaseUrl();
+ basePathToSucceed = ep1.getBaseUrl();
+ }
+
+ for (int i = 0; i < 10; i++) {
+ // i: we'll try 10 times to see if it behaves the same every time.
+
+ QueryRequest queryRequest = new QueryRequest(new
MapSolrParams(Map.of("q", "" + i)));
+ LBSolrClient.Req req = new LBSolrClient.Req(queryRequest,
endpointList);
+ String iterMessage = "iter j/i " + j + "/" + i;
+ try {
+ testClient.requestAsync(req).get(1, TimeUnit.MINUTES);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ fail("interrupted");
+ } catch (TimeoutException | ExecutionException e) {
+ fail(iterMessage + " Response ended in failure: " + e);
+ }
+ if (j == 0) {
+ // The first endpoint gives an exception, so it retries.
+ assertEquals(iterMessage, 2, client.lastBasePaths.size());
+
+ String failedBasePath = client.lastBasePaths.remove(0);
+ assertEquals(iterMessage, client.basePathToFail, failedBasePath);
+ } else {
+ // The first endpoint does not give the exception, it doesn't
retry.
+ assertEquals(iterMessage, 1, client.lastBasePaths.size());
+ }
+ String successBasePath = client.lastBasePaths.remove(0);
+ assertEquals(iterMessage, basePathToSucceed, successBasePath);
+ }
+ }
+ }
+ }
+
+ private void testAsync(boolean useDeprecatedApi) {
+ LBSolrClient.Endpoint ep1 = new
LBSolrClient.Endpoint("http://endpoint.one";);
+ LBSolrClient.Endpoint ep2 = new
LBSolrClient.Endpoint("http://endpoint.two";);
+ List<LBSolrClient.Endpoint> endpointList = List.of(ep1, ep2);
+
+ Http2SolrClient.Builder b = new Http2SolrClient.Builder("http://base.url";);
+ try (MockHttp2SolrClient client = new
MockHttp2SolrClient("http://base.url";, b);
+ LBHttp2SolrClient testClient = new LBHttp2SolrClient.Builder(client,
ep1, ep2).build()) {
+
+ int limit = 10; // For simplicity use an even limit
+ int halfLimit = limit / 2; // see TODO below
+
+ CountDownLatch cdl = new CountDownLatch(limit); // deprecated API use
+ List<LBTestAsyncListener> listeners = new ArrayList<>(); // deprecated
API use
+ List<CompletableFuture<LBSolrClient.Rsp>> responses = new ArrayList<>();
+
+ for (int i = 0; i < limit; i++) {
+ QueryRequest queryRequest = new QueryRequest(new
MapSolrParams(Map.of("q", "" + i)));
+ LBSolrClient.Req req = new LBSolrClient.Req(queryRequest,
endpointList);
+ if (useDeprecatedApi) {
+ LBTestAsyncListener listener = new LBTestAsyncListener(cdl);
+ listeners.add(listener);
+ testClient.asyncReq(req, listener);
+ } else {
+ responses.add(testClient.requestAsync(req));
+ }
+ }
+
+ if (useDeprecatedApi) {
+ try {
+ // This is just a formality. This is a single-threaded test.
+ cdl.await(1, TimeUnit.MINUTES);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ fail("interrupted");
+ }
+ }
+
+ QueryRequest[] queryRequests = new QueryRequest[limit];
+ int numEndpointOne = 0;
+ int numEndpointTwo = 0; // see TODO below
+ for (int i = 0; i < limit; i++) {
+ SolrRequest<?> lastSolrReq = client.lastSolrRequests.get(i);
+ assertTrue(lastSolrReq instanceof QueryRequest);
+ QueryRequest lastQueryReq = (QueryRequest) lastSolrReq;
+ int index = Integer.parseInt(lastQueryReq.getParams().get("q"));
+ assertNull("Found same request twice: " + index, queryRequests[index]);
+ queryRequests[index] = lastQueryReq;
+ if (lastQueryReq.getBasePath().equals(ep1.toString())) {
+ numEndpointOne++;
+ } else if (lastQueryReq.getBasePath().equals(ep2.toString())) {
+ numEndpointTwo++;
+ }
+ NamedList<Object> lastResponse;
+ if (useDeprecatedApi) {
+ LBTestAsyncListener lastAsyncListener = listeners.get(index);
+ assertTrue(lastAsyncListener.onStartCalled);
+ assertNull(lastAsyncListener.failure);
+ assertNotNull(lastAsyncListener.success);
+ lastResponse = lastAsyncListener.success.getResponse();
+ } else {
+ LBSolrClient.Rsp lastRsp = null;
+ try {
+ lastRsp = responses.get(index).get();
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ fail("interrupted");
+ } catch (ExecutionException ee) {
+ fail("Response " + index + " ended in failure: " + ee);
+ }
+ lastResponse = lastRsp.getResponse();
+ }
+
+ // The Mock will return {"response": index}.
+ assertEquals("" + index, lastResponse.get("response"));
+ }
+
+ // TODO: LBHttp2SolrClient creates a new "endpoint iterator" per
request, thus
Review Comment:
My concern is that unless the first node in the passed-in list is in the
Zombies, it always gets all the requests. Maybe this is the intention, that
the caller is supposed to shuffle the list themselves so as to give ultimate
control. But this is not what I expected from a class named "Load Balancing
Client". There seemed to be little by way of doc comments and unit tests to
confirm this. Probably, we need a follow-up issue to add more test coverage to
these LB clients and then better document or fix the behavior.
##########
solr/solrj/src/test/org/apache/solr/client/solrj/impl/HttpSolrClientTestBase.java:
##########
@@ -585,17 +597,21 @@ protected void testUpdateAsync() throws Exception {
}
}
- protected void testQueryAsync() throws Exception {
+ protected void testQueryAsync(boolean useDeprecatedApi) throws Exception {
ResponseParser rp = new XMLResponseParser();
DebugServlet.clear();
DebugServlet.addResponseHeader("Content-Type", "application/xml;
charset=UTF-8");
String url = getBaseUrl() + DEBUG_SERVLET_PATH;
HttpSolrClientBuilderBase<?, ?> b =
builder(url, DEFAULT_CONNECTION_TIMEOUT,
DEFAULT_CONNECTION_TIMEOUT).withResponseParser(rp);
int limit = 10;
- CountDownLatch cdl = new CountDownLatch(limit);
- DebugAsyncListener[] listeners = new DebugAsyncListener[limit];
- Cancellable[] cancellables = new Cancellable[limit];
+
+ CountDownLatch cdl = new CountDownLatch(limit); // Deprecated API use
Review Comment:
I think I hunted these all down and changed them.
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/Http2SolrClient.java:
##########
@@ -417,67 +419,101 @@ public void send(OutStream outStream, SolrRequest<?>
req, String collection) thr
outStream.flush();
}
- @SuppressWarnings("StaticAssignmentOfThrowable")
- private static final Exception CANCELLED_EXCEPTION = new Exception();
-
- private static final Cancellable FAILED_MAKING_REQUEST_CANCELLABLE = () ->
{};
-
@Override
public Cancellable asyncRequest(
SolrRequest<?> solrRequest,
String collection,
AsyncListener<NamedList<Object>> asyncListener) {
- MDCCopyHelper mdcCopyHelper = new MDCCopyHelper();
- Request req;
+ asyncListener.onStart();
+ CompletableFuture<NamedList<Object>> cf =
+ requestAsync(solrRequest, collection)
+ .whenComplete(
+ (nl, t) -> {
+ if (t != null) {
+ asyncListener.onFailure(t);
+ } else {
+ asyncListener.onSuccess(nl);
+ }
+ });
+ return () -> cf.cancel(true);
+ }
+
+ @Override
+ public CompletableFuture<NamedList<Object>> requestAsync(
+ final SolrRequest<?> solrRequest, String collection) {
+ if (ClientUtils.shouldApplyDefaultCollection(collection, solrRequest)) {
+ collection = defaultCollection;
+ }
+ MDCCopyHelper mdcCopyHelper = new MDCCopyHelper();
+ CompletableFuture<NamedList<Object>> future = new CompletableFuture<>();
+ final MakeRequestReturnValue mrrv;
+ final String url;
try {
- String url = getRequestPath(solrRequest, collection);
- InputStreamResponseListener listener =
- new InputStreamReleaseTrackingResponseListener() {
- @Override
- public void onHeaders(Response response) {
- super.onHeaders(response);
- executor.execute(
- () -> {
- InputStream is = getInputStream();
- try {
- NamedList<Object> body =
- processErrorsAndResponse(solrRequest, response, is,
url);
- mdcCopyHelper.onBegin(null);
- log.debug("response processing success");
- asyncListener.onSuccess(body);
- } catch (RemoteSolrException e) {
- if (SolrException.getRootCause(e) !=
CANCELLED_EXCEPTION) {
+ url = getRequestPath(solrRequest, collection);
+ mrrv = makeRequest(solrRequest, url, true);
+ } catch (SolrServerException | IOException e) {
+ future.completeExceptionally(e);
+ return future;
+ }
+ final ResponseParser parser =
+ solrRequest.getResponseParser() == null ? this.parser :
solrRequest.getResponseParser();
+ mrrv.request
+ .onRequestQueued(asyncTracker.queuedListener)
+ .onComplete(asyncTracker.completeListener)
+ .send(
+ new InputStreamResponseListener() {
+ @Override
+ public void onHeaders(Response response) {
+ super.onHeaders(response);
+ InputStreamResponseListener listener = this;
+ executor.execute(
+ () -> {
+ InputStream is = listener.getInputStream();
+
+ // TODO: Original PR had this, but we do not close the
stream.
+ // The legacy implementation did not track the input
stream,
+ // or close it.
+ // assert ObjectReleaseTracker.track(is);
+
+ try {
+ NamedList<Object> body =
+ processErrorsAndResponse(solrRequest, response,
is, url);
+ mdcCopyHelper.onBegin(null);
+ log.debug("response processing success");
+ future.complete(body);
+ } catch (RemoteSolrException | SolrServerException e) {
mdcCopyHelper.onBegin(null);
log.debug("response processing failed", e);
- asyncListener.onFailure(e);
+ future.completeExceptionally(e);
+ } finally {
+ log.debug("response processing completed");
Review Comment:
so true...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
