pekko-connectors-google-cloud-pub-sub-grpc: StreamingPull does not reconnect on server-side disconnect

`GooglePubSub.subscribe` (Streaming Pull) silently stops receiving messages when the gRPC server closes the connection. Unlike the official Google Cloud Pub/Sub client libraries, pekko-connectors does not implement automatic reconnection. ## Environment - pekko-connectors-google-cloud-pub-sub-grpc: **1.2.0** - pekko-stream: **1.4.0** - JVM: 21 - Runtime: Google Cloud Run ## Problem Google Pub/Sub servers periodically close StreamingPull connections for expected reasons (idle timeout, server-side rebalancing, etc.). When this happens, the gRPC stream fails with: ``` io.grpc.StatusRuntimeException: UNAVAILABLE: The StreamingPull stream closed for an expected reason and should be recreated, which is done automatically if using Cloud Pub/Sub client libraries. ``` The error message states reconnection is automatic "if using Cloud Pub/Sub client libraries" — however, pekko-connectors is **not** a Google client library. It uses low-level gRPC stubs directly and does **not** implement reconnection logic. When the gRPC stream is closed: 1. `PekkoNettyGrpcClientGraphStage` receives `onCallClosed` with `UNAVAILABLE` status 2. The Pekko Source emits `onError` (stream failure) 3. `Supervision.Resume` does **not** help — it only handles errors in downstream processing stages, not source failure 4. The entire Pekko Stream terminates silently 5. No more messages are received until the application is restarted This is particularly problematic on Cloud Run where the instance stays alive (healthcheck passes) but the subscriber is dead. ## Steps to Reproduce 1. Create a Streaming Pull subscriber using `GooglePubSub.subscribe`: ```scala GooglePubSub .subscribe(request, pollInterval = 1.second) .mapAsync(parallelism)(processMessage) .to(GooglePubSub.acknowledge(parallelism = 1)) .run() ``` 2. Wait for the server to close the connection (typically after idle period) 3. Observe that no new messages are received 4. The application continues running (healthcheck passes) but the stream is dead ## Misleading Error Message The gRPC error message says: > "...which is done automatically if using Cloud Pub/Sub client libraries" This is misleading in the context of pekko-connectors because: - Users may assume pekko-connectors qualifies as a "client library" that handles reconnection - The official Google `google-cloud-pubsub` Java library (`com.google.cloud:google-cloud-pubsub`) **does** reconnect automatically - pekko-connectors uses raw gRPC stubs (`com.google.pubsub.v1`) without reconnection logic ## Current Workaround Wrap `GooglePubSub.subscribe` with `RestartSource.withBackoff`: ```scala val restartSettings = RestartSettings( minBackoff = 3.seconds, maxBackoff = 30.seconds, randomFactor = 0.2 ) RestartSource .withBackoff(restartSettings) { () => GooglePubSub.subscribe(request, pollInterval = 1.second) } .via(killSwitch.flow) // SharedKillSwitch for graceful shutdown .mapAsync(parallelism)(processMessage) .to(GooglePubSub.acknowledge(parallelism = 1)) .run() ``` Note: `UniqueKillSwitch` via `KillSwitches.single` cannot be used with `RestartSource` because `RestartSource` materializes to `NotUsed`. Use `SharedKillSwitch` via `KillSwitches.shared(name)` instead. ## Suggested Fix One or more of the following: 1. **Document the reconnection requirement** — Add a prominent note in the `GooglePubSub.subscribe` scaladoc and user guide that `RestartSource` is required for production use 2. **Built-in reconnection** — Implement reconnection logic inside `GooglePubSub.subscribe`, similar to the official Google client libraries 3. **Provide a convenience method** — e.g. `GooglePubSub.subscribeWithRestart(request, restartSettings)` that wraps the source with `RestartSource` out of the box ## Stack Trace ``` io.grpc.StatusRuntimeException: UNAVAILABLE: The StreamingPull stream closed for an expected reason and should be recreated at io.grpc.Status.asRuntimeException(Status.java:532) at o.a.pekko.grpc.internal.PekkoNettyGrpcClientGraphStage$anon$1.onCallClosed(PekkoNettyGrpcClientGraphStage.scala:178) at o.a.pekko.grpc.internal.PekkoNettyGrpcClientGraphStage$anon$1.$anonfun$callback$1(PekkoNettyGrpcClientGraphStage.scala:81) at o.a.pekko.stream.impl.fusing.GraphInterpreter.runAsyncInput(GraphInterpreter.scala:484) ... ```