An awaiting DCountDownLatch thread can hang if the latch is destroyed

Author: jdeppe-pivotal

src/main/java/dev/gemfire/dtype/internal/DCountDownLatchImpl.java

@@ -16,11 +16,11 @@ public class DCountDownLatchImpl extends AbstractDType implements DCountDownLatc
 
   private final DTypeCollectionsFunction AWAIT_FN = x -> {
     DCountDownLatchImpl latch = (DCountDownLatchImpl) x;
-    latch.ensureUsable();
     while (latch.count > 0) {
+      latch.ensureUsable();
       try {
         latch.waiters++;
-        latch.wait();
+        latch.wait(100);
       } catch (InterruptedException e) {
         throw new RuntimeException(e);
       } finally {

src/test/java/dev/gemfire/dtype/AbstractDCountDownLatchTest.java

@@ -2,6 +2,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatNoException;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.time.Duration;
 import java.util.ArrayList;
@@ -131,4 +132,18 @@ public void testCountDownConcurrency() {
     assertThat(ref1.getCount()).isEqualTo(latches - (iterations * 5));
   }
 
+  @Test
+  public void testDestroyReleasesAwaitingClient() throws Exception {
+    DCountDownLatch ref1 = getFactory().createDCountDownLatch(testName.getMethodName(), 1);
+
+    Future<?> future = executor.submit(() -> ref1.await());
+
+    DCountDownLatch ref2 = getFactory().createDCountDownLatch(testName.getMethodName(), 1);
+
+    ref2.destroy();
+
+    assertThatThrownBy(future::get).hasRootCauseInstanceOf(DTypeException.class)
+        .withFailMessage("countdown latch is destroyed");
+  }
+
 }

src/test/java/dev/gemfire/dtype/DCountDownLatchDUnitTest.java

@@ -8,10 +8,12 @@
 import static org.apache.geode.distributed.ConfigurationProperties.SERIALIZABLE_OBJECT_FILTER;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatNoException;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.util.Properties;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
@@ -97,6 +99,34 @@ public void testServerRestartRetriesExistingAwait() throws Exception {
     ref2.countDown();
 
     assertThatNoException().isThrownBy(() -> future.get(2, TimeUnit.SECONDS));
+    assertThat(ref1.getCount()).isEqualTo(0);
+  }
+
+  @Test
+  public void testServerStopDoesStopAwaitingClient() throws Exception {
+    String latchName = testName.getMethodName();
+    DCountDownLatch ref1 = getFactory().createDCountDownLatch(latchName, 1);
+    DCountDownLatch ref2 = getFactory().createDCountDownLatch(latchName, 1);
+    Future<Void> future = executor.submit(() -> ref1.await());
+
+    MemberVM primary = TestUtils.getServerForKey(testName.getMethodName(), server1, server2);
+    primary.stop();
+
+    // Check that the future is still waiting
+    assertThatThrownBy(() -> future.get(10, TimeUnit.SECONDS))
+        .isInstanceOf(TimeoutException.class);
+
+    ref2.countDown();
+
+    // Should not throw a TimeoutException and should be much shorter than the default lease
+    // expiration timeout (60s).
+    future.get(10, TimeUnit.SECONDS);
+
+    if (primary.equals(server1)) {
+      server1 = cluster.startServerVM(1, props, locator.getPort());
+    } else {
+      server2 = cluster.startServerVM(2, props, locator.getPort());
+    }
   }
 
 }