google-cloud-pub-sub-grpc: subscribe() leaks initial-only fields into subsequent StreamingPullRequests, server returns INVALID_ARGUMENT

[hanishi]

Summary

GooglePubSub.subscribe(...) constructs the keepalive request by clearing only subscription and streamAckDeadlineSeconds, leaving clientId, maxOutstandingMessages, and maxOutstandingBytes from the initial StreamingPullRequest. The Pub/Sub server forbids those three on subsequent requests and returns INVALID_ARGUMENT. Any caller who sets maxOutstandingMessages to bound server delivery hits this ~1s after the stream starts.

The leak has been here since commit 3f1a4f7f28 (Jan 2019). It only matters once you have a use case for setting maxOutstandingMessages on a long-lived stream, which isn't very common in the current alpakka surface (no autoExtendAckDeadlines, no RestartSettings overload of subscribe).

Buggy code

google-cloud-pub-sub-grpc/src/main/scala/akka/stream/alpakka/googlecloud/pubsub/grpc/scaladsl/GooglePubSub.scala:52-54:

val subsequentRequest = request
  .withSubscription("")
  .withStreamAckDeadlineSeconds(0)

Server response on the second polling tick:

INVALID_ARGUMENT: You must only set max_outstanding_messages and
max_outstanding_bytes on the initial request.

One-line fix

Build the subsequent request from the proto's default instance:

val subsequentRequest = StreamingPullRequest.defaultInstance

Clears every initial-only field at once and is forward-safe against future proto revisions.

Cross-reference

This is the same bug we just fixed in pekko-connectors: apache/pekko-connectors#1620. That issue also covers two pekko-only regressions (autoExtendAckDeadlines tracking lag and flowControlGate permit accounting) that don't apply to alpakka because the operators don't exist here, plus a higher-level Subscriber resource that bundles the corrected operators. The architectural insight section there may still be useful if anyone plans to add similar functionality to alpakka.

Happy to send a PR if there's interest.