fix: Fail the partition watcher if its consumer throws an exception (#…

…728)
googleapis · Jun 25, 2021 · c297c77 · c297c77
1 parent 5cf5783
commit c297c77
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 1 deletion.
diff --git a/...te/src/main/java/com/google/cloud/pubsublite/internal/wire/PartitionCountWatcherImpl.java b/...te/src/main/java/com/google/cloud/pubsublite/internal/wire/PartitionCountWatcherImpl.java
@@ -78,7 +78,12 @@ private void pollTopicConfig() {
     if (currentPartitionCount.isPresent() && currentPartitionCount.get().equals(partitionCount)) {
       return;
     }
-    partitionCountReceiver.accept(partitionCount);
+    try {
+      partitionCountReceiver.accept(partitionCount);
+    } catch (Throwable t) {
+      notifyFailed(ExtractStatus.toCanonical(t));
+      stop();
+    }
     // Notify started after we successfully receive the config once.
     if (!currentPartitionCount.isPresent()) {
       notifyStarted();

diff --git a/...rc/test/java/com/google/cloud/pubsublite/internal/wire/PartitionCountWatcherImplTest.java b/...rc/test/java/com/google/cloud/pubsublite/internal/wire/PartitionCountWatcherImplTest.java
@@ -70,6 +70,17 @@ public void testFirstCallFails() {
     verify(mockClient, times(1)).getTopicPartitionCount(path());
   }
 
+  @Test
+  public void testConsumerExcepts() {
+    when(mockClient.getTopicPartitionCount(path())).thenReturn(ApiFutures.immediateFuture(1L));
+    PartitionCountWatcher watcher = watcherFactory.newWatcher(mockConsumer);
+    doThrow(new IllegalArgumentException("bad batching settings")).when(mockConsumer).accept(1L);
+    watcher.startAsync();
+    assertThrows(IllegalStateException.class, watcher::awaitTerminated);
+    ApiExceptionMatcher.assertThrowableMatches(watcher.failureCause(), StatusCode.Code.INTERNAL);
+    verify(mockClient, times(1)).getTopicPartitionCount(path());
+  }
+
   @Test
   public void testCallsHandlerOnStart() {
     when(mockClient.getTopicPartitionCount(path())).thenReturn(ApiFutures.immediateFuture(1L));