Permanently failed Subscriber is never detected: binding stays "running", health stays UP, messages silently stop

[kmharish]

Issue Description

We had a incident where our stream binder consumers stopped processing for approx 1 hr with no alert firing. The app looked healthy the entire time, binding state running, actuator health UP. The only trace was the pubsub client's own internal log:

ERROR c.g.c.p.v.StreamingSubscriberConnection : terminated streaming with exception

Root cause after digging: Subscriber is a Guava ApiService. When the streaming pull hits a non-retryable error it transitions to FAILED and stops permanently. Nothing in spring-cloud-gcp listens for that transition, as far as I can tell there is no ApiService.Listener.failed() override anywhere in the codebase:

• PubSubSubscriberTemplate.subscribeAndConvert() calls subscriber.startAsync() and returns; the lifecycle is dropped there
• PubSubInboundChannelAdapter.addListeners() only attaches a listener if a HealthTrackerRegistry is configured, and that listener (HealthTrackerRegistryImpl) only overrides terminated()
• PubSubHealthIndicator probes with a unary pullAsync, which is a different code path from streaming pull, so health stays UP even when every streaming subscriber in the JVM is dead

So a failed subscriber is completely invisible: isRunning() returns true, the binding reports running, health reports UP, throughput is zero, and the only recovery is restarting the app.

Version: 7.x, but the relevant code is unchanged on current main so I'm reporting against main.

Sample

This test passes on current main, which is exactly the problem, no listener is ever attached to the Subscriber, and even with health tracking on, a failed() transition changes nothing:

package com.google.cloud.spring.pubsub.integration.inbound;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import com.google.api.core.ApiService;
import com.google.api.gax.core.FixedExecutorProvider;
import com.google.cloud.monitoring.v3.MetricServiceClient;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.cloud.spring.pubsub.core.health.HealthTrackerRegistry;
import com.google.cloud.spring.pubsub.core.health.HealthTrackerRegistryImpl;
import com.google.cloud.spring.pubsub.core.subscriber.PubSubSubscriberOperations;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.function.Consumer;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.MessageChannel;

@ExtendWith(MockitoExtension.class)
class SubscriberFailureReproTests {

  private final TestUtils.TestApplicationContext context = TestUtils.createTestApplicationContext();

  @Mock private PubSubSubscriberOperations subscriberOperations;
  @Mock private MessageChannel outputChannel;

  @SuppressWarnings("unchecked")
  private PubSubInboundChannelAdapter adapterFor(Subscriber subscriber) {
    when(this.subscriberOperations.subscribeAndConvert(
            anyString(), any(Consumer.class), any(Class.class)))
        .thenReturn(subscriber);

    PubSubInboundChannelAdapter adapter =
        new PubSubInboundChannelAdapter(this.subscriberOperations, "testSubscription");
    adapter.setOutputChannel(this.outputChannel);
    adapter.setBeanFactory(this.context);
    return adapter;
  }

  // Default config: no listener at all on the Subscriber, so failure can't be observed.
  @Test
  void defaultConfig_noListenerIsAttached() {
    Subscriber subscriber = mock(Subscriber.class);
    PubSubInboundChannelAdapter adapter = adapterFor(subscriber);

    adapter.start();

    verify(subscriber, never()).addListener(any(), any());
    assertThat(adapter.isRunning()).isTrue();
  }

  // With health tracking: the only listener overrides terminated(), so failed() is ignored.
  @Test
  void withHealthTracking_failedTransitionIsIgnored() {
    Subscriber subscriber = mock(Subscriber.class);
    when(subscriber.getSubscriptionNameString())
        .thenReturn("projects/test-project/subscriptions/testSubscription");

    HealthTrackerRegistry registry =
        new HealthTrackerRegistryImpl(
            "test-project",
            mock(MetricServiceClient.class),
            1, 1, 1,
            FixedExecutorProvider.create(Executors.newSingleThreadScheduledExecutor()));

    PubSubInboundChannelAdapter adapter = adapterFor(subscriber);
    adapter.setHealthTrackerRegistry(registry);
    adapter.start();

    ArgumentCaptor<ApiService.Listener> captor = ArgumentCaptor.forClass(ApiService.Listener.class);
    verify(subscriber).addListener(captor.capture(), any(Executor.class));

    // what StreamingSubscriberConnection does on a non-retryable stream error
    captor.getValue()
        .failed(ApiService.State.RUNNING, new IllegalStateException("streaming pull terminated"));

    // nothing reacted; adapter still claims to be running
    assertThat(adapter.isRunning()).isTrue();
  }
}

./mvnw test -pl spring-cloud-gcp-pubsub -am -Dtest=SubscriberFailureReproTests

[jinseopkim0]

Thank you for opening this issue. If you have a support contract with Google, please create an issue in the support console instead of filing on GitHub. This will ensure a timely response.