SOLR-12833: Use timed-out lock in DistributedUpdateProcessor

solr/core/src/java/org/apache/solr/update/VersionBucket.java

@@ -16,17 +16,72 @@
  */
 package org.apache.solr.update;
 
+import com.google.common.annotations.VisibleForTesting;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
 // TODO: make inner?
 // TODO: store the highest possible in the index on a commit (but how to not block adds?)
 // TODO: could also store highest possible in the transaction log after a commit.
 // Or on a new index, just scan "version" for the max?
 /** @lucene.internal */
 public class VersionBucket {
+  private int versionLockInMill;
+
+  public VersionBucket(int versionLockInMill) {
+    this.versionLockInMill = versionLockInMill;
+  }
+
+  private final Lock lock = new ReentrantLock(true);
+  private final Condition condition = lock.newCondition();
+
   public long highest;
 
   public void updateHighest(long val) {
     if (highest != 0) {
       highest = Math.max(highest, Math.abs(val));
     }
   }
+
+  public int getVersionLockInMill() {
+    return versionLockInMill;
+  }
+
+  @VisibleForTesting
+  public void setVersionLockInMill(int versionLockInMill) {
+    this.versionLockInMill = versionLockInMill;
+  }
+
+  public boolean tryLock() {
+    if (versionLockInMill > 0) {
+      try {
+        return lock.tryLock(versionLockInMill, TimeUnit.MILLISECONDS);
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        return false;
+      }
+    } else {
+      lock.lock();
+      return true;
+    }
+  }
+
+  public void unlock() {
+    lock.unlock();
+  }
+
+  public void wakeUpAll() {
+    condition.signalAll();
+  }
+
+  public void awaitNanos(long nanosTimeout) {
+    try {
+      condition.awaitNanos(nanosTimeout);
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new RuntimeException(e);
+    }
+  }
 }

solr/core/src/java/org/apache/solr/update/VersionInfo.java

@@ -45,15 +45,22 @@
 import static org.apache.solr.common.params.CommonParams.VERSION_FIELD;
 
 public class VersionInfo {
-
   private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+  private static final String SYS_PROP_VERSION_LOCK_TIMEOUT_IN_MILL = "versionLockTimeoutInMill";
+
+  /**
+   * same as default client read timeout: 10 mins
+   */
+  private static final int DEFAULT_VERSION_LOCK_TIMEOUT_IN_MILL = 19 * 60 * 1000;
 
   private final UpdateLog ulog;
   private final VersionBucket[] buckets;
   private SchemaField versionField;
   private SchemaField idField;
   final ReadWriteLock lock = new ReentrantReadWriteLock(true);
 
+  private int versionLockInMill;
+
   /**
    * Gets and returns the {@link org.apache.solr.common.params.CommonParams#VERSION_FIELD} from the specified
    * schema, after verifying that it is indexed, stored, and single-valued.  
@@ -94,9 +101,11 @@ public VersionInfo(UpdateLog ulog, int nBuckets) {
     IndexSchema schema = ulog.uhandler.core.getLatestSchema(); 
     versionField = getAndCheckVersionField(schema);
     idField = schema.getUniqueKeyField();
+    versionLockInMill = ulog.uhandler.core.getSolrConfig().getInt("updateHandler/versionLock/timeoutInMill",
+        Integer.parseInt(System.getProperty(SYS_PROP_VERSION_LOCK_TIMEOUT_IN_MILL, "" + DEFAULT_VERSION_LOCK_TIMEOUT_IN_MILL)));
     buckets = new VersionBucket[ BitUtil.nextHighestPowerOfTwo(nBuckets) ];
     for (int i=0; i<buckets.length; i++) {
-      buckets[i] = new VersionBucket();
+      buckets[i] = new VersionBucket(versionLockInMill);
     }
   }