dsmiley commented on code in PR #4632:
URL: https://github.com/apache/solr/pull/4632#discussion_r3572796742
##########
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:
Just use "id" please. We've abused that CommonParams so much; it's
*supposed* be only _params_.
##########
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;
Review Comment:
this wording here is indicative of a bug fix in the process of development
but isn't useful in the delivered documentation. Of course "not just the
first". AI loves to do this.
Likewise saying "only IDs are retains and why" is IMO not something to put
here.
##########
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:
lets improve this slightly to add a doc successfully, call
blockUntilFinished, then proceed to mess things up and only have the failed
docs get reported.
Note this would mean switching from a bogus URL to real Solr but put fields
on invalid docs that don't fit the schema. TestServlet isn't needed in this
test, I think.
##########
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:
please rename `newSentStream`
##########
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:
Can't we catch these and throw without instanceof?
note: java lang 17 here
##########
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(
+ StreamingResponse response, List<String> docIds, String collection) {
+ return new SentStream(response, docIds, collection);
+ }
+
+ /** True when an error handler is registered, so transports skip id
extraction otherwise. */
+ protected boolean hasErrorHandler() {
+ return errorHandler != null;
+ }
+
+ /** Extracts a document id via the registered handler (defaults to the
{@code id} field). */
+ protected String extractId(SolrInputDocument doc) {
+ return errorHandler == null ? null : errorHandler.extractId(doc);
+ }
+
+ /**
+ * The ids of a request's documents, for failure reporting. Empty when no
error handler is
+ * registered (so the documents are never read), preserving streaming memory
efficiency.
+ */
+ protected List<String> docIds(UpdateRequest request) {
+ if (errorHandler == null || request.getDocuments() == null) {
+ return List.of();
+ }
Review Comment:
again; I see a fragile relationship. A method with a straight-forward name
yet is unusable outside of error-tracking purposes.
##########
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:
or just remove this extracted method... I don't see the point of it. If it
wasn't static, I might surmise why you did this.
##########
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(
+ StreamingResponse response, List<String> docIds, String collection) {
+ return new SentStream(response, docIds, collection);
+ }
+
+ /** True when an error handler is registered, so transports skip id
extraction otherwise. */
+ protected boolean hasErrorHandler() {
+ return errorHandler != null;
+ }
+
+ /** Extracts a document id via the registered handler (defaults to the
{@code id} field). */
+ protected String extractId(SolrInputDocument doc) {
+ return errorHandler == null ? null : errorHandler.extractId(doc);
+ }
Review Comment:
this is a fragile relationship IMO. As-written, this method is only useful
if it's being used for error tracking purpose. But it's name is not indicative
of that, so it might eventually be used for logging or whatever other purposes.
One solution is to inline the logic. Another solution is to make this method
the place where the ID is actually extracted... and so remove that
responsibility from the error handler. A rare user who doens't use "id" must
subclass this 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: [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]
