You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@solr.apache.org by ds...@apache.org on 2026/07/16 20:44:08 UTC
(solr) 04/05: SOLR-3284: let ConcurrentUpdateSolrClient report which docs failed to send (#4632)
This is an automated email from the ASF dual-hosted git repository.
dsmiley pushed a commit to branch branch_10x
in repository https://gitbox.apache.org/repos/asf/solr.git
commit c94a323efbed90f206bad8f601bd94aa3e3bd621
Author: Serhiy Bzhezytskyy <me...@serhiy-bzhezytskyy.com>
AuthorDate: Tue Jul 14 16:01:03 2026 +0300
SOLR-3284: let ConcurrentUpdateSolrClient report which docs failed to send (#4632)
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.
Co-authored-by: Claude Opus 4.8 (1M context) <no...@anthropic.com>
(cherry picked from commit 14e850357171b1fbf251ba3eb14e2f7628860aa1)
---
.../unreleased/solr-3284-cusc-failed-docs.yml | 12 ++
.../jetty/ConcurrentUpdateJettySolrClient.java | 28 ++-
.../jetty/ConcurrentUpdateJettySolrClientTest.java | 15 ++
.../solrj/impl/ConcurrentUpdateBaseSolrClient.java | 119 ++++++++++++-
.../solrj/impl/ConcurrentUpdateJdkSolrClient.java | 54 +++---
.../impl/ConcurrentUpdateJdkSolrClientTest.java | 15 ++
.../impl/ConcurrentUpdateSolrClientTestBase.java | 187 +++++++++++++++++++++
7 files changed, 390 insertions(+), 40 deletions(-)
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 00000000000..66849e93654
--- /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 7e8686f1a39..458f57004b7 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.IOException;
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 @@ public class ConcurrentUpdateJettySolrClient extends ConcurrentUpdateBaseSolrCli
}
@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 @@ public class ConcurrentUpdateJettySolrClient extends ConcurrentUpdateBaseSolrCli
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 @@ public class ConcurrentUpdateJettySolrClient extends ConcurrentUpdateBaseSolrCli
private final OutputStreamRequestContent content;
private final InputStreamResponseListener responseListener;
private final boolean isXml;
+ private final List<String> docIds = new ArrayList<>();
public OutStream(
String origCollection,
@@ -123,6 +128,10 @@ public class ConcurrentUpdateJettySolrClient extends ConcurrentUpdateBaseSolrCli
&& Objects.equals(origCollection, collection);
}
+ List<String> getDocIds() {
+ return docIds;
+ }
+
public void write(byte[] b) throws IOException {
this.content.getOutputStream().write(b);
}
@@ -210,8 +219,11 @@ public class ConcurrentUpdateJettySolrClient extends ConcurrentUpdateBaseSolrCli
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 df28d17f3b0..10808dedba4 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 class ConcurrentUpdateJettySolrClientTest extends ConcurrentUpdateSolrCli
.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 752e7131509..ce4c8673bb4 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.io.InputStream;
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.SolrServerException;
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<String> failedIds, String collection);
+
+ /**
+ * Invoked for an error not tied to specific documents (e.g. a failure before any document was
+ * sent). Defaults to logging.
+ */
+ 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<E> implements Iterable<E> {
private final BlockingQueue<E> queue;
private final Semaphore available;
@@ -158,6 +195,7 @@ public abstract class ConcurrentUpdateBaseSolrClient extends SolrClient {
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 @@ public abstract class ConcurrentUpdateBaseSolrClient extends SolrClient {
/** 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<String> 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<String> idsForErrorReporting(UpdateRequest request) {
+ if (errorHandler == null || request.getDocuments() == null) {
+ return List.of();
+ }
+ List<String> 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 @@ public abstract class ConcurrentUpdateBaseSolrClient extends SolrClient {
try {
while (!queue.isEmpty()) {
InputStream rspBody = null;
+ List<String> 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 @@ public abstract class ConcurrentUpdateBaseSolrClient extends SolrClient {
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 @@ public abstract class ConcurrentUpdateBaseSolrClient extends SolrClient {
}
}
- 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 abstract class ConcurrentUpdateBaseSolrClient extends SolrClient {
log.error("error", ex);
}
+ private void handleError(Throwable ex, List<String> 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 class ConcurrentUpdateBaseSolrClient extends SolrClient {
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 abstract class ConcurrentUpdateBaseSolrClient extends SolrClient {
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.
+ *
+ * <p>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.
+ *
+ * <p>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 8ebbd64bd80..fc11824775d 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 @@ public class ConcurrentUpdateJdkSolrClient extends ConcurrentUpdateBaseSolrClien
}
@Override
- protected StreamingResponse doSendUpdateStream(Update update) {
+ protected SentStream doSendUpdateStream(Update update) {
UpdateRequest req = update.request();
String collection = update.collection();
CompletableFuture<HttpResponse<InputStream>> 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 13e790b60b2..9cfc922c975 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 class ConcurrentUpdateJdkSolrClientTest extends ConcurrentUpdateSolrClien
.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 29e73a44007..7a4c387e408 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.SocketTimeoutException;
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 class ConcurrentUpdateSolrClientTestBase extends SolrTestCaseJ4
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 abstract class ConcurrentUpdateSolrClientTestBase extends SolrTestCaseJ4
}
}
+ /**
+ * 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<String> 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<String> failedIds = new CopyOnWriteArrayList<>();
+ ConcurrentUpdateBaseSolrClient.UpdateErrorHandler handler =
+ new ConcurrentUpdateBaseSolrClient.UpdateErrorHandler() {
+ @Override
+ public void onError(Throwable ex, List<String> 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<String> 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<String> 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<String> 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 {