From bcd15dc002fb378881bb9b6fa4e3c37827ef83fe Mon Sep 17 00:00:00 2001 From: Serhiy Bzhezytskyy Date: Mon, 13 Jul 2026 08:32:00 +0300 Subject: [PATCH 1/2] SOLR-3284: let ConcurrentUpdateSolrClient report which documents failed to send Add a Builder-configurable error handler so callers can recover the documents of an update batch that failed to reach the server. handleError(Throwable) alone only gave the exception, not which documents were lost, so callers could not route the failed docs to a retry queue or dead-letter topic. - New UpdateErrorHandler: onError(Throwable, List failedIds, String collection) reports the ids of the failed documents. Ids are obtained via an overridable extractId(SolrInputDocument) that defaults to the "id" field, so callers whose uniqueKey differs can override it. Registered via Builder.withErrorHandler(...). With no handler the client keeps its existing log-only behavior and never reads the documents, so this is backward compatible and preserves the client's streaming memory efficiency (only ids are retained, not the documents). - doSendUpdateStream returns SentStream(response, ids, collection). The Jetty client coalesces several queued updates into one stream, so it collects the ids of all of them -- the handler receives every affected id on failure, not just the first. The JDK client sends one request per update. - An exception thrown by the handler is logged and does not stop the runner. Tests live in ConcurrentUpdateSolrClientTestBase so they run for both the JDK and Jetty clients. --- .../unreleased/solr-3284-cusc-failed-docs.yml | 12 ++ .../ConcurrentUpdateJettySolrClient.java | 28 ++- .../ConcurrentUpdateJettySolrClientTest.java | 15 ++ .../impl/ConcurrentUpdateBaseSolrClient.java | 119 ++++++++++- .../impl/ConcurrentUpdateJdkSolrClient.java | 54 ++--- .../ConcurrentUpdateJdkSolrClientTest.java | 15 ++ .../ConcurrentUpdateSolrClientTestBase.java | 187 ++++++++++++++++++ 7 files changed, 390 insertions(+), 40 deletions(-) create mode 100644 changelog/unreleased/solr-3284-cusc-failed-docs.yml diff --git a/changelog/unreleased/solr-3284-cusc-failed-docs.yml b/changelog/unreleased/solr-3284-cusc-failed-docs.yml new file mode 100644 index 000000000000..66849e93654b --- /dev/null +++ b/changelog/unreleased/solr-3284-cusc-failed-docs.yml @@ -0,0 +1,12 @@ +# See https://github.com/apache/solr/blob/main/dev-docs/changelog.adoc + +title: > + ConcurrentUpdateSolrClient can now report which documents failed to reach the server. + Register a handler via the Builder's withErrorHandler(...) to recover the documents of a + failed batch, for example to route them to a retry queue or dead-letter topic. +type: added +authors: + - name: Serhiy Bzhezytskyy +links: + - name: SOLR-3284 + url: https://issues.apache.org/jira/browse/SOLR-3284 diff --git a/solr/solrj-jetty/src/java/org/apache/solr/client/solrj/jetty/ConcurrentUpdateJettySolrClient.java b/solr/solrj-jetty/src/java/org/apache/solr/client/solrj/jetty/ConcurrentUpdateJettySolrClient.java index 7e8686f1a39c..458f57004b75 100644 --- a/solr/solrj-jetty/src/java/org/apache/solr/client/solrj/jetty/ConcurrentUpdateJettySolrClient.java +++ b/solr/solrj-jetty/src/java/org/apache/solr/client/solrj/jetty/ConcurrentUpdateJettySolrClient.java @@ -22,6 +22,8 @@ import java.io.InputStream; import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; import java.util.Locale; import java.util.Objects; import java.util.concurrent.TimeUnit; @@ -75,10 +77,10 @@ protected ConcurrentUpdateJettySolrClient(Builder builder) { } @Override - protected StreamingResponse doSendUpdateStream(ConcurrentUpdateBaseSolrClient.Update update) + protected SentStream doSendUpdateStream(ConcurrentUpdateBaseSolrClient.Update update) throws IOException, InterruptedException { - InputStreamResponseListener jettyListener; - try (OutStream out = initOutStream(basePath, update.request(), update.collection())) { + OutStream out = initOutStream(basePath, update.request(), update.collection()); + try (out) { ConcurrentUpdateBaseSolrClient.Update upd = update; while (upd != null) { UpdateRequest req = upd.request(); @@ -87,15 +89,17 @@ protected StreamingResponse doSendUpdateStream(ConcurrentUpdateBaseSolrClient.Up queue.add(upd); break; } - send(out, upd.request(), upd.collection()); + send(out, upd); out.flush(); notifyQueueAndRunnersIfEmptyQueue(); upd = queue.poll(pollQueueTimeMillis, TimeUnit.MILLISECONDS); } - jettyListener = out.getResponseListener(); } - return new JettyStreamingResponse(jettyListener); + return new SentStream( + new JettyStreamingResponse(out.getResponseListener()), + out.getDocIds(), + update.collection()); } private static class OutStream implements Closeable { @@ -104,6 +108,7 @@ private static class OutStream implements Closeable { private final OutputStreamRequestContent content; private final InputStreamResponseListener responseListener; private final boolean isXml; + private final List docIds = new ArrayList<>(); public OutStream( String origCollection, @@ -123,6 +128,10 @@ boolean belongToThisStream(SolrRequest solrRequest, String collection) { && Objects.equals(origCollection, collection); } + List getDocIds() { + return docIds; + } + public void write(byte[] b) throws IOException { this.content.getOutputStream().write(b); } @@ -210,8 +219,11 @@ private OutStream initOutStream(String baseUrl, UpdateRequest updateRequest, Str return outStream; } - private void send(OutStream outStream, SolrRequest req, String collection) throws IOException { - assert outStream.belongToThisStream(req, collection); + private void send(OutStream outStream, ConcurrentUpdateBaseSolrClient.Update update) + throws IOException { + UpdateRequest req = update.request(); + assert outStream.belongToThisStream(req, update.collection()); + outStream.docIds.addAll(idsForErrorReporting(req)); client.getRequestWriter().write(req, outStream.content.getOutputStream()); if (outStream.isXml) { // check for commit or optimize diff --git a/solr/solrj-jetty/src/test/org/apache/solr/client/solrj/jetty/ConcurrentUpdateJettySolrClientTest.java b/solr/solrj-jetty/src/test/org/apache/solr/client/solrj/jetty/ConcurrentUpdateJettySolrClientTest.java index df28d17f3b0d..10808dedba45 100644 --- a/solr/solrj-jetty/src/test/org/apache/solr/client/solrj/jetty/ConcurrentUpdateJettySolrClientTest.java +++ b/solr/solrj-jetty/src/test/org/apache/solr/client/solrj/jetty/ConcurrentUpdateJettySolrClientTest.java @@ -43,6 +43,21 @@ public ConcurrentUpdateBaseSolrClient outcomeCountingConcurrentClient( .build(); } + @Override + public ConcurrentUpdateBaseSolrClient errorHandlerConcurrentClient( + String serverUrl, + int queueSize, + int threadCount, + HttpSolrClient solrClient, + ConcurrentUpdateBaseSolrClient.UpdateErrorHandler errorHandler) { + return new ConcurrentUpdateJettySolrClient.Builder(serverUrl, (HttpJettySolrClient) solrClient) + .withQueueSize(queueSize) + .withThreadCount(threadCount) + .setPollQueueTime(0, TimeUnit.MILLISECONDS) + .withErrorHandler(errorHandler) + .build(); + } + @Override public HttpSolrClient solrClient(Integer overrideIdleTimeoutMs) { var builder = new HttpJettySolrClient.Builder(); diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateBaseSolrClient.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateBaseSolrClient.java index 752e7131509d..087155639a4d 100644 --- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateBaseSolrClient.java +++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateBaseSolrClient.java @@ -22,7 +22,9 @@ import java.lang.invoke.MethodHandles; import java.net.HttpURLConnection; import java.util.ArrayDeque; +import java.util.ArrayList; import java.util.Iterator; +import java.util.List; import java.util.Queue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; @@ -37,6 +39,7 @@ import org.apache.solr.client.solrj.request.UpdateRequest; import org.apache.solr.client.solrj.util.ClientUtils; import org.apache.solr.common.SolrException; +import org.apache.solr.common.SolrInputDocument; import org.apache.solr.common.params.SolrParams; import org.apache.solr.common.params.UpdateParams; import org.apache.solr.common.util.ExecutorUtil; @@ -71,6 +74,40 @@ public abstract class ConcurrentUpdateBaseSolrClient extends SolrClient { protected StallDetection stallDetection; + private final UpdateErrorHandler errorHandler; + + /** + * Callback invoked when a batch fails to reach the server. Register one via {@link + * Builder#withErrorHandler} to handle update errors comprehensively -- e.g. route the failed + * document ids to a retry queue or dead-letter topic. Implementations must be thread-safe; see + * {@link Builder#withErrorHandler} for the full threading contract. + */ + @FunctionalInterface + public interface UpdateErrorHandler { + /** + * Invoked when specific documents failed to reach the server. + * + * @param ex the error that occurred + * @param failedIds the ids of the documents that did not reach the server + * @param collection the collection the batch targeted, or null + */ + void onError(Throwable ex, List failedIds, String collection); + + /** + * Invoked for an error not tied to specific documents (e.g. a failure before any document was + * sent). Defaults to logging, so a registered handler never silently swallows such errors. + */ + default void onError(Throwable ex) { + LoggerFactory.getLogger(UpdateErrorHandler.class).error("update error", ex); + } + + /** The id used to report a failed document; defaults to its {@code id} field. */ + default String idOf(SolrInputDocument doc) { + Object id = doc.getFieldValue("id"); + return id == null ? null : id.toString(); + } + } + protected static class CustomBlockingQueue implements Iterable { private final BlockingQueue queue; private final Semaphore available; @@ -158,6 +195,7 @@ protected ConcurrentUpdateBaseSolrClient(Builder builder) { this.basePath = builder.baseSolrUrl; this.defaultCollection = builder.defaultCollection; this.pollQueueTimeMillis = builder.pollQueueTimeMillis; + this.errorHandler = builder.errorHandler; // Initialize stall detection long stallTimeMillis = Integer.getInteger("solr.cloud.client.stallTime", 15000); @@ -187,6 +225,31 @@ protected ConcurrentUpdateBaseSolrClient(Builder builder) { /** Class representing an UpdateRequest and an optional collection. */ protected record Update(UpdateRequest request, String collection) {} + /** + * The result of sending updates as a single stream: the response to await, plus the ids of the + * documents sent (for error reporting) and their collection. + */ + public record SentStream(StreamingResponse response, List docIds, String collection) {} + + /** + * The ids of a request's documents for error reporting, via {@link UpdateErrorHandler#idOf}. + * Empty when no error handler is registered, so the documents are not read and no ids are + * retained. + */ + protected List idsForErrorReporting(UpdateRequest request) { + if (errorHandler == null || request.getDocuments() == null) { + return List.of(); + } + List ids = new ArrayList<>(request.getDocuments().size()); + for (SolrInputDocument doc : request.getDocuments()) { + String id = errorHandler.idOf(doc); + if (id != null) { + ids.add(id); + } + } + return ids; + } + /** Opens a connection and sends everything... */ class Runner implements Runnable { @@ -235,17 +298,20 @@ void sendUpdateStream() throws Exception { try { while (!queue.isEmpty()) { InputStream rspBody = null; + List docIds = List.of(); + String collection = null; try { - Update update; notifyQueueAndRunnersIfEmptyQueue(); - update = queue.poll(pollQueueTimeMillis, TimeUnit.MILLISECONDS); + Update update = queue.poll(pollQueueTimeMillis, TimeUnit.MILLISECONDS); if (update == null) { break; } - StreamingResponse responseListener = null; - responseListener = doSendUpdateStream(update); + SentStream sent = doSendUpdateStream(update); + StreamingResponse responseListener = sent.response(); + docIds = sent.docIds(); + collection = sent.collection(); // just wait for the headers, so the idle timeout is sensible int statusCode = responseListener.awaitResponse(idleTimeoutMillis); @@ -266,12 +332,16 @@ void sendUpdateStream() throws Exception { solrExc = new RemoteSolrException(basePath, statusCode, remoteError); } - handleError(solrExc); + handleError(solrExc, docIds, collection); } else { onSuccess(responseListener.getUnderlyingResponse(), rspBody); } stallDetection.incrementProcessedCount(); + } catch (OutOfMemoryError | InterruptedException e) { + throw e; + } catch (Throwable e) { + handleError(e, docIds, collection); } finally { try { consumeFully(rspBody); @@ -287,7 +357,7 @@ void sendUpdateStream() throws Exception { } } - protected abstract StreamingResponse doSendUpdateStream(Update update) + protected abstract SentStream doSendUpdateStream(Update update) throws IOException, InterruptedException; private void consumeFully(InputStream is) { @@ -544,6 +614,22 @@ public void handleError(Throwable ex) { log.error("error", ex); } + private void handleError(Throwable ex, List failedIds, String collection) { + if (errorHandler == null) { + handleError(ex); + return; + } + try { + if (failedIds.isEmpty()) { + errorHandler.onError(ex); + } else { + errorHandler.onError(ex, failedIds, collection); + } + } catch (Exception handlerEx) { + log.error("errorHandler threw while handling a failed update", handlerEx); + } + } + /** * Intended to be used as an extension point for doing post-processing after a request completes. * @@ -627,6 +713,7 @@ public abstract static class Builder { protected boolean streamDeletes; protected boolean closeHttpClient; protected long pollQueueTimeMillis; + protected UpdateErrorHandler errorHandler; /** * Initialize a Builder object, based on the provided URL and client. @@ -763,6 +850,26 @@ public Builder setPollQueueTime(long pollQueueTime, TimeUnit unit) { return this; } + /** + * Registers a handler invoked when a batch fails to reach the server. It receives the ids of + * the failed documents (via {@link UpdateErrorHandler#idOf}, default the {@code id} field) and + * the target collection, so callers can recover them -- for example by routing the ids to a + * retry queue or dead-letter topic. + * + *

