OAK-12033 : replaced Guava's Monitor with ReentrantLock (#2654)

Author: rishabhdaim

oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/file/FileStoreIT.java

@@ -42,9 +42,10 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
-import org.apache.jackrabbit.guava.common.util.concurrent.Monitor;
-import org.apache.jackrabbit.guava.common.util.concurrent.Monitor.Guard;
 import org.apache.jackrabbit.oak.api.Blob;
 import org.apache.jackrabbit.oak.plugins.memory.AbstractBlob;
 import org.apache.jackrabbit.oak.plugins.memory.ArrayBasedBlob;
@@ -257,20 +258,23 @@ public void blockingBlob() throws Exception {
         /* A blob that blocks on read until unblocked */
         class BlockingBlob extends AbstractBlob {
             private final AtomicBoolean blocking = new AtomicBoolean(true);
-            private final Monitor readMonitor = new Monitor();
+            private final Lock lock = new ReentrantLock();
+            private final Condition readingCondition = lock.newCondition();
             private boolean reading = false;
 
             public boolean waitForRead(int time, TimeUnit unit) throws InterruptedException {
-                readMonitor.enter();
+                long nanos = unit.toNanos(time);
+                lock.lock();
                 try {
-                    return readMonitor.waitFor(new Guard(readMonitor) {
-                        @Override
-                        public boolean isSatisfied() {
-                            return reading;
+                    while (!reading) {
+                        if (nanos <= 0L) {
+                            return false; // timed out
                         }
-                    }, time, unit);
+                        nanos = readingCondition.awaitNanos(nanos);
+                    }
+                    return true; // reading == true
                 } finally {
-                    readMonitor.leave();
+                    lock.unlock();
                 }
             }
 
@@ -295,13 +299,15 @@ public int read(@NotNull byte[] b, int off, int len) throws IOException {
 
                     private int readOrEnd() {
                         if (blocking.get()) {
-                            if (!reading) {
-                                readMonitor.enter();
-                                try {
+                            lock.lock();
+                            try {
+                                if (!reading) {
                                     reading = true;
-                                } finally {
-                                    readMonitor.leave();
+                                    // wake up any threads waiting in waitForRead
+                                    readingCondition.signalAll();
                                 }
+                            } finally {
+                                lock.unlock();
                             }
                             return 0;
                         } else {

oak-store-document/src/test/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStoreIT.java

@@ -17,7 +17,6 @@
 package org.apache.jackrabbit.oak.plugins.document;
 
 import java.io.InputStream;
-import java.util.Date;
 import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
@@ -26,8 +25,9 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
-
-import org.apache.jackrabbit.guava.common.util.concurrent.Monitor;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 import org.apache.jackrabbit.oak.api.CommitFailedException;
 import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser;
@@ -454,20 +454,23 @@ private int getNodesFindCountOfAnUpdate(CountingDocumentStore cds, DocumentNodeS
      */
     class BlockingBlob extends AbstractBlob {
         private final AtomicBoolean blocking = new AtomicBoolean(true);
-        private final Monitor readMonitor = new Monitor();
+        private final Lock lock = new ReentrantLock();
+        private final Condition readingCondition = lock.newCondition();
         private boolean reading = false;
 
         boolean waitForRead(int time, TimeUnit unit) throws InterruptedException {
-            readMonitor.enter();
+            long nanos = unit.toNanos(time);
+            lock.lock();
             try {
-                return readMonitor.waitFor(new Monitor.Guard(readMonitor) {
-                    @Override
-                    public boolean isSatisfied() {
-                        return reading;
+                while (!reading) {
+                    if (nanos <= 0L) {
+                        return false; // timed out
                     }
-                }, time, unit);
+                    nanos = readingCondition.awaitNanos(nanos);
+                }
+                return true; // reading == true
             } finally {
-                readMonitor.leave();
+                lock.unlock();
             }
         }
 
@@ -483,13 +486,15 @@ public InputStream getNewStream() {
                 @Override
                 public int read() {
                     while (blocking.get()) {
-                        if (!reading) {
-                            readMonitor.enter();
-                            try {
+                        lock.lock();
+                        try {
+                            if (!reading) {
                                 reading = true;
-                            } finally {
-                                readMonitor.leave();
+                                // wake up anyone waiting in waitForRead()
+                                readingCondition.signalAll();
                             }
+                        } finally {
+                            lock.unlock();
                         }
                     }
                     return -1;