OAK-12077 : added joinUninterruptibly() in oak-commons (#2711)

* OAK-12077 : added joinUninterruptibly() in oak-commons

* OAK-12077 : added review comments

* OAK-12077 : updated sleep durations to sleep within joining period

* OAK-12077 : fixed assertion errors if occured in joiningThread

Author: rishabhdaim

oak-commons/src/main/java/org/apache/jackrabbit/oak/commons/internal/concurrent/UninterruptibleUtils.java

@@ -110,4 +110,39 @@ public static void sleepUninterruptibly(final long sleep, final TimeUnit unit) {
             }
         }
     }
+
+    /**
+     * Invokes {@link TimeUnit#timedJoin(Thread, long)} uninterruptibly.
+     * <p>
+     * This method repeatedly calls {@link TimeUnit#timedJoin(Thread, long)} until the
+     * specified timeout has elapsed or the target thread terminates, ignoring
+     * {@link InterruptedException} but remembering that an interruption
+     * occurred. When the method finally returns, it restores the current
+     * thread's interrupted status if any interruptions were detected.
+     *
+     * @param toJoin the thread to wait for; must not be {@code null}
+     * @throws NullPointerException     if {@code toJoin} or {@code unit} is {@code null}
+     * @throws IllegalArgumentException if {@code timeout} is negative
+     */
+    public static void joinUninterruptibly(final Thread toJoin) {
+
+        Objects.requireNonNull(toJoin, "thread to join is null");
+
+        boolean interrupted = false;
+
+        try {
+            for(;;) {
+                try {
+                    toJoin.join();
+                    return;
+                } catch (InterruptedException var6) {
+                    interrupted = true;
+                }
+            }
+        } finally {
+            if (interrupted) {
+                Thread.currentThread().interrupt();
+            }
+        }
+    }
 }

oak-commons/src/test/java/org/apache/jackrabbit/oak/commons/internal/concurrent/UninterruptibleUtilsTest.java

@@ -23,6 +23,7 @@
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * Unit cases for {@link UninterruptibleUtils}
@@ -162,4 +163,126 @@ public void testZeroSleepReturnsQuickly() {
         Assert.assertTrue("Zero sleep should return quickly", elapsedMillis < 50L);
     }
 
+    @Test
+    public void testNullThread() {
+        Assert.assertThrows(NullPointerException.class,
+                () -> UninterruptibleUtils.joinUninterruptibly(null));
+    }
+
+    @Test
+    public void testReturnsWhenThreadFinishesBeforeTimeout() throws Exception {
+        final long workMillis = 10L;
+        final Thread worker = new Thread(() -> {
+            try {
+                Thread.sleep(workMillis);
+            } catch (InterruptedException ignored) {}
+        });
+
+        worker.start();
+
+        long start = System.nanoTime();
+        UninterruptibleUtils.joinUninterruptibly(worker);
+        long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
+
+        Assert.assertFalse("Worker should be finished", worker.isAlive());
+        Assert.assertTrue("Join should not take excessively long",
+                elapsedMillis >= workMillis && elapsedMillis < 100L);
+    }
+
+    @Test
+    public void testJoinShouldWaitUntilThreadFinishes() {
+        final Thread worker = new Thread(() -> {
+            try {
+                Thread.sleep(20L);
+            } catch (InterruptedException ignored) {
+            }
+        });
+
+        worker.start();
+
+        long start = System.nanoTime();
+        UninterruptibleUtils.joinUninterruptibly(worker);
+        long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
+
+        Assert.assertTrue("Join should respect timeout", elapsedMillis >= 20L );
+    }
+
+    @Test
+    public void testJoinUninterruptiblyIgnoresInterruptsButRestoresFlag() throws Exception {
+        final Thread worker = new Thread(() -> {
+            try {
+                Thread.sleep(200L);
+            } catch (InterruptedException ignored) {
+            }
+        });
+
+        worker.start();
+
+        final AtomicReference<Throwable> t = new AtomicReference<>();
+
+        Thread joiningThread = new Thread(() -> {
+            try {
+                UninterruptibleUtils.joinUninterruptibly(worker);
+                // After returning, interrupted flag should be set if we interrupted during join
+                Assert.assertTrue("Interrupt flag should be restored", Thread.currentThread().isInterrupted());
+            } catch (Throwable e) {
+                t.set(e);
+            }
+        });
+
+        joiningThread.start();
+
+        // Let the joining thread enter join
+        Thread.sleep(5L);
+
+        // Interrupt while it is joining
+        joiningThread.interrupt();
+
+        joiningThread.join();
+
+        // fail if any exception occurred in the thread
+        if (t.get() != null) {
+            Assert.fail("Got exception: " + t.get());
+        }
+    }
+
+    @Test
+    public void testJoinUninterruptiblyMultipleInterruptsStillCompleteAndRestoreFlag() throws Exception {
+        final Thread worker = new Thread(() -> {
+            try {
+                Thread.sleep(300L);
+            } catch (InterruptedException ignored) {
+            }
+        });
+
+        worker.start();
+
+        final AtomicReference<Throwable> t = new AtomicReference<>();
+
+        Thread joiningThread = new Thread(() -> {
+            try {
+                UninterruptibleUtils.joinUninterruptibly(worker);
+                Assert.assertTrue("Interrupt flag should be restored after multiple interrupts",
+                        Thread.currentThread().isInterrupted());
+            }  catch (Throwable e) {
+                t.set(e);
+            }
+        });
+
+        joiningThread.start();
+
+        // Interrupt the joining thread multiple times while it is waiting
+        for (int i = 0; i < 3; i++) {
+            Thread.sleep(5L);
+            joiningThread.interrupt();
+        }
+
+        joiningThread.join();
+
+        // fail if any exception occurred in the thread
+        if (t.get() != null) {
+            Assert.fail("Got exception: " + t.get());
+        }
+    }
+
 }