serhiy-bzhezytskyy commented on code in PR #4632:
URL: https://github.com/apache/solr/pull/4632#discussion_r3573798835
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateBaseSolrClient.java:
##########
@@ -71,6 +75,35 @@ public abstract class ConcurrentUpdateBaseSolrClient extends
SolrClient {
protected StallDetection stallDetection;
+ private final UpdateErrorHandler errorHandler;
+
+ /**
+ * Callback invoked when an update batch fails to reach the server,
receiving the ids of the
+ * documents that did not make it. Register one via {@link
Builder#withErrorHandler}. This is the
+ * hook that lets a caller recover the failed documents (e.g. route their
ids to a retry queue or
+ * dead-letter topic). Only the id is retained -- not the whole document --
to preserve the
+ * streaming memory efficiency of this client. Implementations must be
thread-safe; see {@link
+ * Builder#withErrorHandler} for the full threading contract.
+ */
+ @FunctionalInterface
+ public interface UpdateErrorHandler {
+ /**
+ * @param ex the error that occurred while sending the batch
+ * @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);
+
+ /**
+ * Extracts the id of a document about to be sent, so a failure can be
reported by id. Defaults
+ * to the {@code id} field; override for a different uniqueKey.
+ */
+ default String extractId(SolrInputDocument doc) {
+ Object id = doc.getFieldValue(CommonParams.ID);
Review Comment:
Done.
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateBaseSolrClient.java:
##########
@@ -187,6 +221,51 @@ protected ConcurrentUpdateBaseSolrClient(Builder builder) {
/** Class representing an UpdateRequest and an optional collection. */
protected record Update(UpdateRequest request, String collection) {}
+ /**
+ * The result of sending one or more updates as a single stream: the
response to await, plus the
+ * ids of the documents actually sent and their collection. A transport may
coalesce several
+ * queued updates into one stream (e.g. the Jetty client), so it reports the
ids of all of them
+ * here -- letting the error handler learn every failed document, not just
the first. Only ids are
+ * retained (see {@link UpdateErrorHandler#extractId}), preserving streaming
memory efficiency;
+ * the list is empty when no error handler is registered. Subclasses build
this via {@link
+ * #sentStream}.
+ */
+ protected record SentStream(StreamingResponse response, List<String> docIds,
String collection) {}
+
+ /** Factory for {@link SentStream}, callable by {@link #doSendUpdateStream}
implementations. */
+ protected static SentStream sentStream(
Review Comment:
Removed it — made the record public instead: a protected record constructor
isn't reachable from the Jetty package.
##########
solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateBaseSolrClient.java:
##########
@@ -266,12 +348,20 @@ void sendUpdateStream() throws Exception {
solrExc = new RemoteSolrException(basePath, statusCode,
remoteError);
}
- handleError(solrExc);
+ handleError(solrExc, docIds, collection);
} else {
onSuccess(responseListener.getUnderlyingResponse(), rspBody);
}
stallDetection.incrementProcessedCount();
+ } catch (Throwable e) {
+ if (e instanceof OutOfMemoryError) {
+ throw (OutOfMemoryError) e;
+ }
+ if (e instanceof InterruptedException) {
+ throw (InterruptedException) e;
+ }
Review Comment:
Done.
##########
solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java:
##########
@@ -273,6 +282,112 @@ public void testConcurrentUpdate() throws Exception {
}
}
+ /** A handler registered via the Builder receives the ids of every failed
batch's documents. */
+ @Test
+ public void testFailedDocsAreRecoverableViaErrorHandler() throws Exception {
+ TestServlet.clear();
+ TestServlet.setErrorCode(500); // every request fails server-side
+
+ String serverUrl = solrTestRule.getBaseUrl() + "/cuss/foo";
+ List<String> failedIds = new CopyOnWriteArrayList<>();
+
Review Comment:
Rewritten per your sketch: good doc first, then schema-rejected docs; only
the failures are reported.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
