OAK-12072 : removed Guava's Monitor and Guard (#2721)

rishabhdaim · web-flow · commit d29089890217 · 2026-02-23T19:20:18.000+05:30
* OAK-12072 : removed Guava's Monitor and Guard * OAK-12072 : added unit cases for ChangeProcessor class
diff --git a/oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/ChangeProcessor.java b/oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/ChangeProcessor.java
@@ -29,8 +29,10 @@
 
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantLock;
 
 import javax.jcr.observation.Event;
 import javax.jcr.observation.EventIterator;
@@ -73,9 +75,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import org.apache.jackrabbit.guava.common.util.concurrent.Monitor;
-import org.apache.jackrabbit.guava.common.util.concurrent.Monitor.Guard;
-
 /**
  * A {@code ChangeProcessor} generates observation {@link javax.jcr.observation.Event}s
  * based on a {@link FilterProvider filter} and delivers them to an {@link EventListener}.
@@ -417,8 +416,8 @@ public boolean excludes(NodeState root, CommitInfo info) {
         });
     }
 
-    private final Monitor runningMonitor = new Monitor();
-    private final RunningGuard running = new RunningGuard(runningMonitor);
+    private final AtomicBoolean stopped = new AtomicBoolean(false);
+    private final ReentrantLock runningLock = new ReentrantLock(false);
 
     /**
      * Try to stop this change processor if running. This method will wait
@@ -437,18 +436,23 @@ public boolean excludes(NodeState root, CommitInfo info) {
      */
     public synchronized boolean stopAndWait(int timeOut, TimeUnit unit) {
         Validate.checkState(registration != null, "Change processor not started");
-        if (running.stop()) {
-            if (runningMonitor.enter(timeOut, unit)) {
-                registration.unregister();
-                runningMonitor.leave();
-                return true;
-            } else {
-                // Timed out
-                return false;
-            }
-        } else {
-            // Stopped already
+
+        if (!stopFlagSet()) {
+            // already stopped, return from here
+            return true;
+        }
+
+        if (!enterWithTimeout(timeOut, unit)) {
+            // timed out acquiring the running lock, return early
+            return false;
+        }
+
+        // lock has been acquired
+        try {
+            registration.unregister();
             return true;
+        } finally {
+            runningLock.unlock();
         }
     }
 
@@ -460,9 +464,15 @@ public synchronized boolean stopAndWait(int timeOut, TimeUnit unit) {
      */
     public synchronized void stop() {
         Validate.checkState(registration != null, "Change processor not started");
-        if (running.stop()) {
-            registration.unregister();
-            runningMonitor.leave();
+
+        if (stopFlagSet()) {
+            // Wait until no contentChanged is in the critical section
+            runningLock.lock();
+            try {
+                registration.unregister();
+            } finally {
+                runningLock.unlock();
+            }
         }
     }
 
@@ -503,19 +513,24 @@ public void contentChanged(@NotNull NodeState before,
                 long time = System.nanoTime();
                 boolean hasEvents = events.hasNext();
                 tracker.recordProducerTime(System.nanoTime() - time, TimeUnit.NANOSECONDS);
-                if (hasEvents && runningMonitor.enterIf(running)) {
-                    if (commitRateLimiter != null) {
-                        commitRateLimiter.beforeNonBlocking();
-                    }
+                if (hasEvents && enterIfRunning()) {
+                    // lock has been acquired
                     try {
-                        CountingIterator countingEvents = new CountingIterator(events);
-                        eventListener.onEvent(countingEvents);
-                        countingEvents.updateCounters(eventCount, eventDuration);
-                    } finally {
                         if (commitRateLimiter != null) {
-                            commitRateLimiter.afterNonBlocking();
+                            commitRateLimiter.beforeNonBlocking();
                         }
-                        runningMonitor.leave();
+                        try {
+                            CountingIterator countingEvents = new CountingIterator(events);
+                            eventListener.onEvent(countingEvents);
+                            countingEvents.updateCounters(eventCount, eventDuration);
+                        } finally {
+                            if (commitRateLimiter != null) {
+                                commitRateLimiter.afterNonBlocking();
+                            }
+                        }
+                    } finally {
+                        // unlock now
+                        runningLock.unlock();
                     }
                 }
             }
@@ -602,29 +617,6 @@ public void remove() {
         }
     }
 
-    private static class RunningGuard extends Guard {
-        private boolean stopped;
-
-        public RunningGuard(Monitor monitor) {
-            super(monitor);
-        }
-
-        @Override
-        public boolean isSatisfied() {
-            return !stopped;
-        }
-
-        /**
-         * @return  {@code true} if this call set this guard to stopped,
-         *          {@code false} if another call set this guard to stopped before.
-         */
-        public boolean stop() {
-            boolean wasStopped = stopped;
-            stopped = true;
-            return !wasStopped;
-        }
-    }
-
     @Override
     public String toString() {
         return "ChangeProcessor ["
@@ -634,7 +626,7 @@ public String toString() {
                 + ", eventCount=" + eventCount 
                 + ", eventDuration=" + eventDuration 
                 + ", commitRateLimiter=" + commitRateLimiter
-                + ", running=" + running.isSatisfied() + "]";
+                + ", running=" + !stopped.get() + "]";
     }
     
     /** for logging only **/
@@ -689,4 +681,56 @@ private FilterResult evalPrefilter(NodeState root, CommitInfo info, ChangeSet ch
             return FilterResult.INCLUDE;
         }
     }
+
+    // helper methods for lock/unlocking
+
+    private boolean stopFlagSet() {
+        // true only for the first caller that changes false -> true
+        return stopped.compareAndSet(false, true);
+    }
+
+    private boolean enterIfRunning() {
+        runningLock.lock();
+        boolean ok = false;
+        try {
+            ok = !stopped.get();   // guard: same as RunningGuard.isSatisfied()
+            return ok;
+        } finally {
+            if (!ok) {
+                runningLock.unlock();
+            }
+        }
+    }
+
+    private boolean enterWithTimeout(long timeout, TimeUnit unit) {
+
+        // non-fair fast path
+        if (runningLock.tryLock()) {
+            return true;
+        }
+
+        long timeoutNanos = unit.toNanos(timeout);
+
+        boolean interrupted = false;
+        try {
+            long start = System.nanoTime();
+            long remaining = timeoutNanos;
+
+            while (remaining > 0L) {
+                try {
+                    // timed out
+                    return runningLock.tryLock(remaining, TimeUnit.NANOSECONDS);
+                } catch (InterruptedException e) {
+                    interrupted = true;
+                    long elapsed = System.nanoTime() - start;
+                    remaining = timeoutNanos - elapsed;
+                }
+            }
+            return false; // timeout
+        } finally {
+            if (interrupted) {
+                Thread.currentThread().interrupt();
+            }
+        }
+    }
 }
diff --git a/oak-jcr/src/test/java/org/apache/jackrabbit/oak/jcr/observation/ChangeProcessorTest.java b/oak-jcr/src/test/java/org/apache/jackrabbit/oak/jcr/observation/ChangeProcessorTest.java
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.jackrabbit.oak.jcr.observation;
+
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.apache.jackrabbit.oak.spi.whiteboard.CompositeRegistration;
+import org.apache.jackrabbit.oak.api.ContentSession;
+import org.apache.jackrabbit.oak.namepath.NamePathMapper;
+import org.apache.jackrabbit.commons.observation.ListenerTracker;
+import org.apache.jackrabbit.oak.plugins.observation.filter.FilterProvider;
+import org.apache.jackrabbit.oak.stats.StatisticManager;
+import org.apache.jackrabbit.oak.plugins.observation.CommitRateLimiter;
+import org.apache.jackrabbit.oak.api.blob.BlobAccessProvider;
+import org.junit.Assert;
+
+import java.lang.reflect.Field;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Unit cases for {@link ChangeProcessor} class
+ */
+public class ChangeProcessorTest {
+
+    @Test
+    public void testStopSetStoppedFlagCorrectly() {
+        ChangeProcessor changeProcessor = createChangeProcessor();
+        CompositeRegistration registration = Mockito.mock(CompositeRegistration.class);
+        setRegistration(changeProcessor, registration);
+        // Assert stopped flag is false before calling stop()
+        Assert.assertFalse("stopped flag should be false before stop()", getStoppedFlag(changeProcessor).get());
+        // Now call stop(), which should throw due to leave() without enter()
+        changeProcessor.stop();
+        Mockito.verify(registration).unregister();
+        // Assert stopped flag is true
+        Assert.assertTrue("stopped flag should be true after stop()", getStoppedFlag(changeProcessor).get());
+
+
+    }
+
+    @Test
+    public void testToStringRunningStateBeforeAndAfterStop() {
+        ChangeProcessor changeProcessor = createChangeProcessor();
+        CompositeRegistration registration = Mockito.mock(CompositeRegistration.class);
+        setRegistration(changeProcessor, registration);
+
+        String toStringBeforeStop = changeProcessor.toString();
+        System.out.println("Before stop: " + toStringBeforeStop);
+        Assert.assertTrue("toString() before stop should indicate running=true: " + toStringBeforeStop,
+                toStringBeforeStop.contains("running=true"));
+
+        changeProcessor.stop();
+
+        String toStringAfterStop = changeProcessor.toString();
+        System.out.println("After stop: " + toStringAfterStop);
+        Assert.assertTrue("toString() after stop should indicate running=false: " + toStringAfterStop,
+                toStringAfterStop.contains("running=false"));
+    }
+
+    @Test
+    public void testStopIdempotency() {
+        ChangeProcessor changeProcessor = createChangeProcessor();
+        CompositeRegistration registration = Mockito.mock(CompositeRegistration.class);
+        setRegistration(changeProcessor, registration);
+        // Assert stopped flag is false before calling stop()
+        Assert.assertFalse("stopped flag should be false before stop()", getStoppedFlag(changeProcessor).get());
+        // Call stop() first time
+        changeProcessor.stop();
+        Assert.assertTrue("stopped flag should be true after first stop()", getStoppedFlag(changeProcessor).get());
+        Mockito.verify(registration).unregister();
+        // Call stop() second time
+        changeProcessor.stop();
+        Assert.assertTrue("stopped flag should remain true after second stop()", getStoppedFlag(changeProcessor).get());
+        // unregister should only be called once
+        Mockito.verify(registration, Mockito.times(1)).unregister();
+    }
+
+
+    // helper methods
+
+    private static ChangeProcessor createChangeProcessor() {
+        ContentSession contentSession = Mockito.mock(ContentSession.class);
+        NamePathMapper namePathMapper = Mockito.mock(NamePathMapper.class);
+        ListenerTracker tracker = Mockito.mock(ListenerTracker.class);
+        FilterProvider filter = Mockito.mock(FilterProvider.class);
+        StatisticManager statisticManager = Mockito.mock(StatisticManager.class);
+        int queueLength = 1;
+        CommitRateLimiter commitRateLimiter = Mockito.mock(CommitRateLimiter.class);
+        BlobAccessProvider blobAccessProvider = Mockito.mock(BlobAccessProvider.class);
+
+        return new ChangeProcessor(
+            contentSession,
+            namePathMapper,
+            tracker,
+            filter,
+            statisticManager,
+            queueLength,
+            commitRateLimiter,
+            blobAccessProvider
+        );
+    }
+
+    private static AtomicBoolean getStoppedFlag(ChangeProcessor changeProcessor) {
+        try {
+            Field stoppedField = ChangeProcessor.class.getDeclaredField("stopped");
+            stoppedField.setAccessible(true);
+            return (AtomicBoolean) stoppedField.get(changeProcessor);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static void setRegistration(ChangeProcessor changeProcessor, CompositeRegistration registration) {
+        try {
+            Field regField = ChangeProcessor.class.getDeclaredField("registration");
+            regField.setAccessible(true);
+            regField.set(changeProcessor, registration);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+}
