hanishi opened a new pull request, #1621:
URL: https://github.com/apache/pekko-connectors/pull/1621
## Summary
Fixes three correctness gaps in the `google-cloud-pub-sub-grpc` subscriber
pipeline (two introduced by #1494, one latent since 2019 and surfaced by it),
and adds a high-level `Subscriber` resource so users get a correct
subscriber by construction.
Closes #1620
## What this changes
- **`EagerPullTrackingStage`** replaces the demand-driven `.map` tracker
inside `autoExtendAckDeadlines`. Tracking now fires on receipt from the gRPC
adapter, so messages buffered behind a slow `mapAsync` no longer hit their
ack deadline before being tracked. (Bug 1)
- **`StreamingPullRequest.defaultInstance`** for subsequent keepalive
requests in `subscribe(...)`. The Pub/Sub server forbids `clientId`,
`maxOutstandingMessages`, and `maxOutstandingBytes` on subsequent requests; the
previous code only cleared `subscription` and `streamAckDeadlineSeconds`,
so any caller setting `maxOutstandingMessages` got `INVALID_ARGUMENT` ~1s after
stream start. (Bug 2, latent since `3f1a4f7f28` from Jan 2019)
- **`FlowControlGateStage` rewritten** with the same eager-pull pattern as
bug 1. Permits acquired on receipt rather than on push downstream, so
`outstandingCount` reflects actual in-flight delivery and the gate's limit
finally bounds what the server sends. Internal `ArrayDeque` pre-sized to
`maxOutstandingMessages` (clamped to 65536). (Bug 3)
- **`AckDeadlineExtender`** is a new caller-owned class holding the
tracking map and background ticker. Both survive `RestartSource.withBackoff`
re-materializations, matching `MessageDispatcher`'s lifecycle in Google's
official client.
- **`GooglePubSub.subscriber(...)`** is the new high-level entry point.
Returns a `Subscriber` resource that bundles `subscribe` + restart +
`autoExtendAckDeadlines(extender)` + optional `flowControlGate`. Composition is
correct by construction: restart wraps only the inner subscribe, the
extender persists across reconnects, `acknowledge`/`nack` sinks release
flow-control permits and record `AckDeadlineDistribution` completion latencies
automatically when configured.
- **Java DSL** inherits all three bug fixes through the existing
factories. A Java DSL `Subscriber` wrapper is deferred (Java/Scala generated
client split needs its own design pass; Java users can still compose the
operators
directly or use `AckDeadlineExtender.create(...)`).
- **`k8s/build-and-push.sh`** hardened with an `EXIT` trap so script
failures no longer leak `GkeAuthTest.scala` / `GkeFullFeatureTest.scala` into
the source tree.
- **Scenario 8 added to `GkeFullFeatureTest`** exercising the new
`Subscriber` resource end-to-end against real Pub/Sub via Workload Identity.
Asset committed but not run this PR cycle.
## Verification
- [x] 14 unit tests (`AutoExtendAckDeadlinesSpec` + `SubscriberSpec`) pass
against stub `SubscriberClient`s. No live infrastructure needed.
- [x] 18/18 `IntegrationSpec` tests pass against the Pub/Sub emulator
(`docker compose up gcloud-pubsub-emulator`), including the new `Subscriber
resource` scenario. No regressions in the 17 pre-existing tests.
- [x] The new `Subscriber resource` scenario also passed against real GCP
(project `pekko-connectors`) via `gcloud auth application-default login`. 35s;
recorded in `google-cloud-pub-sub-grpc/README.md` test report table.
## Notes for reviewers
- `Subscriber` is intended as the primary entry point going forward. The
low-level operators (`subscribe`, `autoExtendAckDeadlines`, `flowControlGate`,
etc.) stay available for callers who genuinely need a different composition
(sharing one extender across multiple subscriptions, splitting the source
for fan-out, etc.).
- `FlowControl.outstandingCount` semantics changed: permits are now
acquired on receipt, not on push downstream. Telemetry built on this counter
will report higher numbers for the same workload, but those numbers are now
accurate. Doc updated.
- `AckDeadline`, `AckDeadlineExtender`, and `Subscriber` are
`@ApiMayChange`.
- The internal-ticker `autoExtendAckDeadlines(subscription, ...)`
overloads stay intact for back-compat, with a doc note that they're not
restart-safe and pointing at the `AckDeadlineExtender` form for that case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
