You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@solr.apache.org by "jdyer1 (via GitHub)" <gi...@apache.org> on 2026/01/14 22:24:31 UTC
[PR] SOLR-18065: ConcurrentUpdateBaseSolrClient: support for HttpJdkSolrClient [solr]
jdyer1 opened a new pull request, #4050:
URL: https://github.com/apache/solr/pull/4050
This adds `ConcurrentUpdateJdkSolrClient.java` and refactors the main unit tests to test both the Jetty and the JDK clients.
Unlike with `LBAsyncSolrClient` and `CloudHttp2SolrClient`, this new client requires an `HttpJdkSolrClient` and not any subclass of `HttpSolrClientBase`. It did not seem possible to implement the abstract method `ConcurrentUpdateBaseSolrClient#doSendUpdateStream` without requiring the concrete client.
Unlike with `ConcurrentUpdateJettySolrClient`, this implementation does not loop through the queue in `doSendUpdateStream` (the base class does this).
I did not extend the tests `ConcurrentUpdateSolrClientBadInputTest` or `ConcurrentUpdateSolrClientMultiCollectionTest` to test this new client. These are testing the legacy client only.
--
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: issues-unsubscribe@solr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For additional commands, e-mail: issues-help@solr.apache.org
Re: [PR] SOLR-18065: ConcurrentUpdateBaseSolrClient: support for HttpJdkSolrClient [solr]
Posted by "janhoy (via GitHub)" <gi...@apache.org>.
janhoy commented on code in PR #4050:
URL: https://github.com/apache/solr/pull/4050#discussion_r2693720193
##########
solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java:
##########
@@ -0,0 +1,414 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import jakarta.servlet.ServletException;
+import jakarta.servlet.http.HttpServlet;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.invoke.MethodHandles;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.SocketTimeoutException;
+import java.net.http.HttpConnectTimeoutException;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.request.JavaBinUpdateRequestCodec;
+import org.apache.solr.client.solrj.request.SolrQuery;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.SolrNamedThreadFactory;
+import org.apache.solr.embedded.JettyConfig;
+import org.apache.solr.util.SolrJettyTestRule;
+import org.eclipse.jetty.ee10.servlet.ServletHolder;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class ConcurrentUpdateSolrClientTestBase extends SolrTestCaseJ4 {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public abstract HttpSolrClientBase solrClient(Integer overrideIdleTimeoutMs);
+
+ public abstract ConcurrentUpdateBaseSolrClient concurrentClient(
+ HttpSolrClientBase solrClient,
+ String baseUrl,
+ String defaultCollection,
+ int queueSize,
+ int threadCount,
+ boolean disablePollQueue);
+
+ public abstract ConcurrentUpdateBaseSolrClient outcomeCountingConcurrentClient(
+ String serverUrl,
+ int queueSize,
+ int threadCount,
+ HttpSolrClientBase solrClient,
+ AtomicInteger successCounter,
+ AtomicInteger failureCounter,
+ StringBuilder errors);
+
+ /** Mock endpoint where the CUSS being tested in this class sends requests. */
+ public static class TestServlet extends HttpServlet
+ implements JavaBinUpdateRequestCodec.StreamingUpdateHandler {
+ private static final long serialVersionUID = 1L;
+
+ public static void clear() {
+ lastMethod = null;
+ headers = null;
+ parameters = null;
+ errorCode = null;
+ numReqsRcvd.set(0);
+ numDocsRcvd.set(0);
+ }
+
+ public static Integer errorCode = null;
+ public static String lastMethod = null;
+ public static HashMap<String, String> headers = null;
+ public static Map<String, String[]> parameters = null;
+ public static AtomicInteger numReqsRcvd = new AtomicInteger(0);
+ public static AtomicInteger numDocsRcvd = new AtomicInteger(0);
+
+ public static void setErrorCode(Integer code) {
+ errorCode = code;
+ }
+
+ private void setHeaders(HttpServletRequest req) {
+ Enumeration<String> headerNames = req.getHeaderNames();
+ headers = new HashMap<>();
+ while (headerNames.hasMoreElements()) {
+ final String name = headerNames.nextElement();
+ headers.put(name, req.getHeader(name));
+ }
+ }
+
+ private void setParameters(HttpServletRequest req) {
+ // parameters = req.getParameterMap();
+ }
+
+ @Override
+ protected void doPost(HttpServletRequest req, HttpServletResponse resp)
+ throws ServletException, IOException {
+
+ numReqsRcvd.incrementAndGet();
+ lastMethod = "post";
+ recordRequest(req, resp);
+
+ InputStream reqIn = req.getInputStream();
+ JavaBinUpdateRequestCodec javabin = new JavaBinUpdateRequestCodec();
+ for (; ; ) {
+ try {
+ javabin.unmarshal(reqIn, this);
+ } catch (EOFException e) {
+ break; // this is expected
+ }
+ }
+ }
+
+ private void recordRequest(HttpServletRequest req, HttpServletResponse resp) {
+ setHeaders(req);
+ setParameters(req);
+ if (null != errorCode) {
+ try {
+ resp.sendError(errorCode);
+ } catch (IOException e) {
+ throw new RuntimeException("sendError IO fail in TestServlet", e);
+ }
+ }
+ }
+
+ @Override
+ public void update(
+ SolrInputDocument document, UpdateRequest req, Integer commitWithin, Boolean override) {
+ numDocsRcvd.incrementAndGet();
+ }
+ } // end TestServlet
+
+ static class SendDocsRunnable implements Runnable {
+
+ private String id;
+ private int numDocs;
+ private SolrClient cuss;
+ private String collection;
+
+ SendDocsRunnable(String id, int numDocs, SolrClient cuss) {
+ this(id, numDocs, cuss, null);
+ }
+
+ SendDocsRunnable(String id, int numDocs, SolrClient cuss, String collection) {
+ this.id = id;
+ this.numDocs = numDocs;
+ this.cuss = cuss;
+ this.collection = collection;
+ }
+
+ @Override
+ public void run() {
+ for (int d = 0; d < numDocs; d++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ String docId = id + "_" + d;
+ doc.setField("id", docId);
+ UpdateRequest req = new UpdateRequest();
+ req.add(doc);
+ try {
+ if (this.collection == null) cuss.request(req);
+ else cuss.request(req, this.collection);
+ } catch (Throwable t) {
+ log.error("error making request", t);
+ }
+ }
+ }
+ }
+
+ @ClassRule public static SolrJettyTestRule solrTestRule = new SolrJettyTestRule();
+
+ @BeforeClass
+ public static void beforeTest() throws Exception {
+ JettyConfig jettyConfig =
+ JettyConfig.builder().withServlet(new ServletHolder(TestServlet.class), "/cuss/*").build();
+ solrTestRule.startSolr(legacyExampleCollection1SolrHome(), new Properties(), jettyConfig);
+ }
+
+ @AfterClass
+ public static void afterTest() throws Exception {}
+
+ @Test
+ public void testConcurrentUpdate() throws Exception {
+ TestServlet.clear();
+
+ String serverUrl = solrTestRule.getBaseUrl() + "/cuss/foo";
+
+ int cussThreadCount = 2;
+ int cussQueueSize = 100;
+
+ // for tracking callbacks from CUSS
+ final AtomicInteger successCounter = new AtomicInteger(0);
+ final AtomicInteger errorCounter = new AtomicInteger(0);
+ final StringBuilder errors = new StringBuilder();
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ outcomeCountingConcurrentClient(
+ serverUrl,
+ cussQueueSize,
+ cussThreadCount,
+ http2Client,
+ successCounter,
+ errorCounter,
+ errors)) {
+
+ // ensure it doesn't block where there's nothing to do yet
+ concurrentClient.blockUntilFinished();
+
+ int poolSize = 5;
+ ExecutorService threadPool =
+ ExecutorUtil.newMDCAwareFixedThreadPool(poolSize, new SolrNamedThreadFactory("testCUSS"));
+
+ int numDocs = 100;
+ int numRunnables = 5;
+ for (int r = 0; r < numRunnables; r++)
+ threadPool.execute(new SendDocsRunnable(String.valueOf(r), numDocs, concurrentClient));
+
+ // ensure all docs are sent
+ threadPool.awaitTermination(5, TimeUnit.SECONDS);
+ threadPool.shutdown();
+
+ // wait until all requests are processed by CUSS
+ concurrentClient.blockUntilFinished();
+ concurrentClient.shutdownNow();
+
+ assertEquals("post", TestServlet.lastMethod);
+
+ // expect all requests to be successful
+ int expectedSuccesses = TestServlet.numReqsRcvd.get();
+ assertTrue(expectedSuccesses > 0); // at least one request must have been sent
+
+ assertEquals(
+ "Expected no errors but got " + errorCounter.get() + ", due to: " + errors.toString(),
+ 0,
+ errorCounter.get());
+ assertEquals(
+ "Expected " + expectedSuccesses + " successes, but got " + successCounter.get(),
+ successCounter.get(),
+ expectedSuccesses);
+
+ int expectedDocs = numDocs * numRunnables;
+ assertEquals(
+ "Expected CUSS to send " + expectedDocs + " but got " + TestServlet.numDocsRcvd.get(),
+ TestServlet.numDocsRcvd.get(),
+ expectedDocs);
+ }
+ }
+
+ @Test
+ public void testCollectionParameters() throws IOException, SolrServerException {
+
+ int cussThreadCount = 2;
+ int cussQueueSize = 10;
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ null,
+ cussThreadCount,
+ cussQueueSize,
Review Comment:
Swap order: `queueSize, threadCount`
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.http.HttpResponse;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+
+/** A ConcurrentUpdate SolrClient using {@link HttpJdkSolrClient}. */
+public class ConcurrentUpdateJdkSolrClient extends ConcurrentUpdateBaseSolrClient {
+
+ private final HttpJdkSolrClient client;
+
+ protected ConcurrentUpdateJdkSolrClient(ConcurrentUpdateJdkSolrClient.Builder builder) {
+ super(builder);
+ this.client = (HttpJdkSolrClient) builder.getClient();
+ }
+
+ @Override
+ protected StreamingResponse doSendUpdateStream(Update update) {
+ UpdateRequest req = update.request();
+ String collection = update.collection();
+ CompletableFuture<HttpResponse<InputStream>> resp =
+ client.requestInputStreamAsync(basePath, req, collection);
+
+ return new StreamingResponse() {
+
+ @Override
+ public int awaitResponse(long timeoutMillis) throws Exception {
+ return resp.get(timeoutMillis, TimeUnit.MILLISECONDS).statusCode();
+ }
+
+ @Override
+ public InputStream getInputStream() {
+ try {
+ return resp.get().body();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return InputStream.nullInputStream();
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Object getUnderlyingResponse() {
+ return resp;
+ }
+
+ @Override
+ public void close() throws IOException {
+ getInputStream().close();
+ }
+ };
+ }
+
+ public static class Builder extends ConcurrentUpdateBaseSolrClient.Builder {
+ /**
+ * @see org.apache.solr.client.solrj.impl.ConcurrentUpdateBaseSolrClient.Builder#Builder(String,
+ * HttpSolrClientBase)
+ */
+ public Builder(String baseUrl, HttpJdkSolrClient client) {
+
+ this(baseUrl, client, false);
+ // The base class uses idle timeout with StreamingResponse#awaitResponse so it needs to be
+ // set!
+ this.idleTimeoutMillis = 1000;
Review Comment:
Why is this not configurable?
When implementing the tikaserver code, we had a bug where we configured the connectTimeout, but let the idleTimeout at its default value (5s), which was too low. Solr would send the payload to tikaServer, and in some cases, Tika would spend more than 5s parsing the document and the connection became idle. If we need to hardcode a value here, it should likely be >1s
##########
solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java:
##########
@@ -0,0 +1,414 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import jakarta.servlet.ServletException;
+import jakarta.servlet.http.HttpServlet;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.invoke.MethodHandles;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.SocketTimeoutException;
+import java.net.http.HttpConnectTimeoutException;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.request.JavaBinUpdateRequestCodec;
+import org.apache.solr.client.solrj.request.SolrQuery;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.SolrNamedThreadFactory;
+import org.apache.solr.embedded.JettyConfig;
+import org.apache.solr.util.SolrJettyTestRule;
+import org.eclipse.jetty.ee10.servlet.ServletHolder;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class ConcurrentUpdateSolrClientTestBase extends SolrTestCaseJ4 {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public abstract HttpSolrClientBase solrClient(Integer overrideIdleTimeoutMs);
+
+ public abstract ConcurrentUpdateBaseSolrClient concurrentClient(
+ HttpSolrClientBase solrClient,
+ String baseUrl,
+ String defaultCollection,
+ int queueSize,
+ int threadCount,
+ boolean disablePollQueue);
+
+ public abstract ConcurrentUpdateBaseSolrClient outcomeCountingConcurrentClient(
+ String serverUrl,
+ int queueSize,
+ int threadCount,
+ HttpSolrClientBase solrClient,
+ AtomicInteger successCounter,
+ AtomicInteger failureCounter,
+ StringBuilder errors);
+
+ /** Mock endpoint where the CUSS being tested in this class sends requests. */
+ public static class TestServlet extends HttpServlet
+ implements JavaBinUpdateRequestCodec.StreamingUpdateHandler {
+ private static final long serialVersionUID = 1L;
+
+ public static void clear() {
+ lastMethod = null;
+ headers = null;
+ parameters = null;
+ errorCode = null;
+ numReqsRcvd.set(0);
+ numDocsRcvd.set(0);
+ }
+
+ public static Integer errorCode = null;
+ public static String lastMethod = null;
+ public static HashMap<String, String> headers = null;
+ public static Map<String, String[]> parameters = null;
+ public static AtomicInteger numReqsRcvd = new AtomicInteger(0);
+ public static AtomicInteger numDocsRcvd = new AtomicInteger(0);
+
+ public static void setErrorCode(Integer code) {
+ errorCode = code;
+ }
+
+ private void setHeaders(HttpServletRequest req) {
+ Enumeration<String> headerNames = req.getHeaderNames();
+ headers = new HashMap<>();
+ while (headerNames.hasMoreElements()) {
+ final String name = headerNames.nextElement();
+ headers.put(name, req.getHeader(name));
+ }
+ }
+
+ private void setParameters(HttpServletRequest req) {
+ // parameters = req.getParameterMap();
+ }
+
+ @Override
+ protected void doPost(HttpServletRequest req, HttpServletResponse resp)
+ throws ServletException, IOException {
+
+ numReqsRcvd.incrementAndGet();
+ lastMethod = "post";
+ recordRequest(req, resp);
+
+ InputStream reqIn = req.getInputStream();
+ JavaBinUpdateRequestCodec javabin = new JavaBinUpdateRequestCodec();
+ for (; ; ) {
+ try {
+ javabin.unmarshal(reqIn, this);
+ } catch (EOFException e) {
+ break; // this is expected
+ }
+ }
+ }
+
+ private void recordRequest(HttpServletRequest req, HttpServletResponse resp) {
+ setHeaders(req);
+ setParameters(req);
+ if (null != errorCode) {
+ try {
+ resp.sendError(errorCode);
+ } catch (IOException e) {
+ throw new RuntimeException("sendError IO fail in TestServlet", e);
+ }
+ }
+ }
+
+ @Override
+ public void update(
+ SolrInputDocument document, UpdateRequest req, Integer commitWithin, Boolean override) {
+ numDocsRcvd.incrementAndGet();
+ }
+ } // end TestServlet
+
+ static class SendDocsRunnable implements Runnable {
+
+ private String id;
+ private int numDocs;
+ private SolrClient cuss;
+ private String collection;
+
+ SendDocsRunnable(String id, int numDocs, SolrClient cuss) {
+ this(id, numDocs, cuss, null);
+ }
+
+ SendDocsRunnable(String id, int numDocs, SolrClient cuss, String collection) {
+ this.id = id;
+ this.numDocs = numDocs;
+ this.cuss = cuss;
+ this.collection = collection;
+ }
+
+ @Override
+ public void run() {
+ for (int d = 0; d < numDocs; d++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ String docId = id + "_" + d;
+ doc.setField("id", docId);
+ UpdateRequest req = new UpdateRequest();
+ req.add(doc);
+ try {
+ if (this.collection == null) cuss.request(req);
+ else cuss.request(req, this.collection);
+ } catch (Throwable t) {
+ log.error("error making request", t);
+ }
+ }
+ }
+ }
+
+ @ClassRule public static SolrJettyTestRule solrTestRule = new SolrJettyTestRule();
+
+ @BeforeClass
+ public static void beforeTest() throws Exception {
+ JettyConfig jettyConfig =
+ JettyConfig.builder().withServlet(new ServletHolder(TestServlet.class), "/cuss/*").build();
+ solrTestRule.startSolr(legacyExampleCollection1SolrHome(), new Properties(), jettyConfig);
+ }
+
+ @AfterClass
+ public static void afterTest() throws Exception {}
+
+ @Test
+ public void testConcurrentUpdate() throws Exception {
+ TestServlet.clear();
+
+ String serverUrl = solrTestRule.getBaseUrl() + "/cuss/foo";
+
+ int cussThreadCount = 2;
+ int cussQueueSize = 100;
+
+ // for tracking callbacks from CUSS
+ final AtomicInteger successCounter = new AtomicInteger(0);
+ final AtomicInteger errorCounter = new AtomicInteger(0);
+ final StringBuilder errors = new StringBuilder();
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ outcomeCountingConcurrentClient(
+ serverUrl,
+ cussQueueSize,
+ cussThreadCount,
+ http2Client,
+ successCounter,
+ errorCounter,
+ errors)) {
+
+ // ensure it doesn't block where there's nothing to do yet
+ concurrentClient.blockUntilFinished();
+
+ int poolSize = 5;
+ ExecutorService threadPool =
+ ExecutorUtil.newMDCAwareFixedThreadPool(poolSize, new SolrNamedThreadFactory("testCUSS"));
+
+ int numDocs = 100;
+ int numRunnables = 5;
+ for (int r = 0; r < numRunnables; r++)
+ threadPool.execute(new SendDocsRunnable(String.valueOf(r), numDocs, concurrentClient));
+
+ // ensure all docs are sent
+ threadPool.awaitTermination(5, TimeUnit.SECONDS);
+ threadPool.shutdown();
+
+ // wait until all requests are processed by CUSS
+ concurrentClient.blockUntilFinished();
+ concurrentClient.shutdownNow();
+
+ assertEquals("post", TestServlet.lastMethod);
+
+ // expect all requests to be successful
+ int expectedSuccesses = TestServlet.numReqsRcvd.get();
+ assertTrue(expectedSuccesses > 0); // at least one request must have been sent
+
+ assertEquals(
+ "Expected no errors but got " + errorCounter.get() + ", due to: " + errors.toString(),
+ 0,
+ errorCounter.get());
+ assertEquals(
+ "Expected " + expectedSuccesses + " successes, but got " + successCounter.get(),
+ successCounter.get(),
+ expectedSuccesses);
+
+ int expectedDocs = numDocs * numRunnables;
+ assertEquals(
+ "Expected CUSS to send " + expectedDocs + " but got " + TestServlet.numDocsRcvd.get(),
+ TestServlet.numDocsRcvd.get(),
+ expectedDocs);
+ }
+ }
+
+ @Test
+ public void testCollectionParameters() throws IOException, SolrServerException {
+
+ int cussThreadCount = 2;
+ int cussQueueSize = 10;
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ null,
+ cussThreadCount,
+ cussQueueSize,
+ false)) {
+
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", "collection");
+ concurrentClient.add("collection1", doc);
+ concurrentClient.commit("collection1");
+
+ assertEquals(
+ 1,
+ concurrentClient
+ .query("collection1", new SolrQuery("id:collection"))
+ .getResults()
+ .getNumFound());
+ }
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ DEFAULT_TEST_CORENAME,
+ cussThreadCount,
+ cussQueueSize,
+ false)) {
+
+ assertEquals(
+ 1, concurrentClient.query(new SolrQuery("id:collection")).getResults().getNumFound());
+ }
+ }
+
+ @Test
+ public void testConcurrentCollectionUpdate() throws Exception {
+
+ int cussThreadCount = 2;
+ int cussQueueSize = 100;
+ int numDocs = 100;
+ int numRunnables = 5;
+ int expected = numDocs * numRunnables;
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ null,
+ cussThreadCount,
+ cussQueueSize,
+ true)) {
+
+ // ensure it doesn't block where there's nothing to do yet
+ concurrentClient.blockUntilFinished();
+
+ // Delete all existing documents.
+ concurrentClient.deleteByQuery("collection1", "*:*");
+
+ int poolSize = 5;
+ ExecutorService threadPool =
+ ExecutorUtil.newMDCAwareFixedThreadPool(poolSize, new SolrNamedThreadFactory("testCUSS"));
+
+ for (int r = 0; r < numRunnables; r++)
+ threadPool.execute(
+ new SendDocsRunnable(String.valueOf(r), numDocs, concurrentClient, "collection1"));
+
+ // ensure all docs are sent
+ threadPool.awaitTermination(5, TimeUnit.SECONDS);
+ threadPool.shutdown();
+
+ concurrentClient.commit("collection1");
+
+ assertEquals(
+ expected,
+ concurrentClient.query("collection1", new SolrQuery("*:*")).getResults().getNumFound());
+
+ // wait until all requests are processed by CUSS
+ concurrentClient.blockUntilFinished();
+ concurrentClient.shutdownNow();
+ }
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ DEFAULT_TEST_CORENAME,
+ cussThreadCount,
+ cussQueueSize,
+ false)) {
+
+ assertEquals(
+ expected, concurrentClient.query(new SolrQuery("*:*")).getResults().getNumFound());
+ }
+ }
+
+ /**
+ * Test that connection timeout information is passed to the HttpSolrClient that handles non add
+ * operations.
+ */
+ @Test(timeout = 10000)
+ public void testSocketTimeoutOnCommit() throws IOException, SolrServerException {
+ InetAddress localHost = InetAddress.getLocalHost();
+ try (ServerSocket server = new ServerSocket(0, 1, localHost);
+ var http2Client = solrClient(1);
+ var client =
+ concurrentClient(
+ http2Client,
+ "http://"
+ + localHost.getHostAddress()
+ + ":"
+ + server.getLocalPort()
+ + "/noOneThere",
+ null,
+ 10,
+ 1,
Review Comment:
Correct order `queueSize=10, threadCount=1`?
##########
solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java:
##########
@@ -0,0 +1,414 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import jakarta.servlet.ServletException;
+import jakarta.servlet.http.HttpServlet;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.invoke.MethodHandles;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.SocketTimeoutException;
+import java.net.http.HttpConnectTimeoutException;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.request.JavaBinUpdateRequestCodec;
+import org.apache.solr.client.solrj.request.SolrQuery;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.SolrNamedThreadFactory;
+import org.apache.solr.embedded.JettyConfig;
+import org.apache.solr.util.SolrJettyTestRule;
+import org.eclipse.jetty.ee10.servlet.ServletHolder;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class ConcurrentUpdateSolrClientTestBase extends SolrTestCaseJ4 {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public abstract HttpSolrClientBase solrClient(Integer overrideIdleTimeoutMs);
+
+ public abstract ConcurrentUpdateBaseSolrClient concurrentClient(
+ HttpSolrClientBase solrClient,
+ String baseUrl,
+ String defaultCollection,
+ int queueSize,
+ int threadCount,
+ boolean disablePollQueue);
+
+ public abstract ConcurrentUpdateBaseSolrClient outcomeCountingConcurrentClient(
+ String serverUrl,
+ int queueSize,
+ int threadCount,
+ HttpSolrClientBase solrClient,
+ AtomicInteger successCounter,
+ AtomicInteger failureCounter,
+ StringBuilder errors);
+
+ /** Mock endpoint where the CUSS being tested in this class sends requests. */
+ public static class TestServlet extends HttpServlet
+ implements JavaBinUpdateRequestCodec.StreamingUpdateHandler {
+ private static final long serialVersionUID = 1L;
+
+ public static void clear() {
+ lastMethod = null;
+ headers = null;
+ parameters = null;
+ errorCode = null;
+ numReqsRcvd.set(0);
+ numDocsRcvd.set(0);
+ }
+
+ public static Integer errorCode = null;
+ public static String lastMethod = null;
+ public static HashMap<String, String> headers = null;
+ public static Map<String, String[]> parameters = null;
+ public static AtomicInteger numReqsRcvd = new AtomicInteger(0);
+ public static AtomicInteger numDocsRcvd = new AtomicInteger(0);
+
+ public static void setErrorCode(Integer code) {
+ errorCode = code;
+ }
+
+ private void setHeaders(HttpServletRequest req) {
+ Enumeration<String> headerNames = req.getHeaderNames();
+ headers = new HashMap<>();
+ while (headerNames.hasMoreElements()) {
+ final String name = headerNames.nextElement();
+ headers.put(name, req.getHeader(name));
+ }
+ }
+
+ private void setParameters(HttpServletRequest req) {
+ // parameters = req.getParameterMap();
+ }
+
+ @Override
+ protected void doPost(HttpServletRequest req, HttpServletResponse resp)
+ throws ServletException, IOException {
+
+ numReqsRcvd.incrementAndGet();
+ lastMethod = "post";
+ recordRequest(req, resp);
+
+ InputStream reqIn = req.getInputStream();
+ JavaBinUpdateRequestCodec javabin = new JavaBinUpdateRequestCodec();
+ for (; ; ) {
+ try {
+ javabin.unmarshal(reqIn, this);
+ } catch (EOFException e) {
+ break; // this is expected
+ }
+ }
+ }
+
+ private void recordRequest(HttpServletRequest req, HttpServletResponse resp) {
+ setHeaders(req);
+ setParameters(req);
+ if (null != errorCode) {
+ try {
+ resp.sendError(errorCode);
+ } catch (IOException e) {
+ throw new RuntimeException("sendError IO fail in TestServlet", e);
+ }
+ }
+ }
+
+ @Override
+ public void update(
+ SolrInputDocument document, UpdateRequest req, Integer commitWithin, Boolean override) {
+ numDocsRcvd.incrementAndGet();
+ }
+ } // end TestServlet
+
+ static class SendDocsRunnable implements Runnable {
+
+ private String id;
+ private int numDocs;
+ private SolrClient cuss;
+ private String collection;
+
+ SendDocsRunnable(String id, int numDocs, SolrClient cuss) {
+ this(id, numDocs, cuss, null);
+ }
+
+ SendDocsRunnable(String id, int numDocs, SolrClient cuss, String collection) {
+ this.id = id;
+ this.numDocs = numDocs;
+ this.cuss = cuss;
+ this.collection = collection;
+ }
+
+ @Override
+ public void run() {
+ for (int d = 0; d < numDocs; d++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ String docId = id + "_" + d;
+ doc.setField("id", docId);
+ UpdateRequest req = new UpdateRequest();
+ req.add(doc);
+ try {
+ if (this.collection == null) cuss.request(req);
+ else cuss.request(req, this.collection);
+ } catch (Throwable t) {
+ log.error("error making request", t);
+ }
+ }
+ }
+ }
+
+ @ClassRule public static SolrJettyTestRule solrTestRule = new SolrJettyTestRule();
+
+ @BeforeClass
+ public static void beforeTest() throws Exception {
+ JettyConfig jettyConfig =
+ JettyConfig.builder().withServlet(new ServletHolder(TestServlet.class), "/cuss/*").build();
+ solrTestRule.startSolr(legacyExampleCollection1SolrHome(), new Properties(), jettyConfig);
+ }
+
+ @AfterClass
+ public static void afterTest() throws Exception {}
+
+ @Test
+ public void testConcurrentUpdate() throws Exception {
+ TestServlet.clear();
+
+ String serverUrl = solrTestRule.getBaseUrl() + "/cuss/foo";
+
+ int cussThreadCount = 2;
+ int cussQueueSize = 100;
+
+ // for tracking callbacks from CUSS
+ final AtomicInteger successCounter = new AtomicInteger(0);
+ final AtomicInteger errorCounter = new AtomicInteger(0);
+ final StringBuilder errors = new StringBuilder();
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ outcomeCountingConcurrentClient(
+ serverUrl,
+ cussQueueSize,
+ cussThreadCount,
+ http2Client,
+ successCounter,
+ errorCounter,
+ errors)) {
+
+ // ensure it doesn't block where there's nothing to do yet
+ concurrentClient.blockUntilFinished();
+
+ int poolSize = 5;
+ ExecutorService threadPool =
+ ExecutorUtil.newMDCAwareFixedThreadPool(poolSize, new SolrNamedThreadFactory("testCUSS"));
+
+ int numDocs = 100;
+ int numRunnables = 5;
+ for (int r = 0; r < numRunnables; r++)
+ threadPool.execute(new SendDocsRunnable(String.valueOf(r), numDocs, concurrentClient));
+
+ // ensure all docs are sent
+ threadPool.awaitTermination(5, TimeUnit.SECONDS);
+ threadPool.shutdown();
+
+ // wait until all requests are processed by CUSS
+ concurrentClient.blockUntilFinished();
+ concurrentClient.shutdownNow();
+
+ assertEquals("post", TestServlet.lastMethod);
+
+ // expect all requests to be successful
+ int expectedSuccesses = TestServlet.numReqsRcvd.get();
+ assertTrue(expectedSuccesses > 0); // at least one request must have been sent
+
+ assertEquals(
+ "Expected no errors but got " + errorCounter.get() + ", due to: " + errors.toString(),
+ 0,
+ errorCounter.get());
+ assertEquals(
+ "Expected " + expectedSuccesses + " successes, but got " + successCounter.get(),
+ successCounter.get(),
+ expectedSuccesses);
+
+ int expectedDocs = numDocs * numRunnables;
+ assertEquals(
+ "Expected CUSS to send " + expectedDocs + " but got " + TestServlet.numDocsRcvd.get(),
+ TestServlet.numDocsRcvd.get(),
+ expectedDocs);
+ }
+ }
+
+ @Test
+ public void testCollectionParameters() throws IOException, SolrServerException {
+
+ int cussThreadCount = 2;
+ int cussQueueSize = 10;
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ null,
+ cussThreadCount,
+ cussQueueSize,
+ false)) {
+
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", "collection");
+ concurrentClient.add("collection1", doc);
+ concurrentClient.commit("collection1");
+
+ assertEquals(
+ 1,
+ concurrentClient
+ .query("collection1", new SolrQuery("id:collection"))
+ .getResults()
+ .getNumFound());
+ }
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ DEFAULT_TEST_CORENAME,
+ cussThreadCount,
+ cussQueueSize,
Review Comment:
Swap order: `queueSize, threadCount`
##########
solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java:
##########
@@ -0,0 +1,414 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import jakarta.servlet.ServletException;
+import jakarta.servlet.http.HttpServlet;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.invoke.MethodHandles;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.SocketTimeoutException;
+import java.net.http.HttpConnectTimeoutException;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.request.JavaBinUpdateRequestCodec;
+import org.apache.solr.client.solrj.request.SolrQuery;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.SolrNamedThreadFactory;
+import org.apache.solr.embedded.JettyConfig;
+import org.apache.solr.util.SolrJettyTestRule;
+import org.eclipse.jetty.ee10.servlet.ServletHolder;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class ConcurrentUpdateSolrClientTestBase extends SolrTestCaseJ4 {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public abstract HttpSolrClientBase solrClient(Integer overrideIdleTimeoutMs);
+
+ public abstract ConcurrentUpdateBaseSolrClient concurrentClient(
+ HttpSolrClientBase solrClient,
+ String baseUrl,
+ String defaultCollection,
+ int queueSize,
+ int threadCount,
+ boolean disablePollQueue);
+
+ public abstract ConcurrentUpdateBaseSolrClient outcomeCountingConcurrentClient(
+ String serverUrl,
+ int queueSize,
+ int threadCount,
+ HttpSolrClientBase solrClient,
+ AtomicInteger successCounter,
+ AtomicInteger failureCounter,
+ StringBuilder errors);
+
+ /** Mock endpoint where the CUSS being tested in this class sends requests. */
+ public static class TestServlet extends HttpServlet
+ implements JavaBinUpdateRequestCodec.StreamingUpdateHandler {
+ private static final long serialVersionUID = 1L;
+
+ public static void clear() {
+ lastMethod = null;
+ headers = null;
+ parameters = null;
+ errorCode = null;
+ numReqsRcvd.set(0);
+ numDocsRcvd.set(0);
+ }
+
+ public static Integer errorCode = null;
+ public static String lastMethod = null;
+ public static HashMap<String, String> headers = null;
+ public static Map<String, String[]> parameters = null;
+ public static AtomicInteger numReqsRcvd = new AtomicInteger(0);
+ public static AtomicInteger numDocsRcvd = new AtomicInteger(0);
+
+ public static void setErrorCode(Integer code) {
+ errorCode = code;
+ }
+
+ private void setHeaders(HttpServletRequest req) {
+ Enumeration<String> headerNames = req.getHeaderNames();
+ headers = new HashMap<>();
+ while (headerNames.hasMoreElements()) {
+ final String name = headerNames.nextElement();
+ headers.put(name, req.getHeader(name));
+ }
+ }
+
+ private void setParameters(HttpServletRequest req) {
+ // parameters = req.getParameterMap();
+ }
+
+ @Override
+ protected void doPost(HttpServletRequest req, HttpServletResponse resp)
+ throws ServletException, IOException {
+
+ numReqsRcvd.incrementAndGet();
+ lastMethod = "post";
+ recordRequest(req, resp);
+
+ InputStream reqIn = req.getInputStream();
+ JavaBinUpdateRequestCodec javabin = new JavaBinUpdateRequestCodec();
+ for (; ; ) {
+ try {
+ javabin.unmarshal(reqIn, this);
+ } catch (EOFException e) {
+ break; // this is expected
+ }
+ }
+ }
+
+ private void recordRequest(HttpServletRequest req, HttpServletResponse resp) {
+ setHeaders(req);
+ setParameters(req);
+ if (null != errorCode) {
+ try {
+ resp.sendError(errorCode);
+ } catch (IOException e) {
+ throw new RuntimeException("sendError IO fail in TestServlet", e);
+ }
+ }
+ }
+
+ @Override
+ public void update(
+ SolrInputDocument document, UpdateRequest req, Integer commitWithin, Boolean override) {
+ numDocsRcvd.incrementAndGet();
+ }
+ } // end TestServlet
+
+ static class SendDocsRunnable implements Runnable {
+
+ private String id;
+ private int numDocs;
+ private SolrClient cuss;
+ private String collection;
+
+ SendDocsRunnable(String id, int numDocs, SolrClient cuss) {
+ this(id, numDocs, cuss, null);
+ }
+
+ SendDocsRunnable(String id, int numDocs, SolrClient cuss, String collection) {
+ this.id = id;
+ this.numDocs = numDocs;
+ this.cuss = cuss;
+ this.collection = collection;
+ }
+
+ @Override
+ public void run() {
+ for (int d = 0; d < numDocs; d++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ String docId = id + "_" + d;
+ doc.setField("id", docId);
+ UpdateRequest req = new UpdateRequest();
+ req.add(doc);
+ try {
+ if (this.collection == null) cuss.request(req);
+ else cuss.request(req, this.collection);
+ } catch (Throwable t) {
+ log.error("error making request", t);
+ }
+ }
+ }
+ }
+
+ @ClassRule public static SolrJettyTestRule solrTestRule = new SolrJettyTestRule();
+
+ @BeforeClass
+ public static void beforeTest() throws Exception {
+ JettyConfig jettyConfig =
+ JettyConfig.builder().withServlet(new ServletHolder(TestServlet.class), "/cuss/*").build();
+ solrTestRule.startSolr(legacyExampleCollection1SolrHome(), new Properties(), jettyConfig);
+ }
+
+ @AfterClass
+ public static void afterTest() throws Exception {}
+
+ @Test
+ public void testConcurrentUpdate() throws Exception {
+ TestServlet.clear();
+
+ String serverUrl = solrTestRule.getBaseUrl() + "/cuss/foo";
+
+ int cussThreadCount = 2;
+ int cussQueueSize = 100;
+
+ // for tracking callbacks from CUSS
+ final AtomicInteger successCounter = new AtomicInteger(0);
+ final AtomicInteger errorCounter = new AtomicInteger(0);
+ final StringBuilder errors = new StringBuilder();
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ outcomeCountingConcurrentClient(
+ serverUrl,
+ cussQueueSize,
+ cussThreadCount,
+ http2Client,
+ successCounter,
+ errorCounter,
+ errors)) {
+
+ // ensure it doesn't block where there's nothing to do yet
+ concurrentClient.blockUntilFinished();
+
+ int poolSize = 5;
+ ExecutorService threadPool =
+ ExecutorUtil.newMDCAwareFixedThreadPool(poolSize, new SolrNamedThreadFactory("testCUSS"));
+
+ int numDocs = 100;
+ int numRunnables = 5;
+ for (int r = 0; r < numRunnables; r++)
+ threadPool.execute(new SendDocsRunnable(String.valueOf(r), numDocs, concurrentClient));
+
+ // ensure all docs are sent
+ threadPool.awaitTermination(5, TimeUnit.SECONDS);
+ threadPool.shutdown();
+
+ // wait until all requests are processed by CUSS
+ concurrentClient.blockUntilFinished();
+ concurrentClient.shutdownNow();
+
+ assertEquals("post", TestServlet.lastMethod);
+
+ // expect all requests to be successful
+ int expectedSuccesses = TestServlet.numReqsRcvd.get();
+ assertTrue(expectedSuccesses > 0); // at least one request must have been sent
+
+ assertEquals(
+ "Expected no errors but got " + errorCounter.get() + ", due to: " + errors.toString(),
+ 0,
+ errorCounter.get());
+ assertEquals(
+ "Expected " + expectedSuccesses + " successes, but got " + successCounter.get(),
+ successCounter.get(),
+ expectedSuccesses);
+
+ int expectedDocs = numDocs * numRunnables;
+ assertEquals(
+ "Expected CUSS to send " + expectedDocs + " but got " + TestServlet.numDocsRcvd.get(),
+ TestServlet.numDocsRcvd.get(),
+ expectedDocs);
+ }
+ }
+
+ @Test
+ public void testCollectionParameters() throws IOException, SolrServerException {
+
+ int cussThreadCount = 2;
+ int cussQueueSize = 10;
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ null,
+ cussThreadCount,
+ cussQueueSize,
+ false)) {
+
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", "collection");
+ concurrentClient.add("collection1", doc);
+ concurrentClient.commit("collection1");
+
+ assertEquals(
+ 1,
+ concurrentClient
+ .query("collection1", new SolrQuery("id:collection"))
+ .getResults()
+ .getNumFound());
+ }
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ DEFAULT_TEST_CORENAME,
+ cussThreadCount,
+ cussQueueSize,
+ false)) {
+
+ assertEquals(
+ 1, concurrentClient.query(new SolrQuery("id:collection")).getResults().getNumFound());
+ }
+ }
+
+ @Test
+ public void testConcurrentCollectionUpdate() throws Exception {
+
+ int cussThreadCount = 2;
+ int cussQueueSize = 100;
+ int numDocs = 100;
+ int numRunnables = 5;
+ int expected = numDocs * numRunnables;
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ null,
+ cussThreadCount,
+ cussQueueSize,
+ true)) {
+
+ // ensure it doesn't block where there's nothing to do yet
+ concurrentClient.blockUntilFinished();
+
+ // Delete all existing documents.
+ concurrentClient.deleteByQuery("collection1", "*:*");
+
+ int poolSize = 5;
+ ExecutorService threadPool =
+ ExecutorUtil.newMDCAwareFixedThreadPool(poolSize, new SolrNamedThreadFactory("testCUSS"));
+
+ for (int r = 0; r < numRunnables; r++)
+ threadPool.execute(
+ new SendDocsRunnable(String.valueOf(r), numDocs, concurrentClient, "collection1"));
+
+ // ensure all docs are sent
+ threadPool.awaitTermination(5, TimeUnit.SECONDS);
+ threadPool.shutdown();
+
+ concurrentClient.commit("collection1");
+
+ assertEquals(
+ expected,
+ concurrentClient.query("collection1", new SolrQuery("*:*")).getResults().getNumFound());
+
+ // wait until all requests are processed by CUSS
+ concurrentClient.blockUntilFinished();
+ concurrentClient.shutdownNow();
+ }
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ DEFAULT_TEST_CORENAME,
+ cussThreadCount,
+ cussQueueSize,
Review Comment:
Swap order: `queueSize, threadCount`
##########
solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java:
##########
@@ -0,0 +1,414 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import jakarta.servlet.ServletException;
+import jakarta.servlet.http.HttpServlet;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.invoke.MethodHandles;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.SocketTimeoutException;
+import java.net.http.HttpConnectTimeoutException;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.request.JavaBinUpdateRequestCodec;
+import org.apache.solr.client.solrj.request.SolrQuery;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.SolrNamedThreadFactory;
+import org.apache.solr.embedded.JettyConfig;
+import org.apache.solr.util.SolrJettyTestRule;
+import org.eclipse.jetty.ee10.servlet.ServletHolder;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class ConcurrentUpdateSolrClientTestBase extends SolrTestCaseJ4 {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public abstract HttpSolrClientBase solrClient(Integer overrideIdleTimeoutMs);
+
+ public abstract ConcurrentUpdateBaseSolrClient concurrentClient(
+ HttpSolrClientBase solrClient,
+ String baseUrl,
+ String defaultCollection,
+ int queueSize,
+ int threadCount,
+ boolean disablePollQueue);
+
+ public abstract ConcurrentUpdateBaseSolrClient outcomeCountingConcurrentClient(
+ String serverUrl,
+ int queueSize,
+ int threadCount,
+ HttpSolrClientBase solrClient,
+ AtomicInteger successCounter,
+ AtomicInteger failureCounter,
+ StringBuilder errors);
+
+ /** Mock endpoint where the CUSS being tested in this class sends requests. */
+ public static class TestServlet extends HttpServlet
+ implements JavaBinUpdateRequestCodec.StreamingUpdateHandler {
+ private static final long serialVersionUID = 1L;
+
+ public static void clear() {
+ lastMethod = null;
+ headers = null;
+ parameters = null;
+ errorCode = null;
+ numReqsRcvd.set(0);
+ numDocsRcvd.set(0);
+ }
+
+ public static Integer errorCode = null;
+ public static String lastMethod = null;
+ public static HashMap<String, String> headers = null;
+ public static Map<String, String[]> parameters = null;
+ public static AtomicInteger numReqsRcvd = new AtomicInteger(0);
+ public static AtomicInteger numDocsRcvd = new AtomicInteger(0);
+
+ public static void setErrorCode(Integer code) {
+ errorCode = code;
+ }
+
+ private void setHeaders(HttpServletRequest req) {
+ Enumeration<String> headerNames = req.getHeaderNames();
+ headers = new HashMap<>();
+ while (headerNames.hasMoreElements()) {
+ final String name = headerNames.nextElement();
+ headers.put(name, req.getHeader(name));
+ }
+ }
+
+ private void setParameters(HttpServletRequest req) {
+ // parameters = req.getParameterMap();
+ }
+
+ @Override
+ protected void doPost(HttpServletRequest req, HttpServletResponse resp)
+ throws ServletException, IOException {
+
+ numReqsRcvd.incrementAndGet();
+ lastMethod = "post";
+ recordRequest(req, resp);
+
+ InputStream reqIn = req.getInputStream();
+ JavaBinUpdateRequestCodec javabin = new JavaBinUpdateRequestCodec();
+ for (; ; ) {
+ try {
+ javabin.unmarshal(reqIn, this);
+ } catch (EOFException e) {
+ break; // this is expected
+ }
+ }
+ }
+
+ private void recordRequest(HttpServletRequest req, HttpServletResponse resp) {
+ setHeaders(req);
+ setParameters(req);
+ if (null != errorCode) {
+ try {
+ resp.sendError(errorCode);
+ } catch (IOException e) {
+ throw new RuntimeException("sendError IO fail in TestServlet", e);
+ }
+ }
+ }
+
+ @Override
+ public void update(
+ SolrInputDocument document, UpdateRequest req, Integer commitWithin, Boolean override) {
+ numDocsRcvd.incrementAndGet();
+ }
+ } // end TestServlet
+
+ static class SendDocsRunnable implements Runnable {
+
+ private String id;
+ private int numDocs;
+ private SolrClient cuss;
+ private String collection;
+
+ SendDocsRunnable(String id, int numDocs, SolrClient cuss) {
+ this(id, numDocs, cuss, null);
+ }
+
+ SendDocsRunnable(String id, int numDocs, SolrClient cuss, String collection) {
+ this.id = id;
+ this.numDocs = numDocs;
+ this.cuss = cuss;
+ this.collection = collection;
+ }
+
+ @Override
+ public void run() {
+ for (int d = 0; d < numDocs; d++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ String docId = id + "_" + d;
+ doc.setField("id", docId);
+ UpdateRequest req = new UpdateRequest();
+ req.add(doc);
+ try {
+ if (this.collection == null) cuss.request(req);
+ else cuss.request(req, this.collection);
+ } catch (Throwable t) {
+ log.error("error making request", t);
+ }
+ }
+ }
+ }
+
+ @ClassRule public static SolrJettyTestRule solrTestRule = new SolrJettyTestRule();
+
+ @BeforeClass
+ public static void beforeTest() throws Exception {
+ JettyConfig jettyConfig =
+ JettyConfig.builder().withServlet(new ServletHolder(TestServlet.class), "/cuss/*").build();
+ solrTestRule.startSolr(legacyExampleCollection1SolrHome(), new Properties(), jettyConfig);
+ }
+
+ @AfterClass
+ public static void afterTest() throws Exception {}
+
+ @Test
+ public void testConcurrentUpdate() throws Exception {
+ TestServlet.clear();
+
+ String serverUrl = solrTestRule.getBaseUrl() + "/cuss/foo";
+
+ int cussThreadCount = 2;
+ int cussQueueSize = 100;
+
+ // for tracking callbacks from CUSS
+ final AtomicInteger successCounter = new AtomicInteger(0);
+ final AtomicInteger errorCounter = new AtomicInteger(0);
+ final StringBuilder errors = new StringBuilder();
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ outcomeCountingConcurrentClient(
+ serverUrl,
+ cussQueueSize,
+ cussThreadCount,
+ http2Client,
+ successCounter,
+ errorCounter,
+ errors)) {
+
+ // ensure it doesn't block where there's nothing to do yet
+ concurrentClient.blockUntilFinished();
+
+ int poolSize = 5;
+ ExecutorService threadPool =
+ ExecutorUtil.newMDCAwareFixedThreadPool(poolSize, new SolrNamedThreadFactory("testCUSS"));
+
+ int numDocs = 100;
+ int numRunnables = 5;
+ for (int r = 0; r < numRunnables; r++)
+ threadPool.execute(new SendDocsRunnable(String.valueOf(r), numDocs, concurrentClient));
+
+ // ensure all docs are sent
+ threadPool.awaitTermination(5, TimeUnit.SECONDS);
+ threadPool.shutdown();
+
+ // wait until all requests are processed by CUSS
+ concurrentClient.blockUntilFinished();
+ concurrentClient.shutdownNow();
+
+ assertEquals("post", TestServlet.lastMethod);
+
+ // expect all requests to be successful
+ int expectedSuccesses = TestServlet.numReqsRcvd.get();
+ assertTrue(expectedSuccesses > 0); // at least one request must have been sent
+
+ assertEquals(
+ "Expected no errors but got " + errorCounter.get() + ", due to: " + errors.toString(),
+ 0,
+ errorCounter.get());
+ assertEquals(
+ "Expected " + expectedSuccesses + " successes, but got " + successCounter.get(),
+ successCounter.get(),
+ expectedSuccesses);
+
+ int expectedDocs = numDocs * numRunnables;
+ assertEquals(
+ "Expected CUSS to send " + expectedDocs + " but got " + TestServlet.numDocsRcvd.get(),
+ TestServlet.numDocsRcvd.get(),
+ expectedDocs);
+ }
+ }
+
+ @Test
+ public void testCollectionParameters() throws IOException, SolrServerException {
+
+ int cussThreadCount = 2;
+ int cussQueueSize = 10;
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ null,
+ cussThreadCount,
+ cussQueueSize,
+ false)) {
+
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", "collection");
+ concurrentClient.add("collection1", doc);
+ concurrentClient.commit("collection1");
+
+ assertEquals(
+ 1,
+ concurrentClient
+ .query("collection1", new SolrQuery("id:collection"))
+ .getResults()
+ .getNumFound());
+ }
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ DEFAULT_TEST_CORENAME,
+ cussThreadCount,
+ cussQueueSize,
+ false)) {
+
+ assertEquals(
+ 1, concurrentClient.query(new SolrQuery("id:collection")).getResults().getNumFound());
+ }
+ }
+
+ @Test
+ public void testConcurrentCollectionUpdate() throws Exception {
+
+ int cussThreadCount = 2;
+ int cussQueueSize = 100;
+ int numDocs = 100;
+ int numRunnables = 5;
+ int expected = numDocs * numRunnables;
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ null,
+ cussThreadCount,
+ cussQueueSize,
Review Comment:
Swap order: `queueSize, threadCount`
--
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: issues-unsubscribe@solr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For additional commands, e-mail: issues-help@solr.apache.org
Re: [PR] SOLR-18065: ConcurrentUpdateBaseSolrClient: support for HttpJdkSolrClient [solr]
Posted by "jdyer1 (via GitHub)" <gi...@apache.org>.
jdyer1 commented on code in PR #4050:
URL: https://github.com/apache/solr/pull/4050#discussion_r2695097774
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.http.HttpResponse;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+
+/** A ConcurrentUpdate SolrClient using {@link HttpJdkSolrClient}. */
+public class ConcurrentUpdateJdkSolrClient extends ConcurrentUpdateBaseSolrClient {
+
+ private final HttpJdkSolrClient client;
+
+ protected ConcurrentUpdateJdkSolrClient(ConcurrentUpdateJdkSolrClient.Builder builder) {
+ super(builder);
+ this.client = (HttpJdkSolrClient) builder.getClient();
+ }
+
+ @Override
+ protected StreamingResponse doSendUpdateStream(Update update) {
+ UpdateRequest req = update.request();
+ String collection = update.collection();
+ CompletableFuture<HttpResponse<InputStream>> resp =
+ client.requestInputStreamAsync(basePath, req, collection);
+
+ return new StreamingResponse() {
+
+ @Override
+ public int awaitResponse(long timeoutMillis) throws Exception {
+ return resp.get(timeoutMillis, TimeUnit.MILLISECONDS).statusCode();
+ }
+
+ @Override
+ public InputStream getInputStream() {
+ try {
+ return resp.get().body();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return InputStream.nullInputStream();
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Object getUnderlyingResponse() {
+ return resp;
+ }
+
+ @Override
+ public void close() throws IOException {
+ getInputStream().close();
Review Comment:
I changed this to a no-op (as with the existing Jetty Cient). The shared code is careful to completely read the Input Stream on exception, and it is ultimately managed by `java.net.httpclient`. Also, the shared code never actually calls `close`.
--
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: issues-unsubscribe@solr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For additional commands, e-mail: issues-help@solr.apache.org
Re: [PR] SOLR-18065: ConcurrentUpdateBaseSolrClient: support for HttpJdkSolrClient [solr]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4050:
URL: https://github.com/apache/solr/pull/4050#discussion_r2692307506
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java:
##########
@@ -0,0 +1,102 @@
+/*
Review Comment:
Missing Apache License header. All Java source files in the Apache Solr project should include the standard Apache License 2.0 header at the top of the file.
##########
solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java:
##########
@@ -0,0 +1,414 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import jakarta.servlet.ServletException;
+import jakarta.servlet.http.HttpServlet;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.invoke.MethodHandles;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.SocketTimeoutException;
+import java.net.http.HttpConnectTimeoutException;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.request.JavaBinUpdateRequestCodec;
+import org.apache.solr.client.solrj.request.SolrQuery;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.SolrNamedThreadFactory;
+import org.apache.solr.embedded.JettyConfig;
+import org.apache.solr.util.SolrJettyTestRule;
+import org.eclipse.jetty.ee10.servlet.ServletHolder;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class ConcurrentUpdateSolrClientTestBase extends SolrTestCaseJ4 {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public abstract HttpSolrClientBase solrClient(Integer overrideIdleTimeoutMs);
+
+ public abstract ConcurrentUpdateBaseSolrClient concurrentClient(
+ HttpSolrClientBase solrClient,
+ String baseUrl,
+ String defaultCollection,
+ int queueSize,
+ int threadCount,
+ boolean disablePollQueue);
+
+ public abstract ConcurrentUpdateBaseSolrClient outcomeCountingConcurrentClient(
+ String serverUrl,
+ int queueSize,
+ int threadCount,
+ HttpSolrClientBase solrClient,
+ AtomicInteger successCoounter,
+ AtomicInteger failureCoounter,
Review Comment:
Corrected spelling of 'Coounter' to 'Counter'.
```suggestion
AtomicInteger successCounter,
AtomicInteger failureCounter,
```
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpJdkSolrClient.java:
##########
@@ -143,29 +143,36 @@ protected HttpJdkSolrClient(String serverBaseUrl, HttpJdkSolrClient.Builder buil
assert ObjectReleaseTracker.track(this);
}
- @Override
- public CompletableFuture<NamedList<Object>> requestAsync(
- final SolrRequest<?> solrRequest, String collection) {
+ protected CompletableFuture<HttpResponse<InputStream>> requestInputStreamAsync(
+ String overrideBaseUrl, final SolrRequest<?> solrRequest, String collection) {
try {
- PreparedRequest pReq = prepareRequest(solrRequest, collection, null);
- return httpClient
- .sendAsync(pReq.reqb.build(), HttpResponse.BodyHandlers.ofInputStream())
- .thenApply(
- httpResponse -> {
- try {
- return processErrorsAndResponse(
- solrRequest, pReq.parserToUse, httpResponse, pReq.url);
- } catch (SolrServerException e) {
- throw new RuntimeException(e);
- }
- });
+ PreparedRequest pReq = prepareRequest(solrRequest, collection, overrideBaseUrl);
+ return httpClient.sendAsync(pReq.reqb.build(), HttpResponse.BodyHandlers.ofInputStream());
} catch (Exception e) {
- CompletableFuture<NamedList<Object>> cf = new CompletableFuture<>();
+ CompletableFuture<HttpResponse<InputStream>> cf = new CompletableFuture<>();
cf.completeExceptionally(e);
return cf;
}
}
+ @Override
+ public CompletableFuture<NamedList<Object>> requestAsync(
+ final SolrRequest<?> solrRequest, String collection) {
+ return requestInputStreamAsync(null, solrRequest, collection)
+ .thenApply(
+ httpResponse -> {
+ try {
+ PreparedRequest pReq = prepareRequest(solrRequest, collection, null);
Review Comment:
The prepareRequest method is called twice for the same request - once in requestInputStreamAsync (line 149) and again here in the thenApply callback. This duplicate work should be avoided by storing and reusing the PreparedRequest from the first call.
```suggestion
final PreparedRequest pReq;
try {
pReq = prepareRequest(solrRequest, collection, null);
} catch (Exception e) {
CompletableFuture<NamedList<Object>> cf = new CompletableFuture<>();
cf.completeExceptionally(e);
return cf;
}
return httpClient
.sendAsync(pReq.reqb.build(), HttpResponse.BodyHandlers.ofInputStream())
.thenApply(
httpResponse -> {
try {
```
##########
solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClientTest.java:
##########
@@ -0,0 +1,122 @@
+/*
Review Comment:
Missing Apache License header. All Java source files in the Apache Solr project should include the standard Apache License 2.0 header at the top of the file.
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.http.HttpResponse;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+
+/** A ConcurrentUpdate SolrClient using {@link HttpJdkSolrClient}. */
+public class ConcurrentUpdateJdkSolrClient extends ConcurrentUpdateBaseSolrClient {
+
+ private final HttpJdkSolrClient client;
+
+ protected ConcurrentUpdateJdkSolrClient(ConcurrentUpdateJdkSolrClient.Builder builder) {
+ super(builder);
+ this.client = (HttpJdkSolrClient) builder.getClient();
+ }
+
+ @Override
+ protected StreamingResponse doSendUpdateStream(Update update) {
+ UpdateRequest req = update.request();
+ String collection = update.collection();
+ CompletableFuture<HttpResponse<InputStream>> resp =
+ client.requestInputStreamAsync(basePath, req, collection);
+
+ return new StreamingResponse() {
+
+ @Override
+ public int awaitResponse(long timeoutMillis) throws Exception {
+ return resp.get(timeoutMillis, TimeUnit.MILLISECONDS).statusCode();
+ }
+
+ @Override
+ public InputStream getInputStream() {
+ try {
+ return resp.get().body();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return InputStream.nullInputStream();
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Object getUnderlyingResponse() {
+ return resp;
+ }
+
+ @Override
Review Comment:
The hardcoded idle timeout value of 1000ms should be documented, explaining why this specific value was chosen. Additionally, consider adding a comment explaining that the JDK HTTP client doesn't use idle timeout in the same way as the Jetty client (as noted in HttpJdkSolrClient line 99), but this value is still required for the base class's StreamingResponse.awaitResponse method.
--
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: issues-unsubscribe@solr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For additional commands, e-mail: issues-help@solr.apache.org
Re: [PR] SOLR-18065: ConcurrentUpdateBaseSolrClient: support for HttpJdkSolrClient [solr]
Posted by "dsmiley (via GitHub)" <gi...@apache.org>.
dsmiley commented on code in PR #4050:
URL: https://github.com/apache/solr/pull/4050#discussion_r2694752738
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.http.HttpResponse;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+
+/** A ConcurrentUpdate SolrClient using {@link HttpJdkSolrClient}. */
+public class ConcurrentUpdateJdkSolrClient extends ConcurrentUpdateBaseSolrClient {
+
+ private final HttpJdkSolrClient client;
+
+ protected ConcurrentUpdateJdkSolrClient(ConcurrentUpdateJdkSolrClient.Builder builder) {
+ super(builder);
+ this.client = (HttpJdkSolrClient) builder.getClient();
+ }
+
+ @Override
+ protected StreamingResponse doSendUpdateStream(Update update) {
+ UpdateRequest req = update.request();
+ String collection = update.collection();
+ CompletableFuture<HttpResponse<InputStream>> resp =
+ client.requestInputStreamAsync(basePath, req, collection);
+
+ return new StreamingResponse() {
+
+ @Override
+ public int awaitResponse(long timeoutMillis) throws Exception {
+ return resp.get(timeoutMillis, TimeUnit.MILLISECONDS).statusCode();
+ }
+
+ @Override
+ public InputStream getInputStream() {
+ try {
+ return resp.get().body();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return InputStream.nullInputStream();
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Object getUnderlyingResponse() {
+ return resp;
+ }
+
+ @Override
+ public void close() throws IOException {
+ getInputStream().close();
+ }
+ };
+ }
+
+ public static class Builder extends ConcurrentUpdateBaseSolrClient.Builder {
+ /**
+ * @see org.apache.solr.client.solrj.impl.ConcurrentUpdateBaseSolrClient.Builder#Builder(String,
+ * HttpSolrClientBase)
+ */
+ public Builder(String baseUrl, HttpJdkSolrClient client) {
+
+ this(baseUrl, client, false);
+ // The base class uses idle timeout with StreamingResponse#awaitResponse so it needs to be
+ // set!
+ this.idleTimeoutMillis = 1000;
Review Comment:
Good catch! The base builder ought to have a toggle for this. And clarified semantics -- like is -1 supported as infinite? I propose that infinite be the default for this new client. HttpJdkSolrClient internal comments document "note: idle timeout isn't used for the JDK client". The Jetty one I see, we're able to ask the client. 1 second is an extremely low timeout for idle/socket timeout.
--
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: issues-unsubscribe@solr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For additional commands, e-mail: issues-help@solr.apache.org
Re: [PR] SOLR-18065: ConcurrentUpdateBaseSolrClient: support for HttpJdkSolrClient [solr]
Posted by "jdyer1 (via GitHub)" <gi...@apache.org>.
jdyer1 commented on code in PR #4050:
URL: https://github.com/apache/solr/pull/4050#discussion_r2694943681
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.http.HttpResponse;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+
+/** A ConcurrentUpdate SolrClient using {@link HttpJdkSolrClient}. */
+public class ConcurrentUpdateJdkSolrClient extends ConcurrentUpdateBaseSolrClient {
+
+ private final HttpJdkSolrClient client;
+
+ protected ConcurrentUpdateJdkSolrClient(ConcurrentUpdateJdkSolrClient.Builder builder) {
+ super(builder);
+ this.client = (HttpJdkSolrClient) builder.getClient();
+ }
+
+ @Override
+ protected StreamingResponse doSendUpdateStream(Update update) {
+ UpdateRequest req = update.request();
+ String collection = update.collection();
+ CompletableFuture<HttpResponse<InputStream>> resp =
+ client.requestInputStreamAsync(basePath, req, collection);
+
+ return new StreamingResponse() {
+
+ @Override
+ public int awaitResponse(long timeoutMillis) throws Exception {
+ return resp.get(timeoutMillis, TimeUnit.MILLISECONDS).statusCode();
+ }
+
+ @Override
+ public InputStream getInputStream() {
+ try {
+ return resp.get().body();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return InputStream.nullInputStream();
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Object getUnderlyingResponse() {
+ return resp;
+ }
+
+ @Override
+ public void close() throws IOException {
+ getInputStream().close();
+ }
+ };
+ }
+
+ public static class Builder extends ConcurrentUpdateBaseSolrClient.Builder {
+ /**
+ * @see org.apache.solr.client.solrj.impl.ConcurrentUpdateBaseSolrClient.Builder#Builder(String,
+ * HttpSolrClientBase)
+ */
+ public Builder(String baseUrl, HttpJdkSolrClient client) {
+
+ this(baseUrl, client, false);
+ // The base class uses idle timeout with StreamingResponse#awaitResponse so it needs to be
+ // set!
+ this.idleTimeoutMillis = 1000;
Review Comment:
it actually is configurable, @janhoy. This is the constructor for the builder and users can set it to anything they want. In actuality the Jdk Client completely ignores idle timeout, and users might be less likely to set it.
This 1000ms default value is a workaround for the fact that the shared code misuses this timeout value. (`ConcurrentUpdateBaseSolrClient:251`) . The original author probably didn't want to introduce yet one more timeout to configure. Before this change, it defaulted to 0, which would fail unit tests because 0ms is too short to do much of anything.
--
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: issues-unsubscribe@solr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For additional commands, e-mail: issues-help@solr.apache.org
Re: [PR] SOLR-18065: ConcurrentUpdateBaseSolrClient: support for HttpJdkSolrClient [solr]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4050:
URL: https://github.com/apache/solr/pull/4050#discussion_r2692368518
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.http.HttpResponse;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+
+/** A ConcurrentUpdate SolrClient using {@link HttpJdkSolrClient}. */
+public class ConcurrentUpdateJdkSolrClient extends ConcurrentUpdateBaseSolrClient {
+
+ private final HttpJdkSolrClient client;
+
+ protected ConcurrentUpdateJdkSolrClient(ConcurrentUpdateJdkSolrClient.Builder builder) {
+ super(builder);
+ this.client = (HttpJdkSolrClient) builder.getClient();
+ }
+
+ @Override
+ protected StreamingResponse doSendUpdateStream(Update update) {
+ UpdateRequest req = update.request();
+ String collection = update.collection();
+ CompletableFuture<HttpResponse<InputStream>> resp =
+ client.requestInputStreamAsync(basePath, req, collection);
+
+ return new StreamingResponse() {
+
+ @Override
+ public int awaitResponse(long timeoutMillis) throws Exception {
+ return resp.get(timeoutMillis, TimeUnit.MILLISECONDS).statusCode();
+ }
+
+ @Override
+ public InputStream getInputStream() {
+ try {
+ return resp.get().body();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return InputStream.nullInputStream();
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Object getUnderlyingResponse() {
+ return resp;
+ }
+
+ @Override
+ public void close() throws IOException {
+ getInputStream().close();
Review Comment:
Potential resource leak: The close method calls getInputStream which may block or throw exceptions. If resp.get() is not yet complete when close is called, this could block indefinitely or throw an exception before closing the stream. Additionally, if getInputStream returns a null stream due to InterruptedException (as per the implementation), calling close on it may not clean up the underlying HTTP response resources. Consider checking the response completion state before attempting to get and close the input stream, or handle the case where the response may not be available yet.
```suggestion
if (!resp.isDone()) {
// Cancel the request if it has not completed yet to avoid blocking and free resources.
resp.cancel(true);
return;
}
HttpResponse<InputStream> httpResponse;
try {
httpResponse = resp.getNow(null);
} catch (RuntimeException e) {
// Response completed exceptionally; nothing to close here.
return;
}
if (httpResponse == null) {
return;
}
InputStream body = httpResponse.body();
if (body != null) {
body.close();
}
```
##########
solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java:
##########
@@ -0,0 +1,414 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import jakarta.servlet.ServletException;
+import jakarta.servlet.http.HttpServlet;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.invoke.MethodHandles;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.SocketTimeoutException;
+import java.net.http.HttpConnectTimeoutException;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.request.JavaBinUpdateRequestCodec;
+import org.apache.solr.client.solrj.request.SolrQuery;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.SolrNamedThreadFactory;
+import org.apache.solr.embedded.JettyConfig;
+import org.apache.solr.util.SolrJettyTestRule;
+import org.eclipse.jetty.ee10.servlet.ServletHolder;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class ConcurrentUpdateSolrClientTestBase extends SolrTestCaseJ4 {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public abstract HttpSolrClientBase solrClient(Integer overrideIdleTimeoutMs);
+
+ public abstract ConcurrentUpdateBaseSolrClient concurrentClient(
+ HttpSolrClientBase solrClient,
+ String baseUrl,
+ String defaultCollection,
+ int queueSize,
+ int threadCount,
+ boolean disablePollQueue);
+
+ public abstract ConcurrentUpdateBaseSolrClient outcomeCountingConcurrentClient(
+ String serverUrl,
+ int queueSize,
+ int threadCount,
+ HttpSolrClientBase solrClient,
+ AtomicInteger successCounter,
+ AtomicInteger failureCounter,
+ StringBuilder errors);
+
+ /** Mock endpoint where the CUSS being tested in this class sends requests. */
+ public static class TestServlet extends HttpServlet
+ implements JavaBinUpdateRequestCodec.StreamingUpdateHandler {
+ private static final long serialVersionUID = 1L;
+
+ public static void clear() {
+ lastMethod = null;
+ headers = null;
+ parameters = null;
+ errorCode = null;
+ numReqsRcvd.set(0);
+ numDocsRcvd.set(0);
+ }
+
+ public static Integer errorCode = null;
+ public static String lastMethod = null;
+ public static HashMap<String, String> headers = null;
+ public static Map<String, String[]> parameters = null;
+ public static AtomicInteger numReqsRcvd = new AtomicInteger(0);
+ public static AtomicInteger numDocsRcvd = new AtomicInteger(0);
+
+ public static void setErrorCode(Integer code) {
+ errorCode = code;
+ }
+
+ private void setHeaders(HttpServletRequest req) {
+ Enumeration<String> headerNames = req.getHeaderNames();
+ headers = new HashMap<>();
+ while (headerNames.hasMoreElements()) {
+ final String name = headerNames.nextElement();
+ headers.put(name, req.getHeader(name));
+ }
+ }
+
+ private void setParameters(HttpServletRequest req) {
+ // parameters = req.getParameterMap();
Review Comment:
The setParameters method body is commented out, which makes the method ineffective. If parameter tracking is not needed for these tests, the method should either be removed entirely or have a comment explaining why it's intentionally empty. If it was meant to be implemented but was left unfinished, this should be completed.
```suggestion
parameters = req.getParameterMap();
```
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.http.HttpResponse;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+
+/** A ConcurrentUpdate SolrClient using {@link HttpJdkSolrClient}. */
+public class ConcurrentUpdateJdkSolrClient extends ConcurrentUpdateBaseSolrClient {
+
+ private final HttpJdkSolrClient client;
+
+ protected ConcurrentUpdateJdkSolrClient(ConcurrentUpdateJdkSolrClient.Builder builder) {
+ super(builder);
+ this.client = (HttpJdkSolrClient) builder.getClient();
+ }
+
+ @Override
+ protected StreamingResponse doSendUpdateStream(Update update) {
+ UpdateRequest req = update.request();
+ String collection = update.collection();
+ CompletableFuture<HttpResponse<InputStream>> resp =
+ client.requestInputStreamAsync(basePath, req, collection);
+
+ return new StreamingResponse() {
+
+ @Override
+ public int awaitResponse(long timeoutMillis) throws Exception {
+ return resp.get(timeoutMillis, TimeUnit.MILLISECONDS).statusCode();
+ }
+
+ @Override
+ public InputStream getInputStream() {
+ try {
+ return resp.get().body();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return InputStream.nullInputStream();
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Object getUnderlyingResponse() {
+ return resp;
+ }
+
+ @Override
+ public void close() throws IOException {
+ getInputStream().close();
+ }
+ };
+ }
Review Comment:
Missing javadoc for the doSendUpdateStream method. This method implements a critical abstract method from the base class and should document its behavior, parameters, return value, and any exceptions it may throw. The StreamingResponse contract and how this implementation fulfills it should be clearly documented.
##########
changelog/unreleased/SOLR-18065.yml:
##########
@@ -0,0 +1,7 @@
+title: Added new ConcurrentUpdateJdkSolrClient that works with HttpJdkSolrClient
Review Comment:
The changelog entry uses "Added new" which is redundant. The "type: added" field already indicates this is a new feature, so the title should simply say "ConcurrentUpdateJdkSolrClient that works with HttpJdkSolrClient" instead of "Added new ConcurrentUpdateJdkSolrClient that works with HttpJdkSolrClient".
```suggestion
title: ConcurrentUpdateJdkSolrClient that works with HttpJdkSolrClient
```
##########
solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java:
##########
@@ -0,0 +1,414 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import jakarta.servlet.ServletException;
+import jakarta.servlet.http.HttpServlet;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.invoke.MethodHandles;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.SocketTimeoutException;
+import java.net.http.HttpConnectTimeoutException;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.request.JavaBinUpdateRequestCodec;
+import org.apache.solr.client.solrj.request.SolrQuery;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.SolrNamedThreadFactory;
+import org.apache.solr.embedded.JettyConfig;
+import org.apache.solr.util.SolrJettyTestRule;
+import org.eclipse.jetty.ee10.servlet.ServletHolder;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class ConcurrentUpdateSolrClientTestBase extends SolrTestCaseJ4 {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public abstract HttpSolrClientBase solrClient(Integer overrideIdleTimeoutMs);
+
+ public abstract ConcurrentUpdateBaseSolrClient concurrentClient(
+ HttpSolrClientBase solrClient,
+ String baseUrl,
+ String defaultCollection,
+ int queueSize,
+ int threadCount,
+ boolean disablePollQueue);
+
+ public abstract ConcurrentUpdateBaseSolrClient outcomeCountingConcurrentClient(
+ String serverUrl,
+ int queueSize,
+ int threadCount,
+ HttpSolrClientBase solrClient,
+ AtomicInteger successCounter,
+ AtomicInteger failureCounter,
+ StringBuilder errors);
Review Comment:
Missing javadoc for abstract methods in the test base class. The abstract methods solrClient, concurrentClient, and outcomeCountingConcurrentClient should have javadoc comments explaining their purpose, parameters, and return values. This would help developers implementing subclasses understand what these factory methods should do and what constraints they must satisfy.
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpJdkSolrClient.java:
##########
@@ -143,6 +143,18 @@ protected HttpJdkSolrClient(String serverBaseUrl, HttpJdkSolrClient.Builder buil
assert ObjectReleaseTracker.track(this);
}
+ protected CompletableFuture<HttpResponse<InputStream>> requestInputStreamAsync(
+ String overrideBaseUrl, final SolrRequest<?> solrRequest, String collection) {
+ try {
+ PreparedRequest pReq = prepareRequest(solrRequest, collection, overrideBaseUrl);
+ return httpClient.sendAsync(pReq.reqb.build(), HttpResponse.BodyHandlers.ofInputStream());
+ } catch (Exception e) {
+ CompletableFuture<HttpResponse<InputStream>> cf = new CompletableFuture<>();
+ cf.completeExceptionally(e);
+ return cf;
+ }
+ }
Review Comment:
Missing javadoc for the new protected method requestInputStreamAsync. This method is protected and likely intended for use by subclasses or related classes like ConcurrentUpdateJdkSolrClient. It should document its parameters, return value, and the contract around the returned CompletableFuture, including what exceptions might be wrapped in the future and when it might complete exceptionally.
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.http.HttpResponse;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+
+/** A ConcurrentUpdate SolrClient using {@link HttpJdkSolrClient}. */
+public class ConcurrentUpdateJdkSolrClient extends ConcurrentUpdateBaseSolrClient {
+
+ private final HttpJdkSolrClient client;
+
+ protected ConcurrentUpdateJdkSolrClient(ConcurrentUpdateJdkSolrClient.Builder builder) {
+ super(builder);
+ this.client = (HttpJdkSolrClient) builder.getClient();
+ }
+
+ @Override
+ protected StreamingResponse doSendUpdateStream(Update update) {
+ UpdateRequest req = update.request();
+ String collection = update.collection();
+ CompletableFuture<HttpResponse<InputStream>> resp =
+ client.requestInputStreamAsync(basePath, req, collection);
+
+ return new StreamingResponse() {
+
+ @Override
+ public int awaitResponse(long timeoutMillis) throws Exception {
+ return resp.get(timeoutMillis, TimeUnit.MILLISECONDS).statusCode();
+ }
+
+ @Override
+ public InputStream getInputStream() {
+ try {
+ return resp.get().body();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return InputStream.nullInputStream();
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Object getUnderlyingResponse() {
+ return resp;
+ }
+
+ @Override
+ public void close() throws IOException {
+ getInputStream().close();
+ }
+ };
+ }
+
+ public static class Builder extends ConcurrentUpdateBaseSolrClient.Builder {
+ /**
+ * @see org.apache.solr.client.solrj.impl.ConcurrentUpdateBaseSolrClient.Builder#Builder(String,
+ * HttpSolrClientBase)
+ */
+ public Builder(String baseUrl, HttpJdkSolrClient client) {
+
+ this(baseUrl, client, false);
+ // The base class uses idle timeout with StreamingResponse#awaitResponse so it needs to be
+ // set!
+ this.idleTimeoutMillis = 1000;
Review Comment:
The idleTimeoutMillis is hardcoded to 1000 milliseconds in the primary constructor. This overrides any value that might have been set on the HttpJdkSolrClient. Unlike the Jetty implementation which retrieves the idle timeout from the client (line 63 in ConcurrentUpdateJettySolrClient), this hardcoded value could cause unexpected behavior if users have configured a different timeout on their HttpJdkSolrClient. Consider retrieving the idle timeout from the client parameter instead of hardcoding it, or document why the hardcoded value is necessary for the JDK client.
```suggestion
// set. Prefer the client's configured idle timeout when available, otherwise fall back to
// the previous default of 1000 ms.
long clientIdleTimeout = client.getIdleTimeout();
this.idleTimeoutMillis = clientIdleTimeout > 0 ? clientIdleTimeout : 1000;
```
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.http.HttpResponse;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+
+/** A ConcurrentUpdate SolrClient using {@link HttpJdkSolrClient}. */
+public class ConcurrentUpdateJdkSolrClient extends ConcurrentUpdateBaseSolrClient {
+
+ private final HttpJdkSolrClient client;
+
+ protected ConcurrentUpdateJdkSolrClient(ConcurrentUpdateJdkSolrClient.Builder builder) {
+ super(builder);
+ this.client = (HttpJdkSolrClient) builder.getClient();
+ }
+
+ @Override
+ protected StreamingResponse doSendUpdateStream(Update update) {
+ UpdateRequest req = update.request();
+ String collection = update.collection();
+ CompletableFuture<HttpResponse<InputStream>> resp =
+ client.requestInputStreamAsync(basePath, req, collection);
+
+ return new StreamingResponse() {
+
+ @Override
+ public int awaitResponse(long timeoutMillis) throws Exception {
+ return resp.get(timeoutMillis, TimeUnit.MILLISECONDS).statusCode();
+ }
+
+ @Override
+ public InputStream getInputStream() {
+ try {
+ return resp.get().body();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return InputStream.nullInputStream();
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Object getUnderlyingResponse() {
+ return resp;
+ }
+
+ @Override
+ public void close() throws IOException {
+ getInputStream().close();
+ }
+ };
+ }
+
+ public static class Builder extends ConcurrentUpdateBaseSolrClient.Builder {
+ /**
+ * @see org.apache.solr.client.solrj.impl.ConcurrentUpdateBaseSolrClient.Builder#Builder(String,
+ * HttpSolrClientBase)
+ */
+ public Builder(String baseUrl, HttpJdkSolrClient client) {
+
+ this(baseUrl, client, false);
+ // The base class uses idle timeout with StreamingResponse#awaitResponse so it needs to be
+ // set!
Review Comment:
The comment on lines 84-86 explains why idleTimeoutMillis is set, but it would be more helpful if this explanation appeared as a proper javadoc comment on the constructor itself, or at least as a more detailed inline comment. The current comment is too terse and doesn't explain why the value 1000 was chosen specifically.
```suggestion
* Create a {@link ConcurrentUpdateJdkSolrClient} builder for an existing {@link HttpJdkSolrClient}.
*
* <p>This constructor configures a default {@code idleTimeoutMillis} value used by the
* {@link ConcurrentUpdateBaseSolrClient} base class when waiting in
* {@code StreamingResponse#awaitResponse}. The idle timeout must be set explicitly for
* the JDK-based HTTP client so that consumer threads do not block indefinitely while still
* polling frequently enough to keep update latency low. A value of {@code 1000} milliseconds
* (1 second) is chosen as a conservative default that avoids busy waiting while providing
* reasonably responsive flushing of asynchronous update responses.
*
* @param baseUrl the base Solr URL this client will send requests to
* @param client the {@link HttpJdkSolrClient} used to execute HTTP requests
* @see org.apache.solr.client.solrj.impl.ConcurrentUpdateBaseSolrClient.Builder#Builder(String,
* HttpSolrClientBase)
*/
public Builder(String baseUrl, HttpJdkSolrClient client) {
this(baseUrl, client, false);
// Ensure the base class has a sensible idle timeout for awaitResponse().
```
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.http.HttpResponse;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+
+/** A ConcurrentUpdate SolrClient using {@link HttpJdkSolrClient}. */
+public class ConcurrentUpdateJdkSolrClient extends ConcurrentUpdateBaseSolrClient {
+
+ private final HttpJdkSolrClient client;
+
+ protected ConcurrentUpdateJdkSolrClient(ConcurrentUpdateJdkSolrClient.Builder builder) {
+ super(builder);
+ this.client = (HttpJdkSolrClient) builder.getClient();
+ }
+
+ @Override
+ protected StreamingResponse doSendUpdateStream(Update update) {
+ UpdateRequest req = update.request();
+ String collection = update.collection();
+ CompletableFuture<HttpResponse<InputStream>> resp =
+ client.requestInputStreamAsync(basePath, req, collection);
+
+ return new StreamingResponse() {
+
+ @Override
+ public int awaitResponse(long timeoutMillis) throws Exception {
+ return resp.get(timeoutMillis, TimeUnit.MILLISECONDS).statusCode();
+ }
+
+ @Override
+ public InputStream getInputStream() {
+ try {
+ return resp.get().body();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return InputStream.nullInputStream();
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
Review Comment:
The getInputStream method has inconsistent error handling. When an InterruptedException occurs, it correctly restores the interrupt status and returns a null input stream, but when an ExecutionException occurs, it throws a RuntimeException. This inconsistency could lead to unpredictable behavior. Consider handling both exceptions consistently, either by returning a null input stream for both cases or by documenting why ExecutionException should be fatal while InterruptedException is not.
```suggestion
return InputStream.nullInputStream();
```
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.http.HttpResponse;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+
+/** A ConcurrentUpdate SolrClient using {@link HttpJdkSolrClient}. */
+public class ConcurrentUpdateJdkSolrClient extends ConcurrentUpdateBaseSolrClient {
+
+ private final HttpJdkSolrClient client;
+
+ protected ConcurrentUpdateJdkSolrClient(ConcurrentUpdateJdkSolrClient.Builder builder) {
+ super(builder);
+ this.client = (HttpJdkSolrClient) builder.getClient();
+ }
Review Comment:
Missing javadoc for the protected constructor. While the class itself has minimal documentation, the constructor should document its parameter and explain when this constructor would be used versus the public Builder pattern, especially since this is part of a public API that extends a base class.
##########
solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java:
##########
@@ -0,0 +1,414 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import jakarta.servlet.ServletException;
+import jakarta.servlet.http.HttpServlet;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.invoke.MethodHandles;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.SocketTimeoutException;
+import java.net.http.HttpConnectTimeoutException;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.request.JavaBinUpdateRequestCodec;
+import org.apache.solr.client.solrj.request.SolrQuery;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.SolrNamedThreadFactory;
+import org.apache.solr.embedded.JettyConfig;
+import org.apache.solr.util.SolrJettyTestRule;
+import org.eclipse.jetty.ee10.servlet.ServletHolder;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class ConcurrentUpdateSolrClientTestBase extends SolrTestCaseJ4 {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public abstract HttpSolrClientBase solrClient(Integer overrideIdleTimeoutMs);
+
+ public abstract ConcurrentUpdateBaseSolrClient concurrentClient(
+ HttpSolrClientBase solrClient,
+ String baseUrl,
+ String defaultCollection,
+ int queueSize,
+ int threadCount,
+ boolean disablePollQueue);
+
+ public abstract ConcurrentUpdateBaseSolrClient outcomeCountingConcurrentClient(
+ String serverUrl,
+ int queueSize,
+ int threadCount,
+ HttpSolrClientBase solrClient,
Review Comment:
Inconsistent parameter ordering in abstract methods. In concurrentClient, the solrClient parameter is first, but in outcomeCountingConcurrentClient, it's fourth (after serverUrl, queueSize, and threadCount). This inconsistency makes the API harder to use and could lead to confusion. Consider reordering the parameters in outcomeCountingConcurrentClient to match the pattern of concurrentClient for better consistency.
```suggestion
HttpSolrClientBase solrClient,
String serverUrl,
int queueSize,
int threadCount,
```
--
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: issues-unsubscribe@solr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For additional commands, e-mail: issues-help@solr.apache.org
Re: [PR] SOLR-18065: ConcurrentUpdateBaseSolrClient: support for HttpJdkSolrClient [solr]
Posted by "jdyer1 (via GitHub)" <gi...@apache.org>.
jdyer1 commented on code in PR #4050:
URL: https://github.com/apache/solr/pull/4050#discussion_r2694950412
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.http.HttpResponse;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+
+/** A ConcurrentUpdate SolrClient using {@link HttpJdkSolrClient}. */
+public class ConcurrentUpdateJdkSolrClient extends ConcurrentUpdateBaseSolrClient {
+
+ private final HttpJdkSolrClient client;
+
+ protected ConcurrentUpdateJdkSolrClient(ConcurrentUpdateJdkSolrClient.Builder builder) {
+ super(builder);
+ this.client = (HttpJdkSolrClient) builder.getClient();
+ }
+
+ @Override
+ protected StreamingResponse doSendUpdateStream(Update update) {
+ UpdateRequest req = update.request();
+ String collection = update.collection();
+ CompletableFuture<HttpResponse<InputStream>> resp =
+ client.requestInputStreamAsync(basePath, req, collection);
+
+ return new StreamingResponse() {
+
+ @Override
+ public int awaitResponse(long timeoutMillis) throws Exception {
+ return resp.get(timeoutMillis, TimeUnit.MILLISECONDS).statusCode();
+ }
+
+ @Override
+ public InputStream getInputStream() {
+ try {
+ return resp.get().body();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return InputStream.nullInputStream();
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Object getUnderlyingResponse() {
+ return resp;
+ }
+
+ @Override
+ public void close() throws IOException {
+ getInputStream().close();
+ }
+ };
+ }
+
+ public static class Builder extends ConcurrentUpdateBaseSolrClient.Builder {
+ /**
+ * @see org.apache.solr.client.solrj.impl.ConcurrentUpdateBaseSolrClient.Builder#Builder(String,
+ * HttpSolrClientBase)
+ */
+ public Builder(String baseUrl, HttpJdkSolrClient client) {
+
+ this(baseUrl, client, false);
+ // The base class uses idle timeout with StreamingResponse#awaitResponse so it needs to be
+ // set!
+ this.idleTimeoutMillis = 1000;
Review Comment:
looking at the jetty impl, it defaults it to 600_000 so I will change this here to use that as well.
--
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: issues-unsubscribe@solr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For additional commands, e-mail: issues-help@solr.apache.org
Re: [PR] SOLR-18065: ConcurrentUpdateBaseSolrClient: support for HttpJdkSolrClient [solr]
Posted by "jdyer1 (via GitHub)" <gi...@apache.org>.
jdyer1 commented on code in PR #4050:
URL: https://github.com/apache/solr/pull/4050#discussion_r2694950412
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.http.HttpResponse;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+
+/** A ConcurrentUpdate SolrClient using {@link HttpJdkSolrClient}. */
+public class ConcurrentUpdateJdkSolrClient extends ConcurrentUpdateBaseSolrClient {
+
+ private final HttpJdkSolrClient client;
+
+ protected ConcurrentUpdateJdkSolrClient(ConcurrentUpdateJdkSolrClient.Builder builder) {
+ super(builder);
+ this.client = (HttpJdkSolrClient) builder.getClient();
+ }
+
+ @Override
+ protected StreamingResponse doSendUpdateStream(Update update) {
+ UpdateRequest req = update.request();
+ String collection = update.collection();
+ CompletableFuture<HttpResponse<InputStream>> resp =
+ client.requestInputStreamAsync(basePath, req, collection);
+
+ return new StreamingResponse() {
+
+ @Override
+ public int awaitResponse(long timeoutMillis) throws Exception {
+ return resp.get(timeoutMillis, TimeUnit.MILLISECONDS).statusCode();
+ }
+
+ @Override
+ public InputStream getInputStream() {
+ try {
+ return resp.get().body();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return InputStream.nullInputStream();
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Object getUnderlyingResponse() {
+ return resp;
+ }
+
+ @Override
+ public void close() throws IOException {
+ getInputStream().close();
+ }
+ };
+ }
+
+ public static class Builder extends ConcurrentUpdateBaseSolrClient.Builder {
+ /**
+ * @see org.apache.solr.client.solrj.impl.ConcurrentUpdateBaseSolrClient.Builder#Builder(String,
+ * HttpSolrClientBase)
+ */
+ public Builder(String baseUrl, HttpJdkSolrClient client) {
+
+ this(baseUrl, client, false);
+ // The base class uses idle timeout with StreamingResponse#awaitResponse so it needs to be
+ // set!
+ this.idleTimeoutMillis = 1000;
Review Comment:
looking at the jetty impl, it defaults it to 60,000 so I will change this here to use that as well.
--
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: issues-unsubscribe@solr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For additional commands, e-mail: issues-help@solr.apache.org
Re: [PR] SOLR-18065: ConcurrentUpdateBaseSolrClient: support for HttpJdkSolrClient [solr]
Posted by "jdyer1 (via GitHub)" <gi...@apache.org>.
jdyer1 commented on code in PR #4050:
URL: https://github.com/apache/solr/pull/4050#discussion_r2694942673
##########
solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java:
##########
@@ -0,0 +1,414 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import jakarta.servlet.ServletException;
+import jakarta.servlet.http.HttpServlet;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.invoke.MethodHandles;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.SocketTimeoutException;
+import java.net.http.HttpConnectTimeoutException;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.request.JavaBinUpdateRequestCodec;
+import org.apache.solr.client.solrj.request.SolrQuery;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.SolrNamedThreadFactory;
+import org.apache.solr.embedded.JettyConfig;
+import org.apache.solr.util.SolrJettyTestRule;
+import org.eclipse.jetty.ee10.servlet.ServletHolder;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class ConcurrentUpdateSolrClientTestBase extends SolrTestCaseJ4 {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public abstract HttpSolrClientBase solrClient(Integer overrideIdleTimeoutMs);
+
+ public abstract ConcurrentUpdateBaseSolrClient concurrentClient(
+ HttpSolrClientBase solrClient,
+ String baseUrl,
+ String defaultCollection,
+ int queueSize,
+ int threadCount,
+ boolean disablePollQueue);
+
+ public abstract ConcurrentUpdateBaseSolrClient outcomeCountingConcurrentClient(
+ String serverUrl,
+ int queueSize,
+ int threadCount,
+ HttpSolrClientBase solrClient,
+ AtomicInteger successCounter,
+ AtomicInteger failureCounter,
+ StringBuilder errors);
+
+ /** Mock endpoint where the CUSS being tested in this class sends requests. */
+ public static class TestServlet extends HttpServlet
+ implements JavaBinUpdateRequestCodec.StreamingUpdateHandler {
+ private static final long serialVersionUID = 1L;
+
+ public static void clear() {
+ lastMethod = null;
+ headers = null;
+ parameters = null;
+ errorCode = null;
+ numReqsRcvd.set(0);
+ numDocsRcvd.set(0);
+ }
+
+ public static Integer errorCode = null;
+ public static String lastMethod = null;
+ public static HashMap<String, String> headers = null;
+ public static Map<String, String[]> parameters = null;
+ public static AtomicInteger numReqsRcvd = new AtomicInteger(0);
+ public static AtomicInteger numDocsRcvd = new AtomicInteger(0);
+
+ public static void setErrorCode(Integer code) {
+ errorCode = code;
+ }
+
+ private void setHeaders(HttpServletRequest req) {
+ Enumeration<String> headerNames = req.getHeaderNames();
+ headers = new HashMap<>();
+ while (headerNames.hasMoreElements()) {
+ final String name = headerNames.nextElement();
+ headers.put(name, req.getHeader(name));
+ }
+ }
+
+ private void setParameters(HttpServletRequest req) {
+ // parameters = req.getParameterMap();
+ }
+
+ @Override
+ protected void doPost(HttpServletRequest req, HttpServletResponse resp)
+ throws ServletException, IOException {
+
+ numReqsRcvd.incrementAndGet();
+ lastMethod = "post";
+ recordRequest(req, resp);
+
+ InputStream reqIn = req.getInputStream();
+ JavaBinUpdateRequestCodec javabin = new JavaBinUpdateRequestCodec();
+ for (; ; ) {
+ try {
+ javabin.unmarshal(reqIn, this);
+ } catch (EOFException e) {
+ break; // this is expected
+ }
+ }
+ }
+
+ private void recordRequest(HttpServletRequest req, HttpServletResponse resp) {
+ setHeaders(req);
+ setParameters(req);
+ if (null != errorCode) {
+ try {
+ resp.sendError(errorCode);
+ } catch (IOException e) {
+ throw new RuntimeException("sendError IO fail in TestServlet", e);
+ }
+ }
+ }
+
+ @Override
+ public void update(
+ SolrInputDocument document, UpdateRequest req, Integer commitWithin, Boolean override) {
+ numDocsRcvd.incrementAndGet();
+ }
+ } // end TestServlet
+
+ static class SendDocsRunnable implements Runnable {
+
+ private String id;
+ private int numDocs;
+ private SolrClient cuss;
+ private String collection;
+
+ SendDocsRunnable(String id, int numDocs, SolrClient cuss) {
+ this(id, numDocs, cuss, null);
+ }
+
+ SendDocsRunnable(String id, int numDocs, SolrClient cuss, String collection) {
+ this.id = id;
+ this.numDocs = numDocs;
+ this.cuss = cuss;
+ this.collection = collection;
+ }
+
+ @Override
+ public void run() {
+ for (int d = 0; d < numDocs; d++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ String docId = id + "_" + d;
+ doc.setField("id", docId);
+ UpdateRequest req = new UpdateRequest();
+ req.add(doc);
+ try {
+ if (this.collection == null) cuss.request(req);
+ else cuss.request(req, this.collection);
+ } catch (Throwable t) {
+ log.error("error making request", t);
+ }
+ }
+ }
+ }
+
+ @ClassRule public static SolrJettyTestRule solrTestRule = new SolrJettyTestRule();
+
+ @BeforeClass
+ public static void beforeTest() throws Exception {
+ JettyConfig jettyConfig =
+ JettyConfig.builder().withServlet(new ServletHolder(TestServlet.class), "/cuss/*").build();
+ solrTestRule.startSolr(legacyExampleCollection1SolrHome(), new Properties(), jettyConfig);
+ }
+
+ @AfterClass
+ public static void afterTest() throws Exception {}
+
+ @Test
+ public void testConcurrentUpdate() throws Exception {
+ TestServlet.clear();
+
+ String serverUrl = solrTestRule.getBaseUrl() + "/cuss/foo";
+
+ int cussThreadCount = 2;
+ int cussQueueSize = 100;
+
+ // for tracking callbacks from CUSS
+ final AtomicInteger successCounter = new AtomicInteger(0);
+ final AtomicInteger errorCounter = new AtomicInteger(0);
+ final StringBuilder errors = new StringBuilder();
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ outcomeCountingConcurrentClient(
+ serverUrl,
+ cussQueueSize,
+ cussThreadCount,
+ http2Client,
+ successCounter,
+ errorCounter,
+ errors)) {
+
+ // ensure it doesn't block where there's nothing to do yet
+ concurrentClient.blockUntilFinished();
+
+ int poolSize = 5;
+ ExecutorService threadPool =
+ ExecutorUtil.newMDCAwareFixedThreadPool(poolSize, new SolrNamedThreadFactory("testCUSS"));
+
+ int numDocs = 100;
+ int numRunnables = 5;
+ for (int r = 0; r < numRunnables; r++)
+ threadPool.execute(new SendDocsRunnable(String.valueOf(r), numDocs, concurrentClient));
+
+ // ensure all docs are sent
+ threadPool.awaitTermination(5, TimeUnit.SECONDS);
+ threadPool.shutdown();
+
+ // wait until all requests are processed by CUSS
+ concurrentClient.blockUntilFinished();
+ concurrentClient.shutdownNow();
+
+ assertEquals("post", TestServlet.lastMethod);
+
+ // expect all requests to be successful
+ int expectedSuccesses = TestServlet.numReqsRcvd.get();
+ assertTrue(expectedSuccesses > 0); // at least one request must have been sent
+
+ assertEquals(
+ "Expected no errors but got " + errorCounter.get() + ", due to: " + errors.toString(),
+ 0,
+ errorCounter.get());
+ assertEquals(
+ "Expected " + expectedSuccesses + " successes, but got " + successCounter.get(),
+ successCounter.get(),
+ expectedSuccesses);
+
+ int expectedDocs = numDocs * numRunnables;
+ assertEquals(
+ "Expected CUSS to send " + expectedDocs + " but got " + TestServlet.numDocsRcvd.get(),
+ TestServlet.numDocsRcvd.get(),
+ expectedDocs);
+ }
+ }
+
+ @Test
+ public void testCollectionParameters() throws IOException, SolrServerException {
+
+ int cussThreadCount = 2;
+ int cussQueueSize = 10;
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ null,
+ cussThreadCount,
+ cussQueueSize,
+ false)) {
+
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", "collection");
+ concurrentClient.add("collection1", doc);
+ concurrentClient.commit("collection1");
+
+ assertEquals(
+ 1,
+ concurrentClient
+ .query("collection1", new SolrQuery("id:collection"))
+ .getResults()
+ .getNumFound());
+ }
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ DEFAULT_TEST_CORENAME,
+ cussThreadCount,
+ cussQueueSize,
+ false)) {
+
+ assertEquals(
+ 1, concurrentClient.query(new SolrQuery("id:collection")).getResults().getNumFound());
+ }
+ }
+
+ @Test
+ public void testConcurrentCollectionUpdate() throws Exception {
+
+ int cussThreadCount = 2;
+ int cussQueueSize = 100;
+ int numDocs = 100;
+ int numRunnables = 5;
+ int expected = numDocs * numRunnables;
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ null,
+ cussThreadCount,
+ cussQueueSize,
+ true)) {
+
+ // ensure it doesn't block where there's nothing to do yet
+ concurrentClient.blockUntilFinished();
+
+ // Delete all existing documents.
+ concurrentClient.deleteByQuery("collection1", "*:*");
+
+ int poolSize = 5;
+ ExecutorService threadPool =
+ ExecutorUtil.newMDCAwareFixedThreadPool(poolSize, new SolrNamedThreadFactory("testCUSS"));
+
+ for (int r = 0; r < numRunnables; r++)
+ threadPool.execute(
+ new SendDocsRunnable(String.valueOf(r), numDocs, concurrentClient, "collection1"));
+
+ // ensure all docs are sent
+ threadPool.awaitTermination(5, TimeUnit.SECONDS);
+ threadPool.shutdown();
+
+ concurrentClient.commit("collection1");
+
+ assertEquals(
+ expected,
+ concurrentClient.query("collection1", new SolrQuery("*:*")).getResults().getNumFound());
+
+ // wait until all requests are processed by CUSS
+ concurrentClient.blockUntilFinished();
+ concurrentClient.shutdownNow();
+ }
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ DEFAULT_TEST_CORENAME,
+ cussThreadCount,
+ cussQueueSize,
+ false)) {
+
+ assertEquals(
+ expected, concurrentClient.query(new SolrQuery("*:*")).getResults().getNumFound());
+ }
+ }
+
+ /**
+ * Test that connection timeout information is passed to the HttpSolrClient that handles non add
+ * operations.
+ */
+ @Test(timeout = 10000)
+ public void testSocketTimeoutOnCommit() throws IOException, SolrServerException {
+ InetAddress localHost = InetAddress.getLocalHost();
+ try (ServerSocket server = new ServerSocket(0, 1, localHost);
+ var http2Client = solrClient(1);
+ var client =
+ concurrentClient(
+ http2Client,
+ "http://"
+ + localHost.getHostAddress()
+ + ":"
+ + server.getLocalPort()
+ + "/noOneThere",
+ null,
+ 10,
+ 1,
Review Comment:
yes, those are the defaults and we are preserving them here!
##########
solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java:
##########
@@ -0,0 +1,414 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import jakarta.servlet.ServletException;
+import jakarta.servlet.http.HttpServlet;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.invoke.MethodHandles;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.SocketTimeoutException;
+import java.net.http.HttpConnectTimeoutException;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.request.JavaBinUpdateRequestCodec;
+import org.apache.solr.client.solrj.request.SolrQuery;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.SolrNamedThreadFactory;
+import org.apache.solr.embedded.JettyConfig;
+import org.apache.solr.util.SolrJettyTestRule;
+import org.eclipse.jetty.ee10.servlet.ServletHolder;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class ConcurrentUpdateSolrClientTestBase extends SolrTestCaseJ4 {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public abstract HttpSolrClientBase solrClient(Integer overrideIdleTimeoutMs);
+
+ public abstract ConcurrentUpdateBaseSolrClient concurrentClient(
+ HttpSolrClientBase solrClient,
+ String baseUrl,
+ String defaultCollection,
+ int queueSize,
+ int threadCount,
+ boolean disablePollQueue);
+
+ public abstract ConcurrentUpdateBaseSolrClient outcomeCountingConcurrentClient(
+ String serverUrl,
+ int queueSize,
+ int threadCount,
+ HttpSolrClientBase solrClient,
+ AtomicInteger successCounter,
+ AtomicInteger failureCounter,
+ StringBuilder errors);
+
+ /** Mock endpoint where the CUSS being tested in this class sends requests. */
+ public static class TestServlet extends HttpServlet
+ implements JavaBinUpdateRequestCodec.StreamingUpdateHandler {
+ private static final long serialVersionUID = 1L;
+
+ public static void clear() {
+ lastMethod = null;
+ headers = null;
+ parameters = null;
+ errorCode = null;
+ numReqsRcvd.set(0);
+ numDocsRcvd.set(0);
+ }
+
+ public static Integer errorCode = null;
+ public static String lastMethod = null;
+ public static HashMap<String, String> headers = null;
+ public static Map<String, String[]> parameters = null;
+ public static AtomicInteger numReqsRcvd = new AtomicInteger(0);
+ public static AtomicInteger numDocsRcvd = new AtomicInteger(0);
+
+ public static void setErrorCode(Integer code) {
+ errorCode = code;
+ }
+
+ private void setHeaders(HttpServletRequest req) {
+ Enumeration<String> headerNames = req.getHeaderNames();
+ headers = new HashMap<>();
+ while (headerNames.hasMoreElements()) {
+ final String name = headerNames.nextElement();
+ headers.put(name, req.getHeader(name));
+ }
+ }
+
+ private void setParameters(HttpServletRequest req) {
+ // parameters = req.getParameterMap();
+ }
+
+ @Override
+ protected void doPost(HttpServletRequest req, HttpServletResponse resp)
+ throws ServletException, IOException {
+
+ numReqsRcvd.incrementAndGet();
+ lastMethod = "post";
+ recordRequest(req, resp);
+
+ InputStream reqIn = req.getInputStream();
+ JavaBinUpdateRequestCodec javabin = new JavaBinUpdateRequestCodec();
+ for (; ; ) {
+ try {
+ javabin.unmarshal(reqIn, this);
+ } catch (EOFException e) {
+ break; // this is expected
+ }
+ }
+ }
+
+ private void recordRequest(HttpServletRequest req, HttpServletResponse resp) {
+ setHeaders(req);
+ setParameters(req);
+ if (null != errorCode) {
+ try {
+ resp.sendError(errorCode);
+ } catch (IOException e) {
+ throw new RuntimeException("sendError IO fail in TestServlet", e);
+ }
+ }
+ }
+
+ @Override
+ public void update(
+ SolrInputDocument document, UpdateRequest req, Integer commitWithin, Boolean override) {
+ numDocsRcvd.incrementAndGet();
+ }
+ } // end TestServlet
+
+ static class SendDocsRunnable implements Runnable {
+
+ private String id;
+ private int numDocs;
+ private SolrClient cuss;
+ private String collection;
+
+ SendDocsRunnable(String id, int numDocs, SolrClient cuss) {
+ this(id, numDocs, cuss, null);
+ }
+
+ SendDocsRunnable(String id, int numDocs, SolrClient cuss, String collection) {
+ this.id = id;
+ this.numDocs = numDocs;
+ this.cuss = cuss;
+ this.collection = collection;
+ }
+
+ @Override
+ public void run() {
+ for (int d = 0; d < numDocs; d++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ String docId = id + "_" + d;
+ doc.setField("id", docId);
+ UpdateRequest req = new UpdateRequest();
+ req.add(doc);
+ try {
+ if (this.collection == null) cuss.request(req);
+ else cuss.request(req, this.collection);
+ } catch (Throwable t) {
+ log.error("error making request", t);
+ }
+ }
+ }
+ }
+
+ @ClassRule public static SolrJettyTestRule solrTestRule = new SolrJettyTestRule();
+
+ @BeforeClass
+ public static void beforeTest() throws Exception {
+ JettyConfig jettyConfig =
+ JettyConfig.builder().withServlet(new ServletHolder(TestServlet.class), "/cuss/*").build();
+ solrTestRule.startSolr(legacyExampleCollection1SolrHome(), new Properties(), jettyConfig);
+ }
+
+ @AfterClass
+ public static void afterTest() throws Exception {}
+
+ @Test
+ public void testConcurrentUpdate() throws Exception {
+ TestServlet.clear();
+
+ String serverUrl = solrTestRule.getBaseUrl() + "/cuss/foo";
+
+ int cussThreadCount = 2;
+ int cussQueueSize = 100;
+
+ // for tracking callbacks from CUSS
+ final AtomicInteger successCounter = new AtomicInteger(0);
+ final AtomicInteger errorCounter = new AtomicInteger(0);
+ final StringBuilder errors = new StringBuilder();
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ outcomeCountingConcurrentClient(
+ serverUrl,
+ cussQueueSize,
+ cussThreadCount,
+ http2Client,
+ successCounter,
+ errorCounter,
+ errors)) {
+
+ // ensure it doesn't block where there's nothing to do yet
+ concurrentClient.blockUntilFinished();
+
+ int poolSize = 5;
+ ExecutorService threadPool =
+ ExecutorUtil.newMDCAwareFixedThreadPool(poolSize, new SolrNamedThreadFactory("testCUSS"));
+
+ int numDocs = 100;
+ int numRunnables = 5;
+ for (int r = 0; r < numRunnables; r++)
+ threadPool.execute(new SendDocsRunnable(String.valueOf(r), numDocs, concurrentClient));
+
+ // ensure all docs are sent
+ threadPool.awaitTermination(5, TimeUnit.SECONDS);
+ threadPool.shutdown();
+
+ // wait until all requests are processed by CUSS
+ concurrentClient.blockUntilFinished();
+ concurrentClient.shutdownNow();
+
+ assertEquals("post", TestServlet.lastMethod);
+
+ // expect all requests to be successful
+ int expectedSuccesses = TestServlet.numReqsRcvd.get();
+ assertTrue(expectedSuccesses > 0); // at least one request must have been sent
+
+ assertEquals(
+ "Expected no errors but got " + errorCounter.get() + ", due to: " + errors.toString(),
+ 0,
+ errorCounter.get());
+ assertEquals(
+ "Expected " + expectedSuccesses + " successes, but got " + successCounter.get(),
+ successCounter.get(),
+ expectedSuccesses);
+
+ int expectedDocs = numDocs * numRunnables;
+ assertEquals(
+ "Expected CUSS to send " + expectedDocs + " but got " + TestServlet.numDocsRcvd.get(),
+ TestServlet.numDocsRcvd.get(),
+ expectedDocs);
+ }
+ }
+
+ @Test
+ public void testCollectionParameters() throws IOException, SolrServerException {
+
+ int cussThreadCount = 2;
+ int cussQueueSize = 10;
+
+ try (var http2Client = solrClient(null);
+ var concurrentClient =
+ concurrentClient(
+ http2Client,
+ solrTestRule.getBaseUrl(),
+ null,
+ cussThreadCount,
+ cussQueueSize,
Review Comment:
oh my! Thank you for noticing that!
--
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: issues-unsubscribe@solr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For additional commands, e-mail: issues-help@solr.apache.org
Re: [PR] SOLR-18065: ConcurrentUpdateBaseSolrClient: support for HttpJdkSolrClient [solr]
Posted by "jdyer1 (via GitHub)" <gi...@apache.org>.
jdyer1 merged PR #4050:
URL: https://github.com/apache/solr/pull/4050
--
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: issues-unsubscribe@solr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@solr.apache.org
For additional commands, e-mail: issues-help@solr.apache.org