SOLR-3284: let ConcurrentUpdateSolrClient report which docs failed to send

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

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<String> docIds = new ArrayList<>();
 
     public OutStream(
         String origCollection,
@@ -123,6 +128,10 @@ boolean belongToThisStream(SolrRequest<?> solrRequest, String collection) {
           && 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 @@ 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

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();

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<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 @@ 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<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 @@ void sendUpdateStream() throws Exception {
       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 @@ 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<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 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.
+     *
+     * <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.
      */

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<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 {