The handler is invoked on the client's internal runner threads and may be called + * concurrently when several batches fail at once, so implementations must be thread-safe. The + * call happens inline on a runner thread, so implementations should return quickly and not + * block, or they will slow update throughput. An exception thrown by the handler is logged and + * does not stop the client. + * + *

If no handler is registered the client falls back to its default behavior of logging the + * error, and document ids are not extracted. + */ + public Builder withErrorHandler(UpdateErrorHandler errorHandler) { + this.errorHandler = errorHandler; + return this; + } + /** * Create a {@link ConcurrentUpdateBaseSolrClient} based on the provided configuration options. */ diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java index 8ebbd64bd804..fc11824775d7 100644 --- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java +++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java @@ -36,41 +36,43 @@ protected ConcurrentUpdateJdkSolrClient(ConcurrentUpdateJdkSolrClient.Builder bu } @Override - protected StreamingResponse doSendUpdateStream(Update update) { + protected SentStream doSendUpdateStream(Update update) { UpdateRequest req = update.request(); String collection = update.collection(); CompletableFuture> resp = client.requestInputStreamAsync(basePath, req, collection); - return new StreamingResponse() { + StreamingResponse response = + new StreamingResponse() { - @Override - public int awaitResponse(long timeoutMillis) throws Exception { - return resp.get(timeoutMillis, TimeUnit.MILLISECONDS).statusCode(); - } + @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 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 Object getUnderlyingResponse() { + return resp; + } - @Override - public void close() throws IOException { - // No-op: InputStream is managed by java.net.http.HttpClient - } - }; + @Override + public void close() throws IOException { + // No-op: InputStream is managed by java.net.http.HttpClient + } + }; + return new SentStream(response, idsForErrorReporting(req), collection); } public static class Builder extends ConcurrentUpdateBaseSolrClient.Builder { diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClientTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClientTest.java index 13e790b60b2b..9cfc922c975f 100644 --- a/solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClientTest.java +++ b/solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClientTest.java @@ -71,6 +71,21 @@ public ConcurrentUpdateBaseSolrClient outcomeCountingConcurrentClient( .build(); } + @Override + public ConcurrentUpdateBaseSolrClient errorHandlerConcurrentClient( + String serverUrl, + int queueSize, + int threadCount, + HttpSolrClient solrClient, + ConcurrentUpdateBaseSolrClient.UpdateErrorHandler errorHandler) { + return new ConcurrentUpdateJdkSolrClient.Builder(serverUrl, (HttpJdkSolrClient) solrClient) + .withQueueSize(queueSize) + .withThreadCount(threadCount) + .setPollQueueTime(0, TimeUnit.MILLISECONDS) + .withErrorHandler(errorHandler) + .build(); + } + public static class OutcomeCountingConcurrentUpdateSolrClient extends ConcurrentUpdateJdkSolrClient { private final AtomicInteger successCounter; diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java b/solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java index 29e73a44007d..7a4c387e4083 100644 --- a/solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java +++ b/solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java @@ -33,8 +33,10 @@ import java.net.http.HttpConnectTimeoutException; import java.util.Enumeration; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Properties; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -82,6 +84,13 @@ public abstract ConcurrentUpdateBaseSolrClient outcomeCountingConcurrentClient( AtomicInteger failureCounter, StringBuilder errors); + public abstract ConcurrentUpdateBaseSolrClient errorHandlerConcurrentClient( + String serverUrl, + int queueSize, + int threadCount, + HttpSolrClient solrClient, + ConcurrentUpdateBaseSolrClient.UpdateErrorHandler errorHandler); + /** Mock endpoint where the CUSS being tested in this class sends requests. */ public static class TestServlet extends HttpServlet implements JavaBinUpdateRequestCodec.StreamingUpdateHandler { @@ -273,6 +282,184 @@ public void testConcurrentUpdate() throws Exception { } } + /** + * Against a real Solr: a valid document succeeds and is not reported, while documents the server + * rejects (a non-numeric value for the pint "popularity" field) are reported by id -- so only the + * failed ids reach the handler. + */ + @Test + public void testFailedDocsAreRecoverableViaErrorHandler() throws Exception { + List failedIds = new CopyOnWriteArrayList<>(); + + try (var httpClient = solrClient(null); + var concurrentClient = + errorHandlerConcurrentClient( + solrTestRule.getBaseUrl(), + 10, + 2, + httpClient, + (ex, ids, collection) -> failedIds.addAll(ids))) { + + SolrInputDocument good = new SolrInputDocument(); + good.addField("id", "good-1"); + concurrentClient.add("collection1", good); + concurrentClient.blockUntilFinished(); + assertTrue("a successful doc must not be reported: " + failedIds, failedIds.isEmpty()); + + for (int i = 1; i <= 5; i++) { + SolrInputDocument bad = new SolrInputDocument(); + bad.addField("id", "bad-" + i); + bad.addField("popularity", "not-an-int"); // rejected: pint field + concurrentClient.add("collection1", bad); + } + concurrentClient.blockUntilFinished(); + } + + assertEquals("only the 5 rejected docs should be reported: " + failedIds, 5, failedIds.size()); + for (int i = 1; i <= 5; i++) { + assertTrue("missing bad-" + i + " in " + failedIds, failedIds.contains("bad-" + i)); + } + assertFalse("the successful doc must not be reported", failedIds.contains("good-1")); + } + + /** Overriding idOf reports failures by a uniqueKey field other than "id". */ + @Test + public void testErrorHandlerWithCustomIdField() throws Exception { + TestServlet.clear(); + TestServlet.setErrorCode(500); // every request fails server-side + + String serverUrl = solrTestRule.getBaseUrl() + "/cuss/foo"; + List failedIds = new CopyOnWriteArrayList<>(); + ConcurrentUpdateBaseSolrClient.UpdateErrorHandler handler = + new ConcurrentUpdateBaseSolrClient.UpdateErrorHandler() { + @Override + public void onError(Throwable ex, List ids, String collection) { + failedIds.addAll(ids); + } + + @Override + public String idOf(SolrInputDocument doc) { + Object v = doc.getFieldValue("record_uuid"); + return v == null ? null : v.toString(); + } + }; + + try (var httpClient = solrClient(null); + var concurrentClient = + errorHandlerConcurrentClient(serverUrl, 10, 2, httpClient, handler)) { + + for (int i = 1; i <= 5; i++) { + SolrInputDocument doc = new SolrInputDocument(); + doc.addField("id", "ignored-" + i); // not the uniqueKey the handler reads + doc.addField("record_uuid", "uuid-" + i); + concurrentClient.add("collection1", doc); + } + concurrentClient.blockUntilFinished(); + } + + assertEquals("all 5 docs should be reported by record_uuid", 5, failedIds.size()); + for (int i = 1; i <= 5; i++) { + assertTrue("missing uuid-" + i + " in " + failedIds, failedIds.contains("uuid-" + i)); + } + } + + /** + * An error not tied to identifiable documents (a send failure where the document has no + * resolvable id) reaches the general onError(Throwable) callback, not the per-id one. + */ + @Test + public void testGeneralErrorCallbackForNonDocumentError() throws Exception { + String unreachable = "http://localhost:1/solr"; // nothing listening -> connect failure + AtomicInteger richCalls = new AtomicInteger(); + AtomicInteger generalCalls = new AtomicInteger(); + ConcurrentUpdateBaseSolrClient.UpdateErrorHandler handler = + new ConcurrentUpdateBaseSolrClient.UpdateErrorHandler() { + @Override + public void onError(Throwable ex, List ids, String collection) { + richCalls.incrementAndGet(); + } + + @Override + public void onError(Throwable ex) { + generalCalls.incrementAndGet(); + } + }; + + try (var httpClient = solrClient(null); + var concurrentClient = + errorHandlerConcurrentClient(unreachable, 10, 2, httpClient, handler)) { + SolrInputDocument doc = new SolrInputDocument(); + doc.addField("name", "no id here"); // no id -> idOf returns null -> nothing to report per-id + concurrentClient.add("collection1", doc); + concurrentClient.blockUntilFinished(); + } + + assertTrue("general callback should fire for a non-document error", generalCalls.get() > 0); + assertEquals( + "the per-id callback should not fire for a non-document error", 0, richCalls.get()); + } + + /** + * A naive lambda handler only registers for per-id failures; an error not tied to identifiable + * documents is logged by the default general callback and does not stop the client. + */ + @Test + public void testNaiveLambdaSurvivesNonDocumentError() throws Exception { + String unreachable = "http://localhost:1/solr"; + List failedIds = new CopyOnWriteArrayList<>(); + + try (var httpClient = solrClient(null); + var concurrentClient = + errorHandlerConcurrentClient( + unreachable, 10, 2, httpClient, (ex, ids, collection) -> failedIds.addAll(ids))) { + SolrInputDocument doc = new SolrInputDocument(); + doc.addField("name", "no id here"); + concurrentClient.add("collection1", doc); + concurrentClient.blockUntilFinished(); // returns normally; default callback logged the error + } + + assertTrue( + "an error not tied to documents must not reach the per-id lambda: " + failedIds, + failedIds.isEmpty()); + } + + /** + * A handler that throws is contained and does not stop the client: every failed doc is still + * reported and blockUntilFinished() returns normally. + */ + @Test + public void testThrowingErrorHandlerDoesNotStopClient() throws Exception { + TestServlet.clear(); + TestServlet.setErrorCode(500); // every request fails server-side + + String serverUrl = solrTestRule.getBaseUrl() + "/cuss/foo"; + List seenIds = new CopyOnWriteArrayList<>(); + + try (var httpClient = solrClient(null); + var concurrentClient = + errorHandlerConcurrentClient( + serverUrl, + 10, + 2, + httpClient, + (ex, ids, collection) -> { + seenIds.addAll(ids); + throw new RuntimeException("handler blew up"); + })) { + + for (int i = 1; i <= 5; i++) { + SolrInputDocument doc = new SolrInputDocument(); + doc.addField("id", "doc-" + i); + concurrentClient.add("collection1", doc); + } + concurrentClient.blockUntilFinished(); + } + + for (int i = 1; i <= 5; i++) { + assertTrue("missing doc-" + i + " in " + seenIds, seenIds.contains("doc-" + i)); + } + } + @Test public void testCollectionParameters() throws IOException, SolrServerException { From f70d6ae489aadb41faad13bff1953595ef5359fd Mon Sep 17 00:00:00 2001 From: Serhiy Bzhezytskyy Date: Mon, 13 Jul 2026 22:53:41 +0300 Subject: [PATCH 2/2] SOLR-3284: trim onError(Throwable) javadoc to what it does Co-Authored-By: Claude Opus 4.8 (1M context) --- .../solr/client/solrj/impl/ConcurrentUpdateBaseSolrClient.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateBaseSolrClient.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateBaseSolrClient.java index 087155639a4d..ce4c8673bb42 100644 --- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateBaseSolrClient.java +++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateBaseSolrClient.java @@ -95,7 +95,7 @@ public interface UpdateErrorHandler { /** * Invoked for an error not tied to specific documents (e.g. a failure before any document was - * sent). Defaults to logging, so a registered handler never silently swallows such errors. + * sent). Defaults to logging. */ default void onError(Throwable ex) { LoggerFactory.getLogger(UpdateErrorHandler.class).error("update error", ex);