fix: Modify CommitterImpl to exit early in case of permanent failure …

…to avoid waiting on its own actions to complete. (#90)
googleapis · Jun 2, 2020 · 8858d58 · 8858d58
1 parent bd67e98
commit 8858d58
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 3 deletions.
diff --git a/...oud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/wire/CommitterImpl.java b/...oud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/wire/CommitterImpl.java
@@ -45,6 +45,9 @@ public class CommitterImpl extends ProxyService
   @GuardedBy("monitor.monitor")
   private boolean shutdown = false;
 
+  @GuardedBy("monitor.monitor")
+  private boolean hadPermanentError = false;
+
   @GuardedBy("monitor.monitor")
   private final CommitState state = new CommitState();
 
@@ -72,6 +75,7 @@ public CommitterImpl(CursorServiceStub stub, InitialCommitCursorRequest request)
   @Override
   protected void handlePermanentError(StatusException error) {
     try (CloseableMonitor.Hold h = monitor.enter()) {
+      hadPermanentError = true;
       shutdown = true;
       state.abort(error);
     }
@@ -90,9 +94,8 @@ protected void stop() {
             new Guard(monitor.monitor) {
               @Override
               public boolean isSatisfied() {
-                // Wait until the state is empty. It will be made empty by a call to state.abort()
-                // if an error occurs.
-                return state.isEmpty();
+                // Wait until the state is empty or a permanent error occurred.
+                return state.isEmpty() || hadPermanentError;
               }
             })) {}
   }

diff --git a/...pubsublite/src/test/java/com/google/cloud/pubsublite/internal/wire/CommitterImplTest.java b/...pubsublite/src/test/java/com/google/cloud/pubsublite/internal/wire/CommitterImplTest.java
@@ -36,6 +36,7 @@
 import com.google.cloud.pubsublite.ProjectNumber;
 import com.google.cloud.pubsublite.SubscriptionName;
 import com.google.cloud.pubsublite.SubscriptionPaths;
+import com.google.cloud.pubsublite.internal.ExtractStatus;
 import com.google.cloud.pubsublite.internal.StatusExceptionMatcher;
 import com.google.cloud.pubsublite.proto.CursorServiceGrpc;
 import com.google.cloud.pubsublite.proto.CursorServiceGrpc.CursorServiceStub;
@@ -44,6 +45,7 @@
 import com.google.cloud.pubsublite.proto.StreamingCommitCursorRequest;
 import com.google.common.util.concurrent.MoreExecutors;
 import io.grpc.ManagedChannel;
+import io.grpc.Status;
 import io.grpc.Status.Code;
 import io.grpc.StatusException;
 import io.grpc.inprocess.InProcessChannelBuilder;
@@ -178,4 +180,19 @@ public void multipleSentCompletedInOrder() {
 
     verify(permanentErrorHandler, times(0)).failed(any(), any());
   }
+
+  @Test
+  public void stopInCommitCallback() throws Exception {
+    ApiFuture<Void> future = committer.commitOffset(Offset.of(10));
+    CountDownLatch latch = new CountDownLatch(1);
+    ExtractStatus.addFailureHandler(future, (error) -> {
+      committer.stopAsync();
+      latch.countDown();
+    });
+    leakedResponseObserver.onError(Status.FAILED_PRECONDITION.asException());
+    latch.await();
+    assertFutureThrowsCode(future, Code.FAILED_PRECONDITION);
+    verify(permanentErrorHandler)
+        .failed(any(), argThat(new StatusExceptionMatcher(Code.FAILED_PRECONDITION)));
+  }
 }