SOLR-17294: ConcurrentUpdateSolrClient no longer detects "false-positive" stalls (#2461)

Author: markrmiller

solr/CHANGES.txt

@@ -265,6 +265,8 @@ Bug Fixes
 
 * SOLR-17740: When the V2 API is receiving raw files, it could sometimes skip the first byte. (David Smiley)
 
+* SOLR-17294: ConcurrentUpdateSolrClient no longer detects "false-positive" stalls. (Mark Miller, Jason Gerlowski)
+
 Dependency Upgrades
 ---------------------
 * SOLR-17471: Upgrade Lucene to 9.12.1. (Pierre Salagnac, Christine Poerschke)

solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateHttp2SolrClient.java

@@ -64,11 +64,12 @@ public class ConcurrentUpdateHttp2SolrClient extends SolrClient {
   private boolean shutdownClient;
   private boolean shutdownExecutor;
   private long pollQueueTimeMillis;
-  private long stallTimeMillis;
   private final boolean streamDeletes;
   private volatile boolean closed;
   private volatile CountDownLatch lock = null; // used to block everything
 
+  protected StallDetection stallDetection;
+
   private static class CustomBlockingQueue<E> implements Iterable<E> {
     private final BlockingQueue<E> queue;
     private final Semaphore available;
@@ -150,15 +151,19 @@ protected ConcurrentUpdateHttp2SolrClient(Builder builder) {
     this.basePath = builder.baseSolrUrl;
     this.defaultCollection = builder.defaultCollection;
     this.pollQueueTimeMillis = builder.pollQueueTimeMillis;
-    this.stallTimeMillis = Integer.getInteger("solr.cloud.client.stallTime", 15000);
+
+    // Initialize stall detection
+    long stallTimeMillis = Integer.getInteger("solr.cloud.client.stallTime", 15000);
 
     // make sure the stall time is larger than the polling time
     // to give a chance for the queue to change
     long minimalStallTimeMillis = pollQueueTimeMillis * 2;
-    if (minimalStallTimeMillis > this.stallTimeMillis) {
-      this.stallTimeMillis = minimalStallTimeMillis;
+    if (minimalStallTimeMillis > stallTimeMillis) {
+      stallTimeMillis = minimalStallTimeMillis;
     }
 
+    this.stallDetection = new StallDetection(stallTimeMillis, queue::size);
+
     if (builder.executorService != null) {
       this.scheduler = builder.executorService;
       this.shutdownExecutor = false;
@@ -168,6 +173,8 @@ protected ConcurrentUpdateHttp2SolrClient(Builder builder) {
               new SolrNamedThreadFactory("concurrentUpdateScheduler"));
       this.shutdownExecutor = true;
     }
+
+    // processedCount is now managed by StallDetection
   }
 
   /** Opens a connection and sends everything... */
@@ -289,6 +296,7 @@ void sendUpdateStream() throws Exception {
             } else {
               onSuccess(response, rspBody);
             }
+            stallDetection.incrementProcessedCount();
 
           } finally {
             try {
@@ -402,8 +410,6 @@ public NamedList<Object> request(final SolrRequest<?> request, String collection
       Update update = new Update(req, effectiveCollection);
       boolean success = queue.offer(update);
 
-      long lastStallTime = -1;
-      int lastQueueSize = -1;
       for (; ; ) {
         synchronized (runners) {
           // see if queue is half full, and we can add more runners
@@ -438,28 +444,7 @@ public NamedList<Object> request(final SolrRequest<?> request, String collection
         }
         if (!success) {
           // stall prevention
-          int currentQueueSize = queue.size();
-          if (currentQueueSize != lastQueueSize) {
-            // there's still some progress in processing the queue - not stalled
-            lastQueueSize = currentQueueSize;
-            lastStallTime = -1;
-          } else {
-            if (lastStallTime == -1) {
-              // mark a stall but keep trying
-              lastStallTime = System.nanoTime();
-            } else {
-              long currentStallTime =
-                  TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - lastStallTime);
-              if (currentStallTime > stallTimeMillis) {
-                throw new IOException(
-                    "Request processing has stalled for "
-                        + currentStallTime
-                        + "ms with "
-                        + queue.size()
-                        + " remaining elements in the queue.");
-              }
-            }
-          }
+          stallDetection.stallCheck();
         }
       }
     } catch (InterruptedException e) {
@@ -480,9 +465,6 @@ public synchronized void blockUntilFinished() throws IOException {
       waitForEmptyQueue();
       interruptRunnerThreadsPolling();
 
-      long lastStallTime = -1;
-      int lastQueueSize = -1;
-
       synchronized (runners) {
 
         // NOTE: if the executor is shut down, runners may never become empty. A scheduled task may
@@ -498,29 +480,11 @@ public synchronized void blockUntilFinished() throws IOException {
           // Need to check if the queue is empty before really considering this is finished
           // (SOLR-4260)
           int queueSize = queue.size();
-          // stall prevention
-          if (lastQueueSize != queueSize) {
-            // init, or no stall
-            lastQueueSize = queueSize;
-            lastStallTime = -1;
-          } else {
-            if (lastStallTime == -1) {
-              lastStallTime = System.nanoTime();
-            } else {
-              long currentStallTime =
-                  TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - lastStallTime);
-              if (currentStallTime > stallTimeMillis) {
-                throw new IOException(
-                    "Task queue processing has stalled for "
-                        + currentStallTime
-                        + " ms with "
-                        + queueSize
-                        + " remaining elements to process.");
-                //                Thread.currentThread().interrupt();
-                //                break;
-              }
-            }
+          // stall prevention - only if queue is not empty
+          if (queueSize > 0) {
+            stallDetection.stallCheck();
           }
+
           if (queueSize > 0 && runners.isEmpty()) {
             // TODO: can this still happen?
             log.warn(
@@ -558,8 +522,6 @@ public synchronized void blockUntilFinished() throws IOException {
   private void waitForEmptyQueue() throws IOException {
     boolean threadInterrupted = Thread.currentThread().isInterrupted();
 
-    long lastStallTime = -1;
-    int lastQueueSize = -1;
     while (!queue.isEmpty()) {
       if (ExecutorUtil.isTerminated(scheduler)) {
         log.warn(
@@ -592,28 +554,9 @@ private void waitForEmptyQueue() throws IOException {
               queue.size());
         }
       }
-      int currentQueueSize = queue.size();
-      // stall prevention
-      if (currentQueueSize != lastQueueSize) {
-        lastQueueSize = currentQueueSize;
-        lastStallTime = -1;
-      } else {
-        lastQueueSize = currentQueueSize;
-        if (lastStallTime == -1) {
-          lastStallTime = System.nanoTime();
-        } else {
-          long currentStallTime = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - lastStallTime);
-          if (currentStallTime > stallTimeMillis) {
-            throw new IOException(
-                "Task queue processing has stalled for "
-                    + currentStallTime
-                    + " ms with "
-                    + currentQueueSize
-                    + " remaining elements to process.");
-            //            threadInterrupted = true;
-            //            break;
-          }
-        }
+      // Only check for stalls if the queue is not empty
+      if (!queue.isEmpty()) {
+        stallDetection.stallCheck();
       }
     }
     if (threadInterrupted) {

solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClient.java

@@ -83,7 +83,6 @@ public class ConcurrentUpdateSolrClient extends SolrClient {
   final int threadCount;
   boolean shutdownExecutor = false;
   int pollQueueTimeMillis = 250;
-  int stallTimeMillis;
   private final boolean streamDeletes;
   private boolean internalHttpClient;
   private final int connectionTimeout;
@@ -95,6 +94,8 @@ public class ConcurrentUpdateSolrClient extends SolrClient {
   AtomicInteger blockLoops;
   AtomicInteger emptyQueueLoops;
 
+  protected StallDetection stallDetection;
+
   /**
    * Use builder to construct this class. Uses the supplied HttpClient to send documents to the Solr
    * server.
@@ -121,16 +122,20 @@ protected ConcurrentUpdateSolrClient(Builder builder) {
     this.connectionTimeout = builder.connectionTimeoutMillis;
     this.soTimeout = builder.socketTimeoutMillis;
     this.pollQueueTimeMillis = builder.pollQueueTime;
-    this.stallTimeMillis = Integer.getInteger("solr.cloud.client.stallTime", 15000);
     this.defaultCollection = builder.defaultCollection;
 
+    // Initialize stall detection
+    int stallTimeMillis = Integer.getInteger("solr.cloud.client.stallTime", 15000);
+
     // make sure the stall time is larger than the polling time
     // to give a chance for the queue to change
     int minimalStallTime = pollQueueTimeMillis * 2;
-    if (minimalStallTime > this.stallTimeMillis) {
-      this.stallTimeMillis = minimalStallTime;
+    if (minimalStallTime > stallTimeMillis) {
+      stallTimeMillis = minimalStallTime;
     }
 
+    this.stallDetection = new StallDetection(stallTimeMillis, () -> queue.size());
+
     if (builder.executorService != null) {
       this.scheduler = builder.executorService;
       this.shutdownExecutor = false;
@@ -147,6 +152,8 @@ protected ConcurrentUpdateSolrClient(Builder builder) {
       this.blockLoops = new AtomicInteger();
       this.emptyQueueLoops = new AtomicInteger();
     }
+
+    // processedCount is now managed by StallDetection
   }
 
   public Set<String> getUrlParamNames() {
@@ -397,6 +404,7 @@ public void writeTo(OutputStream out) throws IOException {
           } else {
             onSuccess(response);
           }
+          stallDetection.incrementProcessedCount();
 
         } finally {
           try {
@@ -527,8 +535,6 @@ public NamedList<Object> request(final SolrRequest<?> request, String collection
       Update update = new Update(req, collection);
       boolean success = queue.offer(update);
 
-      long lastStallTime = -1;
-      int lastQueueSize = -1;
       for (; ; ) {
         synchronized (runners) {
           // see if queue is half full, and we can add more runners
@@ -562,29 +568,7 @@ public NamedList<Object> request(final SolrRequest<?> request, String collection
           success = queue.offer(update, 100, TimeUnit.MILLISECONDS);
         }
         if (!success) {
-          // stall prevention
-          int currentQueueSize = queue.size();
-          if (currentQueueSize != lastQueueSize) {
-            // there's still some progress in processing the queue - not stalled
-            lastQueueSize = currentQueueSize;
-            lastStallTime = -1;
-          } else {
-            if (lastStallTime == -1) {
-              // mark a stall but keep trying
-              lastStallTime = System.nanoTime();
-            } else {
-              long currentStallTime =
-                  TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - lastStallTime);
-              if (currentStallTime > stallTimeMillis) {
-                throw new IOException(
-                    "Request processing has stalled for "
-                        + currentStallTime
-                        + "ms with "
-                        + queue.size()
-                        + " remaining elements in the queue.");
-              }
-            }
-          }
+          stallCheck();
         }
       }
     } catch (InterruptedException e) {
@@ -598,16 +582,17 @@ public NamedList<Object> request(final SolrRequest<?> request, String collection
     return dummy;
   }
 
+  void stallCheck() throws IOException {
+    stallDetection.stallCheck();
+  }
+
   public synchronized void blockUntilFinished() throws IOException {
     lock = new CountDownLatch(1);
     try {
 
       waitForEmptyQueue();
       interruptRunnerThreadsPolling();
 
-      long lastStallTime = -1;
-      int lastQueueSize = -1;
-
       synchronized (runners) {
 
         // NOTE: if the executor is shut down, runners may never become empty. A scheduled task may
@@ -625,29 +610,11 @@ public synchronized void blockUntilFinished() throws IOException {
           // Need to check if the queue is empty before really considering this is finished
           // (SOLR-4260)
           int queueSize = queue.size();
-          // stall prevention
-          if (lastQueueSize != queueSize) {
-            // init, or no stall
-            lastQueueSize = queueSize;
-            lastStallTime = -1;
-          } else {
-            if (lastStallTime == -1) {
-              lastStallTime = System.nanoTime();
-            } else {
-              long currentStallTime =
-                  TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - lastStallTime);
-              if (currentStallTime > stallTimeMillis) {
-                throw new IOException(
-                    "Task queue processing has stalled for "
-                        + currentStallTime
-                        + " ms with "
-                        + queueSize
-                        + " remaining elements to process.");
-                //                Thread.currentThread().interrupt();
-                //                break;
-              }
-            }
+          // stall prevention - only if queue is not empty
+          if (queueSize > 0) {
+            stallCheck();
           }
+
           if (queueSize > 0 && runners.isEmpty()) {
             // TODO: can this still happen?
             log.warn(
@@ -685,8 +652,6 @@ public synchronized void blockUntilFinished() throws IOException {
   private void waitForEmptyQueue() throws IOException {
     boolean threadInterrupted = Thread.currentThread().isInterrupted();
 
-    long lastStallTime = -1;
-    int lastQueueSize = -1;
     while (!queue.isEmpty()) {
       if (log.isDebugEnabled()) emptyQueueLoops.incrementAndGet();
       if (ExecutorUtil.isTerminated(scheduler)) {
@@ -720,28 +685,10 @@ private void waitForEmptyQueue() throws IOException {
               queue.size());
         }
       }
-      int currentQueueSize = queue.size();
-      // stall prevention
-      if (currentQueueSize != lastQueueSize) {
-        lastQueueSize = currentQueueSize;
-        lastStallTime = -1;
-      } else {
-        lastQueueSize = currentQueueSize;
-        if (lastStallTime == -1) {
-          lastStallTime = System.nanoTime();
-        } else {
-          long currentStallTime = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - lastStallTime);
-          if (currentStallTime > stallTimeMillis) {
-            throw new IOException(
-                "Task queue processing has stalled for "
-                    + currentStallTime
-                    + " ms with "
-                    + currentQueueSize
-                    + " remaining elements to process.");
-            //            threadInterrupted = true;
-            //            break;
-          }
-        }
+
+      // Only check for stalls if the queue is not empty
+      if (!queue.isEmpty()) {
+        stallCheck();
       }
     }
     if (threadInterrupted) {