You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by fa...@apache.org on 2026/03/24 22:01:42 UTC
(pekko-connectors) branch main updated: Google Common, Google Cloud Pub/Sub gRPC: Add google-application-default credentials and StreamingPull auto-reconnect (#1494)
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new 83c152b11 Google Common, Google Cloud Pub/Sub gRPC: Add google-application-default credentials and StreamingPull auto-reconnect (#1494)
83c152b11 is described below
commit 83c152b115dac6cdc6888cd03f15c709762dcfdd
Author: Haruhiko Nishi <ha...@users.noreply.github.com>
AuthorDate: Wed Mar 25 07:01:37 2026 +0900
Google Common, Google Cloud Pub/Sub gRPC: Add google-application-default credentials and StreamingPull auto-reconnect (#1494)
* Add google-application-default credentials and StreamingPull feature parity
* Add full-featured GKE integration test and fix build script packaging
* scalafmt
---------
Co-authored-by: PJ Fanning <pj...@users.noreply.github.com>
---
docs/src/main/paradox/google-cloud-pub-sub-grpc.md | 40 +-
google-cloud-pub-sub-grpc/README.md | 87 +++
google-cloud-pub-sub-grpc/k8s/Dockerfile | 25 +
google-cloud-pub-sub-grpc/k8s/GkeAuthTest.scala | 87 +++
.../k8s/GkeFullFeatureTest.scala | 408 ++++++++++++++
google-cloud-pub-sub-grpc/k8s/application.conf | 11 +
google-cloud-pub-sub-grpc/k8s/build-and-push.sh | 103 ++++
.../k8s/gke-auth-test-job.yaml | 44 ++
.../k8s/gke-full-feature-test-job.yaml | 47 ++
.../pubsub/grpc/AckDeadlineDistribution.scala | 238 +++++++++
.../grpc/AckDeadlineExtensionException.scala | 33 ++
.../googlecloud/pubsub/grpc/FlowControl.scala | 78 +++
.../googlecloud/pubsub/grpc/PubSubSettings.scala | 11 +-
.../pubsub/grpc/impl/DeprecatedCredentials.scala | 18 +-
.../pubsub/grpc/impl/FlowControlGateStage.scala | 80 +++
.../pubsub/grpc/impl/PekkoGrpcSettings.scala | 6 +-
.../pubsub/grpc/javadsl/GooglePubSub.scala | 586 ++++++++++++++++++++-
.../pubsub/grpc/javadsl/GrpcSubscriber.scala | 24 +-
.../pubsub/grpc/scaladsl/GooglePubSub.scala | 555 ++++++++++++++++++-
.../pubsub/grpc/scaladsl/GrpcSubscriber.scala | 22 +-
.../test/java/docs/javadsl/IntegrationTest.java | 257 +++++++++
.../pubsub/grpc/AutoExtendAckDeadlinesTest.java | 294 +++++++++++
.../src/test/scala/docs/scaladsl/ExampleApp.scala | 193 ++++++-
.../test/scala/docs/scaladsl/IntegrationSpec.scala | 382 ++++++++++++++
.../pubsub/grpc/AutoExtendAckDeadlinesSpec.scala | 191 +++++++
google-common/src/main/resources/reference.conf | 12 +-
.../google/auth/AccessTokenCredentials.scala | 2 +-
.../connectors/google/auth/Credentials.scala | 17 +-
.../auth/GoogleApplicationDefaultCredentials.scala | 55 ++
.../google/auth/ProvidedCredentials.scala | 62 +++
.../connectors/google/auth/CredentialsSpec.scala | 159 ++++++
.../google/auth/ProvidedCredentialsSpec.scala | 57 ++
project/Dependencies.scala | 1 +
33 files changed, 4122 insertions(+), 63 deletions(-)
diff --git a/docs/src/main/paradox/google-cloud-pub-sub-grpc.md b/docs/src/main/paradox/google-cloud-pub-sub-grpc.md
index 785ca7614..971f57381 100644
--- a/docs/src/main/paradox/google-cloud-pub-sub-grpc.md
+++ b/docs/src/main/paradox/google-cloud-pub-sub-grpc.md
@@ -148,6 +148,41 @@ Scala
Java
: @@snip (/google-cloud-pub-sub-grpc/src/test/java/docs/javadsl/IntegrationTest.java) { #acknowledge }
+### Extracting ack deadline
+
+@@@ note { .new-in-version }
+
+@ApiMayChange The operators in this section are marked with `@ApiMayChange` and may change in future releases.
+
+@@@
+
+When processing messages takes longer than the acknowledgement deadline, Pub/Sub will redeliver the message. To
+prevent this, you can extend the acknowledgement deadline using `modifyAckDeadlineFlow` or automatically with
+`autoExtendAckDeadlines`.
+
+#### Manual deadline extension
+
+Use `modifyAckDeadlineFlow` to explicitly extend the ack deadline of individual messages during processing:
+
+Scala
+: @@snip (/google-cloud-pub-sub-grpc/src/test/scala/docs/scaladsl/IntegrationSpec.scala) { #modify-ack-deadline }
+
+Java
+: @@snip (/google-cloud-pub-sub-grpc/src/test/java/docs/javadsl/IntegrationTest.java) { #modify-ack-deadline }
+
+#### Automatic deadline extension
+
+For a higher-level approach, `autoExtendAckDeadlines` creates a flow that automatically extends acknowledgement
+deadlines for all messages passing through it. A background timer periodically sends `ModifyAckDeadline` RPCs,
+preventing Pub/Sub from redelivering messages while they are being processed downstream. This is similar to the
+lease management behavior of Google's official client library.
+
+Scala
+: @@snip (/google-cloud-pub-sub-grpc/src/test/scala/docs/scaladsl/IntegrationSpec.scala) { #subscribe-auto-extend }
+
+Java
+: @@snip (/google-cloud-pub-sub-grpc/src/test/java/docs/javadsl/IntegrationTest.java) { #subscribe-auto-extend }
+
## Running the test code
@@@ note
@@ -177,9 +212,12 @@ sbt
```bash
env GOOGLE_APPLICATION_CREDENTIALS=/path/to/application/credentials.json sbt
- // receive messages from a subsciptions
+ // receive messages from a subscription
> google-cloud-pub-sub-grpc/Test/run subscribe <project-id> <subscription-name>
+ // receive messages with automatic ack deadline extension
+ > google-cloud-pub-sub-grpc/Test/run subscribe-auto-extend <project-id> <subscription-name>
+
// publish a single message to a topic
> google-cloud-pub-sub-grpc/Test/run publish-single <project-id> <topic-name>
diff --git a/google-cloud-pub-sub-grpc/README.md b/google-cloud-pub-sub-grpc/README.md
new file mode 100644
index 000000000..c3cdf4c15
--- /dev/null
+++ b/google-cloud-pub-sub-grpc/README.md
@@ -0,0 +1,87 @@
+# Google Cloud Pub/Sub gRPC Connector - GCP Integration Testing
+
+## Overview
+
+The integration tests in this module run against the Pub/Sub emulator by default.
+The emulator does not require authentication, so it does not exercise the
+`google-application-default` credential provider or any authenticated code paths.
+
+To verify that authentication works end-to-end, you must run the tests against a
+real GCP project.
+
+## Prerequisites
+
+- [Google Cloud SDK](https://cloud.google.com/sdk/docs/install) (`gcloud` CLI)
+- A GCP project with billing enabled
+- Pub/Sub API enabled on that project
+
+## GCP Setup
+
+```bash
+# Set your project
+gcloud config set project <YOUR_PROJECT_ID>
+
+# Enable the Pub/Sub API
+gcloud services enable pubsub.googleapis.com
+
+# Create topics and subscriptions used by the integration tests
+gcloud pubsub topics create simpleTopic
+gcloud pubsub subscriptions create simpleSubscription --topic=simpleTopic
+
+gcloud pubsub topics create testTopic
+gcloud pubsub subscriptions create testSubscription --topic=testTopic
+
+# Authenticate for application-default credentials
+gcloud auth application-default login
+```
+
+## Application Config
+
+Create or update `src/test/resources/application.conf` to point to real GCP
+instead of the emulator. Comment out the emulator overrides and configure the
+credential provider:
+
+```hocon
+pekko.connectors.google.cloud.pubsub.grpc {
+ # Comment out or remove these emulator settings:
+ # host = "localhost"
+ # port = 8538
+ # use-tls = false
+ # callCredentials = "none"
+}
+
+pekko.connectors.google {
+ credentials {
+ provider = google-application-default
+ google-application-default {
+ project-id = "<YOUR_PROJECT_ID>"
+ }
+ }
+}
+```
+
+You also need to update the `projectId` in `IntegrationSpec.scala` (and the Java
+`IntegrationTest`) to match your GCP project ID, or extract it into config.
+
+## Running the Tests
+
+```bash
+sbt "google-cloud-pub-sub-grpc/test"
+```
+
+## Cleanup
+
+```bash
+gcloud pubsub subscriptions delete simpleSubscription
+gcloud pubsub subscriptions delete testSubscription
+gcloud pubsub topics delete simpleTopic
+gcloud pubsub topics delete testTopic
+```
+
+## Test Report
+
+| Date | Environment | Credential Provider | Result | Notes |
+|------|-------------|---------------------|--------|-------|
+| 2026-03-13 | GCP (project: pekko-connectors-test) | google-application-default | 14/14 passed | Scala 8/8, Java 6/6. User credentials via `gcloud auth application-default login`. |
+
+After running against real GCP, add a row to the table above to record the result.
\ No newline at end of file
diff --git a/google-cloud-pub-sub-grpc/k8s/Dockerfile b/google-cloud-pub-sub-grpc/k8s/Dockerfile
new file mode 100644
index 000000000..bc4bfef6f
--- /dev/null
+++ b/google-cloud-pub-sub-grpc/k8s/Dockerfile
@@ -0,0 +1,25 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+FROM eclipse-temurin:21-jre-alpine
+
+WORKDIR /app
+COPY lib/ lib/
+COPY gke-auth-test.jar .
+
+ENV MAIN_CLASS=org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.gke.GkeAuthTest
+ENTRYPOINT ["sh", "-c", "java -cp gke-auth-test.jar:lib/* $MAIN_CLASS"]
diff --git a/google-cloud-pub-sub-grpc/k8s/GkeAuthTest.scala b/google-cloud-pub-sub-grpc/k8s/GkeAuthTest.scala
new file mode 100644
index 000000000..db90391ed
--- /dev/null
+++ b/google-cloud-pub-sub-grpc/k8s/GkeAuthTest.scala
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.gke
+
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.stream.connectors.googlecloud.pubsub.grpc.scaladsl.GooglePubSub
+import pekko.stream.scaladsl.{ Flow, Sink, Source }
+import com.google.protobuf.ByteString
+import com.google.pubsub.v1.pubsub._
+
+import scala.concurrent.Await
+import scala.concurrent.duration._
+
+/**
+ * Minimal app to verify Pub/Sub auth on GKE.
+ * Publishes a message then subscribes to confirm it arrived.
+ * Exits 0 on success, 1 on failure.
+ */
+object GkeAuthTest {
+ def main(args: Array[String]): Unit = {
+ val projectId = sys.env.getOrElse("PROJECT_ID", "pekko-connectors")
+ val topic = sys.env.getOrElse("TOPIC", "simpleTopic")
+ val subscription = sys.env.getOrElse("SUBSCRIPTION", "simpleSubscription")
+
+ implicit val system: ActorSystem = ActorSystem("GkeAuthTest")
+ import system.dispatcher
+
+ try {
+ val topicFqrs = s"projects/$projectId/topics/$topic"
+ val subFqrs = s"projects/$projectId/subscriptions/$subscription"
+ val testData = ByteString.copyFromUtf8(s"gke-auth-test-${System.nanoTime()}")
+
+ // Publish
+ println(s"Publishing to $topicFqrs ...")
+ val publishResult = Source
+ .single(PublishRequest(topicFqrs, Seq(PubsubMessage().withData(testData))))
+ .via(GooglePubSub.publish(parallelism = 1))
+ .runWith(Sink.head)
+
+ val response = Await.result(publishResult, 30.seconds)
+ println(s"Published: messageId=${response.messageIds.headOption.getOrElse("?")}")
+
+ // Subscribe
+ println(s"Subscribing to $subFqrs ...")
+ val request = StreamingPullRequest(subFqrs, streamAckDeadlineSeconds = 10)
+ val received = Source.futureSource(
+ GooglePubSub.subscribe(request, 1.second)
+ .filter(_.message.exists(_.data == testData))
+ .take(1)
+ .alsoTo(
+ Flow[ReceivedMessage]
+ .map(m => AcknowledgeRequest(subFqrs, Seq(m.ackId)))
+ .to(GooglePubSub.acknowledge(parallelism = 1)))
+ .runWith(Sink.head)
+ .map(Source.single)(dispatcher))
+ .runWith(Sink.head)
+
+ val msg = Await.result(received, 30.seconds)
+ println(s"Received: ${msg.message.map(_.data.toStringUtf8).getOrElse("?")}")
+ println("SUCCESS: GKE auth test passed")
+ Await.result(system.terminate(), 10.seconds)
+ sys.exit(0)
+ } catch {
+ case ex: Throwable =>
+ println(s"FAILED: ${ex.getMessage}")
+ ex.printStackTrace()
+ Await.result(system.terminate(), 10.seconds)
+ sys.exit(1)
+ }
+ }
+}
diff --git a/google-cloud-pub-sub-grpc/k8s/GkeFullFeatureTest.scala b/google-cloud-pub-sub-grpc/k8s/GkeFullFeatureTest.scala
new file mode 100644
index 000000000..3da92fcb0
--- /dev/null
+++ b/google-cloud-pub-sub-grpc/k8s/GkeFullFeatureTest.scala
@@ -0,0 +1,408 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.gke
+
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.stream.RestartSettings
+import pekko.stream.connectors.googlecloud.pubsub.grpc.{ AckDeadlineDistribution, FlowControl }
+import pekko.stream.connectors.googlecloud.pubsub.grpc.scaladsl.GooglePubSub
+import pekko.stream.scaladsl.{ Flow, Sink, Source }
+import com.google.protobuf.ByteString
+import com.google.pubsub.v1.pubsub._
+
+import scala.concurrent.duration._
+import scala.concurrent.{ Await, Future }
+
+/**
+ * Full-featured GKE example app exercising all features:
+ *
+ * 1. google-application-default credentials (via application.conf)
+ * 2. StreamingPull with auto-reconnect (RestartSource)
+ * 3. Auto-extend ack deadlines (fixed deadline)
+ * 4. Auto-extend ack deadlines (adaptive via AckDeadlineDistribution)
+ * 5. Flow control (backpressure on outstanding messages)
+ * 6. Nack (immediate redelivery)
+ * 7. Dynamic ack deadline modification
+ *
+ * Each scenario publishes messages, then subscribes and processes them,
+ * printing results to stdout. Exits 0 on success, 1 on failure.
+ */
+object GkeFullFeatureTest {
+
+ def main(args: Array[String]): Unit = {
+ val projectId = sys.env.getOrElse("PROJECT_ID", "pekko-connectors")
+ val topic = sys.env.getOrElse("TOPIC", "simpleTopic")
+ val subscription = sys.env.getOrElse("SUBSCRIPTION", "simpleSubscription")
+
+ implicit val system: ActorSystem = ActorSystem("GkeFullFeatureTest")
+ val topicFqrs = s"projects/$projectId/topics/$topic"
+ val subFqrs = s"projects/$projectId/subscriptions/$subscription"
+
+ try {
+ scenario1_BasicPublishSubscribe(topicFqrs, subFqrs)
+ scenario2_ReconnectingSubscriber(topicFqrs, subFqrs)
+ scenario3_FixedAutoExtendAckDeadlines(topicFqrs, subFqrs)
+ scenario4_AdaptiveAckDeadlines(topicFqrs, subFqrs)
+ scenario5_FlowControl(topicFqrs, subFqrs)
+ scenario6_NackAndRedeliver(topicFqrs, subFqrs)
+ scenario7_DynamicDeadlineModification(topicFqrs, subFqrs)
+
+ println("\n=== ALL SCENARIOS PASSED ===")
+ Await.result(system.terminate(), 10.seconds)
+ sys.exit(0)
+ } catch {
+ case ex: Throwable =>
+ println(s"\nFAILED: ${ex.getMessage}")
+ ex.printStackTrace()
+ Await.result(system.terminate(), 10.seconds)
+ sys.exit(1)
+ }
+ }
+
+ // ---------------------------------------------------------------------------
+ // Scenario 1: Basic publish + subscribe (google-application-default auth)
+ // ---------------------------------------------------------------------------
+ private def scenario1_BasicPublishSubscribe(topicFqrs: String, subFqrs: String)(
+ implicit system: ActorSystem): Unit = {
+ println("\n--- Scenario 1: Basic publish + subscribe (google-application-default credentials) ---")
+
+ val testData = ByteString.copyFromUtf8(s"scenario1-${System.nanoTime()}")
+
+ // Publish
+ val publishResult = Source
+ .single(PublishRequest(topicFqrs, Seq(PubsubMessage().withData(testData))))
+ .via(GooglePubSub.publish(parallelism = 1))
+ .runWith(Sink.head)
+
+ val response = Await.result(publishResult, 30.seconds)
+ println(s" Published: messageId=${response.messageIds.headOption.getOrElse("?")}")
+
+ // Subscribe and receive
+ val request = StreamingPullRequest(subFqrs, streamAckDeadlineSeconds = 10)
+ val received = GooglePubSub
+ .subscribe(request, 1.second)
+ .filter(_.message.exists(_.data == testData))
+ .take(1)
+ .alsoTo(
+ Flow[ReceivedMessage]
+ .map(m => AcknowledgeRequest(subFqrs, Seq(m.ackId)))
+ .to(GooglePubSub.acknowledge(parallelism = 1)))
+ .runWith(Sink.head)
+
+ val msg = Await.result(received, 30.seconds)
+ println(s" Received: ${msg.message.map(_.data.toStringUtf8).getOrElse("?")}")
+ assert(msg.message.exists(_.data == testData), "Data mismatch!")
+ println(" PASSED")
+ }
+
+ // ---------------------------------------------------------------------------
+ // Scenario 2: Reconnecting subscriber with RestartSettings
+ // ---------------------------------------------------------------------------
+ private def scenario2_ReconnectingSubscriber(topicFqrs: String, subFqrs: String)(
+ implicit system: ActorSystem): Unit = {
+ println("\n--- Scenario 2: Reconnecting subscriber (RestartSource) ---")
+
+ val testData = ByteString.copyFromUtf8(s"scenario2-${System.nanoTime()}")
+
+ // Publish
+ Await.result(
+ Source
+ .single(PublishRequest(topicFqrs, Seq(PubsubMessage().withData(testData))))
+ .via(GooglePubSub.publish(parallelism = 1))
+ .runWith(Sink.head),
+ 30.seconds)
+ println(" Published message")
+
+ // Subscribe with auto-reconnect
+ val restartSettings = RestartSettings(
+ minBackoff = 1.second,
+ maxBackoff = 10.seconds,
+ randomFactor = 0.2).withMaxRestarts(3, 1.minute)
+
+ val request = StreamingPullRequest(subFqrs, streamAckDeadlineSeconds = 10)
+ val received = GooglePubSub
+ .subscribe(request, 1.second, restartSettings)
+ .filter(_.message.exists(_.data == testData))
+ .take(1)
+ .alsoTo(
+ Flow[ReceivedMessage]
+ .map(m => AcknowledgeRequest(subFqrs, Seq(m.ackId)))
+ .to(GooglePubSub.acknowledge(parallelism = 1)))
+ .runWith(Sink.head)
+
+ val msg = Await.result(received, 30.seconds)
+ println(s" Received via reconnecting subscriber: ${msg.message.map(_.data.toStringUtf8).getOrElse("?")}")
+ println(" PASSED")
+ }
+
+ // ---------------------------------------------------------------------------
+ // Scenario 3: Fixed auto-extend ack deadlines
+ // ---------------------------------------------------------------------------
+ private def scenario3_FixedAutoExtendAckDeadlines(topicFqrs: String, subFqrs: String)(
+ implicit system: ActorSystem): Unit = {
+ println("\n--- Scenario 3: Fixed auto-extend ack deadlines ---")
+
+ val testData = ByteString.copyFromUtf8(s"scenario3-${System.nanoTime()}")
+
+ // Publish
+ Await.result(
+ Source
+ .single(PublishRequest(topicFqrs, Seq(PubsubMessage().withData(testData))))
+ .via(GooglePubSub.publish(parallelism = 1))
+ .runWith(Sink.head),
+ 30.seconds)
+ println(" Published message")
+
+ // Subscribe with auto-extend: extend every 8s with a 30s deadline
+ val request = StreamingPullRequest(subFqrs, streamAckDeadlineSeconds = 10)
+ val received = GooglePubSub
+ .subscribe(request, 1.second)
+ .mapMaterializedValue(_ => pekko.NotUsed)
+ .via(GooglePubSub.autoExtendAckDeadlines(subFqrs, 8.seconds, 30))
+ .filter(_.message.exists(_.data == testData))
+ .take(1)
+ .mapAsync(1) { msg =>
+ // Simulate slow processing (15s) — without auto-extend this would cause redelivery
+ println(s" Processing message (simulating 15s delay)...")
+ pekko.pattern.after(15.seconds)(Future.successful(msg))
+ }
+ .alsoTo(
+ Flow[ReceivedMessage]
+ .map(m => AcknowledgeRequest(subFqrs, Seq(m.ackId)))
+ .to(GooglePubSub.acknowledge(parallelism = 1)))
+ .runWith(Sink.head)
+
+ val msg = Await.result(received, 60.seconds)
+ println(s" Received after slow processing: ${msg.message.map(_.data.toStringUtf8).getOrElse("?")}")
+ println(" PASSED (no redelivery during slow processing)")
+ }
+
+ // ---------------------------------------------------------------------------
+ // Scenario 4: Adaptive ack deadlines with AckDeadlineDistribution
+ // ---------------------------------------------------------------------------
+ private def scenario4_AdaptiveAckDeadlines(topicFqrs: String, subFqrs: String)(
+ implicit system: ActorSystem): Unit = {
+ println("\n--- Scenario 4: Adaptive ack deadlines (AckDeadlineDistribution) ---")
+
+ val messageCount = 5
+ val testPrefix = s"scenario4-${System.nanoTime()}"
+ val messages = (1 to messageCount).map(i =>
+ PubsubMessage().withData(ByteString.copyFromUtf8(s"$testPrefix-$i")))
+
+ // Publish batch
+ Await.result(
+ Source
+ .single(PublishRequest(topicFqrs, messages))
+ .via(GooglePubSub.publish(parallelism = 1))
+ .runWith(Sink.head),
+ 30.seconds)
+ println(s" Published $messageCount messages")
+
+ // Set up adaptive distribution
+ val distribution = AckDeadlineDistribution(
+ initialDeadlineSeconds = 10,
+ minDeadlineSeconds = 10,
+ maxDeadlineSeconds = 600)
+
+ val request = StreamingPullRequest(subFqrs, streamAckDeadlineSeconds = 10)
+ val received = GooglePubSub
+ .subscribe(request, 1.second)
+ .mapMaterializedValue(_ => pekko.NotUsed)
+ .via(GooglePubSub.autoExtendAckDeadlines(subFqrs, 3.seconds, distribution))
+ .filter(_.message.exists(_.data.toStringUtf8.startsWith(testPrefix)))
+ .take(messageCount)
+ .mapAsync(1) { msg =>
+ // Simulate variable processing time
+ val delay = (msg.message.map(_.data.toStringUtf8).getOrElse("").hashCode.abs % 3 + 1).seconds
+ println(s" Processing: ${msg.message.map(_.data.toStringUtf8).getOrElse("?")} (${delay.toSeconds}s)")
+ pekko.pattern.after(delay)(Future.successful(msg))
+ }
+ .alsoTo(
+ Flow[ReceivedMessage]
+ .map(m => AcknowledgeRequest(subFqrs, Seq(m.ackId)))
+ .to(GooglePubSub.acknowledge(parallelism = 1, distribution)))
+ .runWith(Sink.seq)
+
+ val msgs = Await.result(received, 120.seconds)
+ println(s" Received ${msgs.size}/$messageCount messages")
+ println(s" Adaptive deadline: ${distribution.currentDeadlineSeconds}s")
+ assert(msgs.size == messageCount, s"Expected $messageCount messages, got ${msgs.size}")
+ println(" PASSED")
+ }
+
+ // ---------------------------------------------------------------------------
+ // Scenario 5: Flow control (backpressure on outstanding messages)
+ // ---------------------------------------------------------------------------
+ private def scenario5_FlowControl(topicFqrs: String, subFqrs: String)(
+ implicit system: ActorSystem): Unit = {
+ println("\n--- Scenario 5: Flow control (max outstanding messages) ---")
+
+ val messageCount = 10
+ val maxOutstanding = 3
+ val testPrefix = s"scenario5-${System.nanoTime()}"
+ val messages = (1 to messageCount).map(i =>
+ PubsubMessage().withData(ByteString.copyFromUtf8(s"$testPrefix-$i")))
+
+ // Publish batch
+ Await.result(
+ Source
+ .single(PublishRequest(topicFqrs, messages))
+ .via(GooglePubSub.publish(parallelism = 1))
+ .runWith(Sink.head),
+ 30.seconds)
+ println(s" Published $messageCount messages")
+
+ val flowControl = FlowControl(maxOutstandingMessages = maxOutstanding)
+ var maxObserved = 0L
+
+ val request = StreamingPullRequest(subFqrs, streamAckDeadlineSeconds = 10)
+ val received = GooglePubSub
+ .subscribe(request, 1.second)
+ .mapMaterializedValue(_ => pekko.NotUsed)
+ .via(GooglePubSub.flowControlGate(flowControl))
+ .filter(_.message.exists(_.data.toStringUtf8.startsWith(testPrefix)))
+ .take(messageCount)
+ .mapAsync(maxOutstanding) { msg =>
+ val current = flowControl.outstandingCount
+ synchronized { if (current > maxObserved) maxObserved = current }
+ println(s" Processing: ${msg.message.map(_.data.toStringUtf8).getOrElse("?")} " +
+ s"(outstanding: $current/$maxOutstanding)")
+ // Simulate work
+ pekko.pattern.after(1.second)(Future.successful(msg))
+ }
+ .alsoTo(
+ Flow[ReceivedMessage]
+ .map(m => AcknowledgeRequest(subFqrs, Seq(m.ackId)))
+ .to(GooglePubSub.acknowledge(parallelism = 1, flowControl)))
+ .runWith(Sink.seq)
+
+ val msgs = Await.result(received, 120.seconds)
+ println(s" Received ${msgs.size}/$messageCount messages")
+ println(s" Max outstanding observed: $maxObserved (limit: $maxOutstanding)")
+ assert(msgs.size == messageCount, s"Expected $messageCount messages, got ${msgs.size}")
+ assert(maxObserved <= maxOutstanding, s"Flow control violated: $maxObserved > $maxOutstanding")
+ println(" PASSED")
+ }
+
+ // ---------------------------------------------------------------------------
+ // Scenario 6: Nack causes immediate redelivery
+ // ---------------------------------------------------------------------------
+ private def scenario6_NackAndRedeliver(topicFqrs: String, subFqrs: String)(
+ implicit system: ActorSystem): Unit = {
+ println("\n--- Scenario 6: Nack and immediate redelivery ---")
+
+ val testData = ByteString.copyFromUtf8(s"scenario6-${System.nanoTime()}")
+
+ // Publish
+ Await.result(
+ Source
+ .single(PublishRequest(topicFqrs, Seq(PubsubMessage().withData(testData))))
+ .via(GooglePubSub.publish(parallelism = 1))
+ .runWith(Sink.head),
+ 30.seconds)
+ println(" Published message")
+
+ // Subscribe: nack the first delivery, ack the second
+ val request = StreamingPullRequest(subFqrs, streamAckDeadlineSeconds = 10)
+ val received = GooglePubSub
+ .subscribe(request, 1.second)
+ .mapMaterializedValue(_ => pekko.NotUsed)
+ .filter(_.message.exists(_.data == testData))
+ .statefulMap(() => 0)(
+ (count, msg) => {
+ val newCount = count + 1
+ (newCount, (newCount, msg))
+ },
+ _ => None)
+ .mapAsync(1) {
+ case (attempt, msg) if attempt == 1 =>
+ println(s" Attempt $attempt: nacking message for redelivery")
+ val nackReq = AcknowledgeRequest(subFqrs, Seq(msg.ackId))
+ Source
+ .single(nackReq)
+ .via(GooglePubSub.nackFlow())
+ .runWith(Sink.head)
+ .map(_ => None)(system.dispatcher)
+ case (attempt, msg) =>
+ println(s" Attempt $attempt: acknowledging message")
+ val ackReq = AcknowledgeRequest(subFqrs, Seq(msg.ackId))
+ Source
+ .single(ackReq)
+ .via(GooglePubSub.acknowledgeFlow())
+ .runWith(Sink.head)
+ .map(_ => Some(msg))(system.dispatcher)
+ }
+ .collect { case Some(m) => m }
+ .take(1)
+ .runWith(Sink.head)
+
+ val msg = Await.result(received, 60.seconds)
+ println(s" Received after nack+redeliver: ${msg.message.map(_.data.toStringUtf8).getOrElse("?")}")
+ println(" PASSED (message was nacked and redelivered)")
+ }
+
+ // ---------------------------------------------------------------------------
+ // Scenario 7: Dynamic per-message deadline modification
+ // ---------------------------------------------------------------------------
+ private def scenario7_DynamicDeadlineModification(topicFqrs: String, subFqrs: String)(
+ implicit system: ActorSystem): Unit = {
+ println("\n--- Scenario 7: Dynamic per-message deadline modification ---")
+
+ val messageCount = 3
+ val testPrefix = s"scenario7-${System.nanoTime()}"
+ val messages = (1 to messageCount).map(i =>
+ PubsubMessage()
+ .withData(ByteString.copyFromUtf8(s"$testPrefix-$i"))
+ .withAttributes(Map("priority" -> (if (i == 1) "high" else "low"))))
+
+ // Publish
+ Await.result(
+ Source
+ .single(PublishRequest(topicFqrs, messages))
+ .via(GooglePubSub.publish(parallelism = 1))
+ .runWith(Sink.head),
+ 30.seconds)
+ println(s" Published $messageCount messages with priority attributes")
+
+ // Subscribe with dynamic deadline: high priority gets longer deadline
+ val request = StreamingPullRequest(subFqrs, streamAckDeadlineSeconds = 10)
+ val received = GooglePubSub
+ .subscribe(request, 1.second)
+ .mapMaterializedValue(_ => pekko.NotUsed)
+ .filter(_.message.exists(_.data.toStringUtf8.startsWith(testPrefix)))
+ .take(messageCount)
+ .via(GooglePubSub.modifyAckDeadlineDynamic(subFqrs, parallelism = 1) { msg =>
+ val priority = msg.message.flatMap(_.attributes.get("priority")).getOrElse("low")
+ val deadline = if (priority == "high") 60 else 15
+ println(s" Setting deadline=${deadline}s for priority=$priority: " +
+ s"${msg.message.map(_.data.toStringUtf8).getOrElse("?")}")
+ deadline
+ })
+ .alsoTo(
+ Flow[ReceivedMessage]
+ .map(m => AcknowledgeRequest(subFqrs, Seq(m.ackId)))
+ .to(GooglePubSub.acknowledge(parallelism = 1)))
+ .runWith(Sink.seq)
+
+ val msgs = Await.result(received, 60.seconds)
+ println(s" Received ${msgs.size}/$messageCount messages with dynamic deadlines")
+ assert(msgs.size == messageCount, s"Expected $messageCount messages, got ${msgs.size}")
+ println(" PASSED")
+ }
+}
diff --git a/google-cloud-pub-sub-grpc/k8s/application.conf b/google-cloud-pub-sub-grpc/k8s/application.conf
new file mode 100644
index 000000000..f90401828
--- /dev/null
+++ b/google-cloud-pub-sub-grpc/k8s/application.conf
@@ -0,0 +1,11 @@
+# SPDX-License-Identifier: Apache-2.0
+pekko {
+ loglevel = "INFO"
+}
+
+pekko.connectors.google.credentials {
+ provider = google-application-default
+ google-application-default {
+ project-id = "pekko-connectors"
+ }
+}
diff --git a/google-cloud-pub-sub-grpc/k8s/build-and-push.sh b/google-cloud-pub-sub-grpc/k8s/build-and-push.sh
new file mode 100755
index 000000000..41e08ae5e
--- /dev/null
+++ b/google-cloud-pub-sub-grpc/k8s/build-and-push.sh
@@ -0,0 +1,103 @@
+#!/bin/bash
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+set -euo pipefail
+
+PROJECT_ID="pekko-connectors"
+REGION="us-central1"
+REPO="pekko-test"
+IMAGE="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPO}/gke-auth-test:latest"
+SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
+ROOT_DIR="$(cd "${SCRIPT_DIR}/../.." && pwd)"
+
+echo "=== Creating Artifact Registry repo (if needed) ==="
+gcloud artifacts repositories create "${REPO}" \
+ --repository-format=docker \
+ --location="${REGION}" \
+ --project="${PROJECT_ID}" 2>/dev/null || true
+
+echo "=== Configuring Docker auth ==="
+gcloud auth configure-docker "${REGION}-docker.pkg.dev" --quiet
+
+echo "=== Building with sbt ==="
+cd "${ROOT_DIR}"
+
+# Compile the GKE test app along with the connector
+# We need to compile it as part of the project to get all dependencies
+mkdir -p google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/gke
+cp "${SCRIPT_DIR}/GkeAuthTest.scala" google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/gke/
+cp "${SCRIPT_DIR}/GkeFullFeatureTest.scala" google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/gke/
+cp "${SCRIPT_DIR}/application.conf" google-cloud-pub-sub-grpc/src/main/resources/gke-application.conf
+
+sbt "google-cloud-pub-sub-grpc/compile"
+
+# Export classpath and build the jar
+echo "=== Packaging ==="
+FULL_CP=$(sbt --error "print google-cloud-pub-sub-grpc/fullClasspath" | tr ',' '\n' | sed 's/.*Attributed(\(.*\))/\1/')
+CLASSES_DIR=$(sbt --error "print google-cloud-pub-sub-grpc/classDirectory" | tr -d '[:space:]')
+
+# Create staging directory
+STAGING="${SCRIPT_DIR}/staging"
+rm -rf "${STAGING}"
+mkdir -p "${STAGING}/lib"
+
+# Copy the compiled classes as a jar
+cd "${CLASSES_DIR}"
+jar cf "${STAGING}/gke-auth-test.jar" .
+cd "${ROOT_DIR}"
+
+# Copy dependency jars and package inter-project class directories as jars
+echo "${FULL_CP}" | tr ':' '\n' | while read -r entry; do
+ entry=$(echo "$entry" | tr -d '[:space:]')
+ [ -z "$entry" ] && continue
+ if [ -f "$entry" ] && echo "$entry" | grep -q '\.jar$'; then
+ # External jar dependency
+ cp "$entry" "${STAGING}/lib/"
+ elif [ -d "$entry" ] && [ "$entry" != "$CLASSES_DIR" ]; then
+ # Inter-project classes directory — package as a jar
+ dir_name=$(echo "$entry" | sed 's|.*/\([^/]*\)/target/.*|\1|')
+ cd "$entry"
+ jar cf "${STAGING}/lib/${dir_name}.jar" .
+ cd "${ROOT_DIR}"
+ fi
+done
+
+# Copy the GKE-specific application.conf into the jar
+cd "${STAGING}"
+mkdir -p tmp
+cd tmp
+jar xf ../gke-auth-test.jar
+cp "${SCRIPT_DIR}/application.conf" .
+jar cf ../gke-auth-test.jar .
+cd "${STAGING}"
+rm -rf tmp
+
+echo "=== Building Docker image ==="
+cp "${SCRIPT_DIR}/Dockerfile" "${STAGING}/"
+docker build -t "${IMAGE}" "${STAGING}"
+
+echo "=== Pushing to Artifact Registry ==="
+docker push "${IMAGE}"
+
+echo "=== Cleaning up ==="
+rm -rf "${STAGING}"
+rm -f "${ROOT_DIR}/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/gke/GkeAuthTest.scala"
+rm -f "${ROOT_DIR}/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/gke/GkeFullFeatureTest.scala"
+rm -f "${ROOT_DIR}/google-cloud-pub-sub-grpc/src/main/resources/gke-application.conf"
+
+echo "=== Done: ${IMAGE} ==="
diff --git a/google-cloud-pub-sub-grpc/k8s/gke-auth-test-job.yaml b/google-cloud-pub-sub-grpc/k8s/gke-auth-test-job.yaml
new file mode 100644
index 000000000..b8c48fd45
--- /dev/null
+++ b/google-cloud-pub-sub-grpc/k8s/gke-auth-test-job.yaml
@@ -0,0 +1,44 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+apiVersion: batch/v1
+kind: Job
+metadata:
+ name: gke-auth-test
+spec:
+ backoffLimit: 0
+ template:
+ spec:
+ serviceAccountName: pekko-test
+ containers:
+ - name: test
+ image: us-central1-docker.pkg.dev/pekko-connectors/pekko-test/gke-auth-test:latest
+ env:
+ - name: PROJECT_ID
+ value: "pekko-connectors"
+ - name: TOPIC
+ value: "simpleTopic"
+ - name: SUBSCRIPTION
+ value: "simpleSubscription"
+ resources:
+ requests:
+ cpu: "250m"
+ memory: "512Mi"
+ limits:
+ cpu: "500m"
+ memory: "1Gi"
+ restartPolicy: Never
diff --git a/google-cloud-pub-sub-grpc/k8s/gke-full-feature-test-job.yaml b/google-cloud-pub-sub-grpc/k8s/gke-full-feature-test-job.yaml
new file mode 100644
index 000000000..c74c1e489
--- /dev/null
+++ b/google-cloud-pub-sub-grpc/k8s/gke-full-feature-test-job.yaml
@@ -0,0 +1,47 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+apiVersion: batch/v1
+kind: Job
+metadata:
+ name: gke-full-feature-test
+spec:
+ backoffLimit: 0
+ activeDeadlineSeconds: 600
+ template:
+ spec:
+ serviceAccountName: pekko-test
+ containers:
+ - name: test
+ image: us-central1-docker.pkg.dev/pekko-connectors/pekko-test/gke-auth-test:latest
+ env:
+ - name: MAIN_CLASS
+ value: "org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.gke.GkeFullFeatureTest"
+ - name: PROJECT_ID
+ value: "pekko-connectors"
+ - name: TOPIC
+ value: "simpleTopic"
+ - name: SUBSCRIPTION
+ value: "simpleSubscription"
+ resources:
+ requests:
+ cpu: "500m"
+ memory: "1Gi"
+ limits:
+ cpu: "1000m"
+ memory: "2Gi"
+ restartPolicy: Never
diff --git a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/AckDeadlineDistribution.scala b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/AckDeadlineDistribution.scala
new file mode 100644
index 000000000..e6d4e222c
--- /dev/null
+++ b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/AckDeadlineDistribution.scala
@@ -0,0 +1,238 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc
+
+import org.apache.pekko.annotation.ApiMayChange
+
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.{ AtomicInteger, AtomicLongArray }
+
+/**
+ * Tracks message processing latencies and computes an adaptive ack deadline
+ * based on the percentile of observed processing times.
+ *
+ * The percentile computation uses the nearest-rank method, matching the
+ * implementation in Google's `com.google.api.gax.core.Distribution` class.
+ * Google's `MessageDispatcher` uses a 99.9th percentile by default; this
+ * class defaults to the same.
+ *
+ * Pass the same instance to `autoExtendAckDeadlines` (which reads the computed
+ * deadline) and to the acknowledge/nack operators (which record completion times).
+ *
+ * Usage:
+ * {{{
+ * val dist = AckDeadlineDistribution(
+ * initialDeadlineSeconds = 10,
+ * minDeadlineSeconds = 10,
+ * maxDeadlineSeconds = 600)
+ *
+ * GooglePubSub.subscribe(request, 1.second)
+ * .via(GooglePubSub.autoExtendAckDeadlines(subscriptionFqrs, 3.seconds, dist))
+ * .mapAsync(4)(processMessage)
+ * .map(msg => AcknowledgeRequest(sub, Seq(msg.ackId)))
+ * .to(GooglePubSub.acknowledge(parallelism = 1, dist))
+ * }}}
+ *
+ * @since 2.0.0
+ */
+@ApiMayChange
+final class AckDeadlineDistribution private (
+ val initialDeadlineSeconds: Int,
+ val minDeadlineSeconds: Int,
+ val maxDeadlineSeconds: Int,
+ val maxAckExtensionPeriodNanos: Long,
+ val percentile: Double) {
+
+ require(minDeadlineSeconds >= AckDeadlineDistribution.MinStreamAckDeadlineSeconds,
+ s"minDeadlineSeconds must be >= ${AckDeadlineDistribution.MinStreamAckDeadlineSeconds} (Pub/Sub minimum)")
+ require(maxDeadlineSeconds <= AckDeadlineDistribution.MaxStreamAckDeadlineSeconds,
+ s"maxDeadlineSeconds must be <= ${AckDeadlineDistribution.MaxStreamAckDeadlineSeconds} (Pub/Sub maximum)")
+ require(minDeadlineSeconds <= maxDeadlineSeconds, "minDeadlineSeconds must be <= maxDeadlineSeconds")
+ require(percentile > 0.0 && percentile <= 100.0, "percentile must be between 0 (exclusive) and 100 (inclusive)")
+
+ // Tracks when each ackId was first seen (entry nanos). Uses java.lang.Long for null-safety with ConcurrentHashMap.
+ private[grpc] val deliveryTimes = new ConcurrentHashMap[String, java.lang.Long]()
+
+ // Histogram using AtomicLongArray — matches Google's com.google.api.gax.core.Distribution.
+ // Buckets cover 0 (inclusive) to endValue (exclusive) where endValue = maxDeadlineSeconds + 1.
+ // Values >= endValue are clamped to endValue - 1 (long-tail handling).
+ private val endValue = maxDeadlineSeconds + 1
+ private val buckets = new AtomicLongArray(endValue)
+ private val count = new AtomicInteger(0)
+
+ /** Record that a message was delivered (start tracking). */
+ private[grpc] def recordDelivery(ackId: String): Unit =
+ deliveryTimes.put(ackId, java.lang.Long.valueOf(System.nanoTime()))
+
+ /**
+ * Record that a message was acked/nacked (stop tracking, record latency).
+ * Latency is recorded in seconds, rounded up via `Math.ceil`.
+ * This matches Google's `MessageDispatcher`:
+ * {{{
+ * int ackLatency = Ints.saturatedCast(
+ * (long) Math.ceil((clock.millisTime() - receivedTimeMillis) / 1000D));
+ * ackLatencyDistribution.record(ackLatency);
+ * }}}
+ */
+ private[grpc] def recordCompletion(ackId: String): Unit = {
+ val startTime: java.lang.Long = deliveryTimes.remove(ackId)
+ if (startTime ne null) {
+ val elapsedNanos = System.nanoTime() - startTime.longValue()
+ val latencySeconds = math.ceil(elapsedNanos / 1000000000.0).toInt
+ record(latencySeconds)
+ }
+ }
+
+ /** Record completion for multiple ack IDs. */
+ private[grpc] def recordCompletions(ackIds: Iterable[String]): Unit =
+ ackIds.foreach(recordCompletion)
+
+ /**
+ * Record a value into the distribution.
+ * Mirrors `com.google.api.gax.core.Distribution.record(int)`:
+ * values >= endValue are clamped to endValue - 1.
+ */
+ private def record(value: Int): Unit = {
+ require(value >= 0, "value must be non-negative")
+ val clamped = if (value >= endValue) endValue - 1 else value
+ buckets.incrementAndGet(clamped)
+ count.incrementAndGet()
+ }
+
+ /**
+ * Get the current adaptive deadline in seconds, based on the configured percentile
+ * of observed processing times. Returns `initialDeadlineSeconds` if no observations yet.
+ *
+ * Uses the nearest-rank method, matching `com.google.api.gax.core.Distribution.getPercentile`:
+ * {{{
+ * long targetRank = (long) Math.ceil(percentile * count.get() / 100);
+ * long rank = 0;
+ * for (int i = 0; i < buckets.length(); i++) {
+ * rank += buckets.get(i);
+ * if (rank >= targetRank) return i;
+ * }
+ * return buckets.length();
+ * }}}
+ *
+ * The result is then clamped to `[minDeadlineSeconds, maxDeadlineSeconds]` and further
+ * bounded by Pub/Sub's stream ack deadline range (10–600s), matching
+ * `MessageDispatcher.computeDeadlineSeconds`.
+ */
+ def currentDeadlineSeconds: Int = {
+ val total = count.get()
+ if (total == 0) return initialDeadlineSeconds
+
+ // Nearest-rank method — read count before iterating (intentional low bias on concurrent access)
+ val targetRank = math.ceil(percentile * total / 100.0).toLong
+ var rank = 0L
+ var i = 0
+ while (i < buckets.length()) {
+ rank += buckets.get(i)
+ if (rank >= targetRank) {
+ return clampDeadline(i)
+ }
+ i += 1
+ }
+ clampDeadline(buckets.length())
+ }
+
+ /**
+ * Clamp deadline to configured bounds, then to Pub/Sub stream ack deadline range.
+ * Matches the clamping logic in `MessageDispatcher.computeDeadlineSeconds`.
+ */
+ private def clampDeadline(computed: Int): Int = {
+ var deadline = computed
+ if (deadline > maxDeadlineSeconds) deadline = maxDeadlineSeconds
+ if (deadline < minDeadlineSeconds) deadline = minDeadlineSeconds
+ if (deadline < AckDeadlineDistribution.MinStreamAckDeadlineSeconds)
+ deadline = AckDeadlineDistribution.MinStreamAckDeadlineSeconds
+ if (deadline > AckDeadlineDistribution.MaxStreamAckDeadlineSeconds)
+ deadline = AckDeadlineDistribution.MaxStreamAckDeadlineSeconds
+ deadline
+ }
+
+ /**
+ * Check if a tracked message has exceeded the max extension period.
+ * If so, remove it from tracking and return true.
+ */
+ private[grpc] def isExpired(ackId: String, now: Long): Boolean = {
+ val startTime: java.lang.Long = deliveryTimes.get(ackId)
+ if ((startTime ne null) && (now - startTime.longValue()) > maxAckExtensionPeriodNanos) {
+ deliveryTimes.remove(ackId)
+ true
+ } else {
+ false
+ }
+ }
+}
+
+@ApiMayChange
+object AckDeadlineDistribution {
+
+ /** Pub/Sub minimum stream ack deadline (matches `Subscriber.MIN_STREAM_ACK_DEADLINE`). */
+ final val MinStreamAckDeadlineSeconds = 10
+
+ /** Pub/Sub maximum stream ack deadline (matches `Subscriber.MAX_STREAM_ACK_DEADLINE`). */
+ final val MaxStreamAckDeadlineSeconds = 600
+
+ /** Default percentile used by Google's `MessageDispatcher` (`PERCENTILE_FOR_ACK_DEADLINE_UPDATES`). */
+ final val DefaultPercentile = 99.9
+
+ /** Default initial stream ack deadline (matches `Subscriber.STREAM_ACK_DEADLINE_DEFAULT`). */
+ final val DefaultInitialDeadlineSeconds = 60
+
+ /**
+ * Create a new [[AckDeadlineDistribution]].
+ *
+ * @param initialDeadlineSeconds deadline to use before enough data is collected
+ * (default 60, matching `Subscriber.STREAM_ACK_DEADLINE_DEFAULT`)
+ * @param minDeadlineSeconds minimum deadline in seconds (default 10, Pub/Sub minimum)
+ * @param maxDeadlineSeconds maximum deadline in seconds (default 600, Pub/Sub maximum)
+ * @param maxAckExtensionPeriodSeconds maximum total time to keep extending a message (default 3600 = 60 min)
+ * @param percentile percentile of processing time distribution to use
+ * (default 99.9, matching Google's `PERCENTILE_FOR_ACK_DEADLINE_UPDATES`)
+ */
+ def apply(
+ initialDeadlineSeconds: Int = DefaultInitialDeadlineSeconds,
+ minDeadlineSeconds: Int = MinStreamAckDeadlineSeconds,
+ maxDeadlineSeconds: Int = MaxStreamAckDeadlineSeconds,
+ maxAckExtensionPeriodSeconds: Int = 3600,
+ percentile: Double = DefaultPercentile): AckDeadlineDistribution =
+ new AckDeadlineDistribution(
+ initialDeadlineSeconds,
+ minDeadlineSeconds,
+ maxDeadlineSeconds,
+ maxAckExtensionPeriodSeconds.toLong * 1000000000L,
+ percentile)
+
+ /**
+ * Java API: Create a new [[AckDeadlineDistribution]].
+ */
+ def create(
+ initialDeadlineSeconds: Int,
+ minDeadlineSeconds: Int,
+ maxDeadlineSeconds: Int,
+ maxAckExtensionPeriodSeconds: Int,
+ percentile: Double): AckDeadlineDistribution =
+ apply(initialDeadlineSeconds, minDeadlineSeconds, maxDeadlineSeconds, maxAckExtensionPeriodSeconds, percentile)
+
+ /**
+ * Java API: Create with defaults matching Google's official client library.
+ */
+ def create(): AckDeadlineDistribution = apply()
+}
diff --git a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/AckDeadlineExtensionException.scala b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/AckDeadlineExtensionException.scala
new file mode 100644
index 000000000..2f6cabcb5
--- /dev/null
+++ b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/AckDeadlineExtensionException.scala
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc
+
+import org.apache.pekko.annotation.ApiMayChange
+
+/**
+ * Thrown when the background ack deadline extension ticker fails.
+ *
+ * This indicates that the `ModifyAckDeadline` RPC failed and lease management
+ * has stopped. Messages currently being processed may be redelivered by Pub/Sub
+ * after their ack deadline expires.
+ *
+ * @since 2.0.0
+ */
+@ApiMayChange
+class AckDeadlineExtensionException(message: String, cause: Throwable)
+ extends RuntimeException(message, cause)
diff --git a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/FlowControl.scala b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/FlowControl.scala
new file mode 100644
index 000000000..15b5030cd
--- /dev/null
+++ b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/FlowControl.scala
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc
+
+import org.apache.pekko.annotation.ApiMayChange
+
+import java.util.concurrent.atomic.AtomicLong
+
+/**
+ * Tracks the number of outstanding (unacknowledged) messages and provides
+ * a mechanism for the downstream acknowledge/nack stages to signal that
+ * messages have been processed, allowing the upstream gate to resume pulling.
+ *
+ * Create an instance and pass it to both `flowControlGate` (which applies
+ * backpressure) and `acknowledge`/`nack` overloads (which release permits).
+ *
+ * Usage:
+ * {{{
+ * val fc = FlowControl(maxOutstandingMessages = 1000)
+ *
+ * GooglePubSub.subscribe(request, 1.second)
+ * .via(GooglePubSub.flowControlGate(fc))
+ * .mapAsync(4)(processMessage)
+ * .map(msg => AcknowledgeRequest(sub, Seq(msg.ackId)))
+ * .to(GooglePubSub.acknowledge(parallelism = 1, fc))
+ * }}}
+ *
+ * @param maxOutstandingMessages maximum number of messages allowed in-flight
+ * @since 2.0.0
+ */
+@ApiMayChange
+final class FlowControl(val maxOutstandingMessages: Long) {
+ require(maxOutstandingMessages > 0, "maxOutstandingMessages must be > 0")
+
+ private[grpc] val outstanding = new AtomicLong(0)
+
+ @volatile private[grpc] var onRelease: () => Unit = () => ()
+
+ /** Current number of outstanding (unacknowledged) messages. */
+ def outstandingCount: Long = outstanding.get()
+
+ /** Release `count` permits, signalling that messages have been acknowledged. */
+ private[grpc] def release(count: Int): Unit = {
+ outstanding.addAndGet(-count)
+ onRelease()
+ }
+
+ /** Acquire one permit. Returns true if under the limit. */
+ private[grpc] def acquire(): Boolean =
+ outstanding.incrementAndGet() <= maxOutstandingMessages
+}
+
+@ApiMayChange
+object FlowControl {
+
+ /** Create a new [[FlowControl]] instance. */
+ def apply(maxOutstandingMessages: Long): FlowControl =
+ new FlowControl(maxOutstandingMessages)
+
+ /** Java API: Create a new [[FlowControl]] instance. */
+ def create(maxOutstandingMessages: Long): FlowControl =
+ new FlowControl(maxOutstandingMessages)
+}
diff --git a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/PubSubSettings.scala b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/PubSubSettings.scala
index 9ea3620d5..ad16e7c90 100644
--- a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/PubSubSettings.scala
+++ b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/PubSubSettings.scala
@@ -16,12 +16,8 @@ package org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc
import org.apache.pekko
import pekko.actor.ClassicActorSystemProvider
import pekko.stream.connectors.googlecloud.pubsub.grpc.impl.DeprecatedCredentials
-import com.google.auth.oauth2.GoogleCredentials
import com.typesafe.config.Config
import io.grpc.CallCredentials
-import io.grpc.auth.MoreCallCredentials
-
-import java.util.Collections
import scala.annotation.nowarn
@@ -92,11 +88,8 @@ object PubSubSettings {
config.getBoolean("use-tls"),
Some(config.getString("rootCa")).filter(_ != "none"),
config.getString("callCredentials") match {
- case "google-application-default" | "deprecated" =>
- val googleCredentials = GoogleCredentials.getApplicationDefault.createScoped(
- Collections.singletonList("https://www.googleapis.com/auth/pubsub"))
- Some(DeprecatedCredentials(MoreCallCredentials.from(googleCredentials)))
- case _ => None
+ case "none" => None
+ case _ => Some(DeprecatedCredentials.marker)
})
/**
diff --git a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/impl/DeprecatedCredentials.scala b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/impl/DeprecatedCredentials.scala
index ef8fc1593..7d2341a43 100644
--- a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/impl/DeprecatedCredentials.scala
+++ b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/impl/DeprecatedCredentials.scala
@@ -14,12 +14,12 @@
package org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.impl
import org.apache.pekko.annotation.InternalApi
-import io.grpc.CallCredentials
+import io.grpc.{ CallCredentials, Metadata }
import java.util.concurrent.Executor
/**
- * Used purely as a wrapper class to help migrate to common Google auth.
+ * Marker class indicating that credentials should be resolved via GoogleSettings.
*/
@InternalApi
private[grpc] final case class DeprecatedCredentials(underlying: CallCredentials) extends CallCredentials {
@@ -31,3 +31,17 @@ private[grpc] final case class DeprecatedCredentials(underlying: CallCredentials
override def thisUsesUnstableApi(): Unit = underlying.thisUsesUnstableApi()
}
+
+@InternalApi
+private[grpc] object DeprecatedCredentials {
+
+ /** A no-op marker instance used to signal "resolve via GoogleSettings". */
+ val marker: CallCredentials = DeprecatedCredentials(new CallCredentials {
+ override def applyRequestMetadata(requestInfo: CallCredentials.RequestInfo,
+ appExecutor: Executor,
+ applier: CallCredentials.MetadataApplier): Unit =
+ applier.apply(new Metadata())
+
+ override def thisUsesUnstableApi(): Unit = ()
+ })
+}
diff --git a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/impl/FlowControlGateStage.scala b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/impl/FlowControlGateStage.scala
new file mode 100644
index 000000000..81236f403
--- /dev/null
+++ b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/impl/FlowControlGateStage.scala
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.impl
+
+import org.apache.pekko.annotation.InternalApi
+import org.apache.pekko.stream._
+import org.apache.pekko.stream.stage._
+import org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.FlowControl
+
+/**
+ * INTERNAL API
+ *
+ * A GraphStage that gates elements through based on a shared [[FlowControl]] counter.
+ * It pulls from upstream only when the number of outstanding messages is below the limit.
+ * When downstream acknowledges/nacks via the same [[FlowControl]], the `onRelease` callback
+ * triggers a re-evaluation of whether to pull.
+ */
+@InternalApi
+private[grpc] final class FlowControlGateStage[T](flowControl: FlowControl)
+ extends GraphStage[FlowShape[T, T]] {
+
+ val in: Inlet[T] = Inlet("FlowControlGate.in")
+ val out: Outlet[T] = Outlet("FlowControlGate.out")
+ override val shape: FlowShape[T, T] = FlowShape(in, out)
+
+ override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
+ new GraphStageLogic(shape) with InHandler with OutHandler {
+
+ private var downstreamWaiting = false
+
+ private val releaseCallback: AsyncCallback[Unit] = getAsyncCallback[Unit] { _ =>
+ if (downstreamWaiting && flowControl.outstanding.get() < flowControl.maxOutstandingMessages &&
+ !hasBeenPulled(in)) {
+ downstreamWaiting = false
+ pull(in)
+ }
+ }
+
+ override def preStart(): Unit = {
+ flowControl.onRelease = () => releaseCallback.invoke(())
+ }
+
+ override def postStop(): Unit = {
+ flowControl.onRelease = () => ()
+ }
+
+ // InHandler
+ override def onPush(): Unit = {
+ val msg = grab(in)
+ flowControl.acquire()
+ push(out, msg)
+ }
+
+ // OutHandler
+ override def onPull(): Unit = {
+ if (flowControl.outstanding.get() < flowControl.maxOutstandingMessages) {
+ pull(in)
+ } else {
+ downstreamWaiting = true
+ }
+ }
+
+ setHandlers(in, out, this)
+ }
+}
diff --git a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/impl/PekkoGrpcSettings.scala b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/impl/PekkoGrpcSettings.scala
index fbe992805..30c5be98d 100644
--- a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/impl/PekkoGrpcSettings.scala
+++ b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/impl/PekkoGrpcSettings.scala
@@ -43,10 +43,8 @@ import scala.annotation.nowarn
.withFallback(sys.settings.config.getConfig("pekko.grpc.client.\"*\"")))
(config.callCredentials: @nowarn("msg=deprecated")) match {
- case None => settings
- case Some(DeprecatedCredentials(_)) => // Deprecated credentials were loaded from config so override them
- sys.log.warning(
- "Config path pekko.connectors.google.cloud.pubsub.grpc.callCredentials is deprecated, use pekko.connectors.google.credentials")
+ case None => settings // explicit opt-out (callCredentials = "none"), e.g. emulator
+ case Some(_: DeprecatedCredentials) => // default: resolve via GoogleSettings
val credentials = googleSettings.credentials.asGoogle(sys.dispatcher, googleSettings.requestSettings)
settings.withCallCredentials(MoreCallCredentials.from(credentials))
case Some(creds) => settings.withCallCredentials(creds)
diff --git a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/javadsl/GooglePubSub.scala b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/javadsl/GooglePubSub.scala
index b254474f0..4f8172945 100644
--- a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/javadsl/GooglePubSub.scala
+++ b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/javadsl/GooglePubSub.scala
@@ -14,14 +14,25 @@
package org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.javadsl
import java.time.Duration
-import java.util.concurrent.{ CompletableFuture, CompletionStage }
+import java.util.concurrent.{ CompletableFuture, CompletionStage, ConcurrentHashMap }
import org.apache.pekko
import pekko.actor.Cancellable
-import pekko.stream.{ Attributes, Materializer }
-import pekko.stream.javadsl.{ Flow, Keep, Sink, Source }
+import pekko.annotation.ApiMayChange
+import pekko.stream.{ Attributes, KillSwitches, Materializer, RestartSettings }
+import pekko.stream.javadsl.{ Flow, Keep, RestartSource, Sink, Source }
+import pekko.stream.connectors.googlecloud.pubsub.grpc.{
+ AckDeadlineDistribution,
+ AckDeadlineExtensionException,
+ FlowControl
+}
+import pekko.stream.connectors.googlecloud.pubsub.grpc.impl.FlowControlGateStage
import pekko.{ Done, NotUsed }
import com.google.pubsub.v1._
+import scala.concurrent.{ ExecutionContext, Future }
+import scala.jdk.CollectionConverters._
+import scala.jdk.DurationConverters._
+
/**
* Google Pub/Sub Pekko Stream operator factory.
*/
@@ -79,6 +90,27 @@ object GooglePubSub {
.mapMaterializedValue(flattenCs(_))
.mapMaterializedValue(_.toCompletableFuture)
+ /**
+ * Create a source that emits messages for a given subscription using a StreamingPullRequest.
+ *
+ * Automatically reconnects when the server closes the StreamingPull connection (due to idle
+ * timeout, rebalancing, etc.), matching the behavior of Google's official client library.
+ * All errors trigger reconnection with exponential backoff, including non-retryable errors
+ * such as PERMISSION_DENIED or NOT_FOUND. Use `RestartSettings.withMaxRestarts` to bound
+ * the total number of restart attempts; when exhausted the stream fails with the last error.
+ *
+ * @param request the subscription FQRS and ack deadline fields are mandatory for the request
+ * @param pollInterval time between StreamingPullRequest messages are being sent
+ * @param restartSettings settings for the exponential backoff reconnection behavior
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def subscribe(request: StreamingPullRequest,
+ pollInterval: Duration,
+ restartSettings: RestartSettings): Source[ReceivedMessage, NotUsed] =
+ RestartSource.withBackoff(restartSettings,
+ () => subscribe(request, pollInterval).mapMaterializedValue[NotUsed](_ => NotUsed))
+
/**
* Create a source that emits messages for a given subscription using a synchronous PullRequest.
*
@@ -110,6 +142,323 @@ object GooglePubSub {
.mapMaterializedValue(flattenCs(_))
.mapMaterializedValue(_.toCompletableFuture)
+ /**
+ * Create a flow that accepts ack deadline modifications.
+ *
+ * This can be used to extend the acknowledgement deadline for messages that require
+ * longer processing times, preventing Pub/Sub from redelivering them.
+ *
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def modifyAckDeadlineFlow(): Flow[ModifyAckDeadlineRequest, ModifyAckDeadlineRequest, NotUsed] =
+ Flow
+ .fromMaterializer { (mat, attr) =>
+ Flow
+ .create[ModifyAckDeadlineRequest]()
+ .mapAsyncUnordered(1,
+ req =>
+ subscriber(mat, attr).client.modifyAckDeadline(req)
+ .thenApply[ModifyAckDeadlineRequest](_ => req))
+ }
+ .mapMaterializedValue(_ => NotUsed)
+
+ /**
+ * Create a sink that accepts ack deadline modifications.
+ *
+ * This can be used to extend the acknowledgement deadline for messages that require
+ * longer processing times, preventing Pub/Sub from redelivering them.
+ *
+ * The materialized value completes on stream completion.
+ *
+ * @param parallelism controls how many modifications can be in-flight at any given time
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def modifyAckDeadline(parallelism: Int): Sink[ModifyAckDeadlineRequest, CompletionStage[Done]] =
+ Sink
+ .fromMaterializer { (mat, attr) =>
+ Flow
+ .create[ModifyAckDeadlineRequest]()
+ .mapAsyncUnordered(parallelism, subscriber(mat, attr).client.modifyAckDeadline(_))
+ .toMat(Sink.ignore(), Keep.right[NotUsed, CompletionStage[Done]])
+ }
+ .mapMaterializedValue(flattenCs(_))
+
+ /**
+ * Create a flow that automatically extends ack deadlines for messages passing through it.
+ *
+ * Uses a default `maxAckExtensionPeriod` of 60 minutes (matching Google's client library).
+ * Messages are NOT removed from tracking upon acknowledgment; extending the deadline of an
+ * already-acknowledged message is a safe server-side no-op. For tracking that stops on
+ * acknowledgment, use the adaptive variant with [[AckDeadlineDistribution]].
+ *
+ * If the background lease-extension ticker fails (e.g. due to a network error in the
+ * `ModifyAckDeadline` RPC), the main stream will be aborted immediately with an
+ * [[org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.AckDeadlineExtensionException]],
+ * even if the stream is idle.
+ *
+ * @param subscription the fully qualified subscription resource string
+ * @param extensionInterval how often to extend deadlines (should be less than the ack deadline)
+ * @param ackDeadlineSeconds the new deadline to set on each extension
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def autoExtendAckDeadlines(
+ subscription: String,
+ extensionInterval: Duration,
+ ackDeadlineSeconds: Int): Flow[ReceivedMessage, ReceivedMessage, NotUsed] =
+ autoExtendAckDeadlines(subscription, extensionInterval, ackDeadlineSeconds, Duration.ofMinutes(60))
+
+ /**
+ * Create a flow that automatically extends ack deadlines for messages passing through it.
+ *
+ * A background timer periodically sends `ModifyAckDeadline` RPCs for all tracked messages,
+ * preventing Pub/Sub from redelivering them while they are being processed downstream.
+ * Messages are tracked from the moment they enter the flow until the stream completes, fails,
+ * or `maxAckExtensionPeriod` elapses. Messages are NOT removed from tracking upon acknowledgment;
+ * extending the deadline of an already-acknowledged message is a safe server-side no-op.
+ * For tracking that stops on acknowledgment, use the adaptive variant with [[AckDeadlineDistribution]].
+ *
+ * If the background lease-extension ticker fails (e.g. due to a network error in the
+ * `ModifyAckDeadline` RPC), the main stream will be aborted immediately with an
+ * [[org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.AckDeadlineExtensionException]],
+ * even if the stream is idle.
+ *
+ * @param subscription the fully qualified subscription resource string
+ * @param extensionInterval how often to extend deadlines (should be less than the ack deadline)
+ * @param ackDeadlineSeconds the new deadline to set on each extension
+ * @param maxAckExtensionPeriod maximum duration to keep extending a message's deadline
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def autoExtendAckDeadlines(
+ subscription: String,
+ extensionInterval: Duration,
+ ackDeadlineSeconds: Int,
+ maxAckExtensionPeriod: Duration): Flow[ReceivedMessage, ReceivedMessage, NotUsed] =
+ Flow
+ .fromMaterializer { (mat, attr) =>
+ val client = subscriber(mat, attr).client
+ val tracked = new ConcurrentHashMap[String, java.lang.Long]()
+ val maxNanos = maxAckExtensionPeriod.toNanos
+ val killSwitch = KillSwitches.shared("autoExtendAckDeadlines")
+
+ val tickerPair = Source
+ .tick(extensionInterval, extensionInterval,
+ ModifyAckDeadlineRequest.getDefaultInstance)
+ .mapAsync(1,
+ (_: ModifyAckDeadlineRequest) => {
+ val now = System.nanoTime()
+ // Remove expired entries
+ tracked.asScala.foreach { case (ackId, entryTime) =>
+ if (now - entryTime > maxNanos) tracked.remove(ackId)
+ }
+ val ids = tracked.asScala.keys.toSeq
+ if (ids.nonEmpty)
+ client
+ .modifyAckDeadline(
+ ModifyAckDeadlineRequest.newBuilder()
+ .setSubscription(subscription)
+ .addAllAckIds(ids.asJava)
+ .setAckDeadlineSeconds(ackDeadlineSeconds)
+ .build())
+ .thenApply[Done](_ => Done.done())
+ else
+ CompletableFuture.completedFuture(Done.done())
+ })
+ .toMat(Sink.ignore(), Keep.both[pekko.actor.Cancellable, CompletionStage[Done]])
+ .run(mat)
+
+ val ticker = tickerPair.first
+ val tickerDone = tickerPair.second
+ tickerDone.whenComplete((_, ex) => {
+ if (ex != null)
+ killSwitch.abort(new AckDeadlineExtensionException(
+ "Lease management ticker failed; ack deadline extensions have stopped", ex))
+ })
+
+ Flow.create[ReceivedMessage]()
+ .via(killSwitch.flow[ReceivedMessage])
+ .map(((msg: ReceivedMessage) => {
+ tracked.put(msg.getAckId, java.lang.Long.valueOf(System.nanoTime()))
+ msg
+ }): pekko.japi.function.Function[ReceivedMessage, ReceivedMessage])
+ .watchTermination((_, done: CompletionStage[Done]) => {
+ done.whenComplete((_, _) => {
+ ticker.cancel()
+ tracked.clear()
+ })
+ NotUsed
+ })
+ }
+ .mapMaterializedValue(_ => NotUsed)
+
+ /**
+ * Create a flow that automatically extends ack deadlines using an adaptive deadline
+ * based on observed processing times, matching Google's official client library behavior.
+ *
+ * The same [[AckDeadlineDistribution]] must be passed to the acknowledge/nack operators
+ * so that processing latencies are recorded.
+ *
+ * @param subscription the fully qualified subscription resource string
+ * @param extensionInterval how often to extend deadlines (should be less than the ack deadline)
+ * @param distribution shared distribution that tracks processing times and computes adaptive deadlines
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def autoExtendAckDeadlines(
+ subscription: String,
+ extensionInterval: Duration,
+ distribution: AckDeadlineDistribution): Flow[ReceivedMessage, ReceivedMessage, NotUsed] =
+ Flow
+ .fromMaterializer { (mat, attr) =>
+ val client = subscriber(mat, attr).client
+ val killSwitch = KillSwitches.shared("autoExtendAckDeadlines")
+
+ val tickerPair = Source
+ .tick(extensionInterval, extensionInterval,
+ ModifyAckDeadlineRequest.getDefaultInstance)
+ .mapAsync(1,
+ (_: ModifyAckDeadlineRequest) => {
+ val now = System.nanoTime()
+ distribution.deliveryTimes.asScala.foreach { case (ackId, _) =>
+ distribution.isExpired(ackId, now)
+ }
+ val ids = distribution.deliveryTimes.asScala.keys.toSeq
+ if (ids.nonEmpty) {
+ val deadline = distribution.currentDeadlineSeconds
+ client
+ .modifyAckDeadline(
+ ModifyAckDeadlineRequest.newBuilder()
+ .setSubscription(subscription)
+ .addAllAckIds(ids.asJava)
+ .setAckDeadlineSeconds(deadline)
+ .build())
+ .thenApply[Done](_ => Done.done())
+ } else
+ CompletableFuture.completedFuture(Done.done())
+ })
+ .toMat(Sink.ignore(), Keep.both[pekko.actor.Cancellable, CompletionStage[Done]])
+ .run(mat)
+
+ val ticker = tickerPair.first
+ val tickerDone = tickerPair.second
+ tickerDone.whenComplete((_, ex) => {
+ if (ex != null)
+ killSwitch.abort(new AckDeadlineExtensionException(
+ "Lease management ticker failed; ack deadline extensions have stopped", ex))
+ })
+
+ Flow.create[ReceivedMessage]()
+ .via(killSwitch.flow[ReceivedMessage])
+ .map(((msg: ReceivedMessage) => {
+ distribution.recordDelivery(msg.getAckId)
+ msg
+ }): pekko.japi.function.Function[ReceivedMessage, ReceivedMessage])
+ .watchTermination((_, done: CompletionStage[Done]) => {
+ done.whenComplete((_, _) => {
+ ticker.cancel()
+ distribution.deliveryTimes.clear()
+ })
+ NotUsed
+ })
+ }
+ .mapMaterializedValue(_ => NotUsed)
+
+ /**
+ * Create a flow that modifies the ack deadline for each message using a dynamic function.
+ *
+ * This allows per-message deadline control — for example, setting longer deadlines for
+ * messages that are expected to take longer to process.
+ *
+ * Returning 0 from the function is equivalent to a nack (immediate redelivery).
+ *
+ * @param subscription the fully qualified subscription resource string
+ * @param parallelism controls how many modifications can be in-flight at any given time
+ * @param deadlineFn function that computes the ack deadline in seconds for each message
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def modifyAckDeadlineDynamic(
+ subscription: String,
+ parallelism: Int,
+ deadlineFn: pekko.japi.function.Function[ReceivedMessage, Integer])
+ : Flow[ReceivedMessage, ReceivedMessage, NotUsed] =
+ Flow
+ .fromMaterializer { (mat, attr) =>
+ Flow
+ .create[ReceivedMessage]()
+ .mapAsync(parallelism,
+ (msg: ReceivedMessage) => {
+ val deadline: Int = deadlineFn(msg)
+ subscriber(mat, attr).client
+ .modifyAckDeadline(
+ ModifyAckDeadlineRequest.newBuilder()
+ .setSubscription(subscription)
+ .addAckIds(msg.getAckId)
+ .setAckDeadlineSeconds(deadline)
+ .build())
+ .thenApply[ReceivedMessage](_ => msg)
+ })
+ }
+ .mapMaterializedValue(_ => NotUsed)
+
+ /**
+ * Create a flow that nacks messages by setting the ack deadline to 0,
+ * causing Pub/Sub to immediately redeliver them.
+ *
+ * Accepts [[AcknowledgeRequest]] for consistency with the acknowledge API.
+ *
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def nackFlow(): Flow[AcknowledgeRequest, AcknowledgeRequest, NotUsed] =
+ Flow
+ .fromMaterializer { (mat, attr) =>
+ Flow
+ .create[AcknowledgeRequest]()
+ .mapAsyncUnordered(1,
+ (req: AcknowledgeRequest) =>
+ subscriber(mat, attr).client
+ .modifyAckDeadline(
+ ModifyAckDeadlineRequest.newBuilder()
+ .setSubscription(req.getSubscription)
+ .addAllAckIds(req.getAckIdsList)
+ .setAckDeadlineSeconds(0)
+ .build())
+ .thenApply[AcknowledgeRequest](_ => req))
+ }
+ .mapMaterializedValue(_ => NotUsed)
+
+ /**
+ * Create a sink that nacks messages by setting the ack deadline to 0,
+ * causing Pub/Sub to immediately redeliver them.
+ *
+ * The materialized value completes on stream completion.
+ *
+ * @param parallelism controls how many nacks can be in-flight at any given time
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def nack(parallelism: Int): Sink[AcknowledgeRequest, CompletionStage[Done]] =
+ Sink
+ .fromMaterializer { (mat, attr) =>
+ Flow
+ .create[AcknowledgeRequest]()
+ .mapAsyncUnordered(parallelism,
+ (req: AcknowledgeRequest) =>
+ subscriber(mat, attr).client
+ .modifyAckDeadline(
+ ModifyAckDeadlineRequest.newBuilder()
+ .setSubscription(req.getSubscription)
+ .addAllAckIds(req.getAckIdsList)
+ .setAckDeadlineSeconds(0)
+ .build()))
+ .toMat(Sink.ignore(), Keep.right[NotUsed, CompletionStage[Done]])
+ }
+ .mapMaterializedValue(flattenCs(_))
+
/**
* Create a flow that accepts consumed message acknowledgements.
*/
@@ -141,6 +490,237 @@ object GooglePubSub {
}
.mapMaterializedValue(flattenCs(_))
+ /**
+ * Create a flow that gates messages based on [[FlowControl]], applying backpressure
+ * when the number of outstanding (unacknowledged) messages reaches the limit.
+ *
+ * @param flowControl shared flow control instance
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def flowControlGate(flowControl: FlowControl): Flow[ReceivedMessage, ReceivedMessage, NotUsed] =
+ Flow.fromGraph(new FlowControlGateStage(flowControl))
+
+ /**
+ * Create a flow that acknowledges messages and releases [[FlowControl]] permits.
+ *
+ * @param flowControl shared flow control instance
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def acknowledgeFlow(flowControl: FlowControl): Flow[AcknowledgeRequest, AcknowledgeRequest, NotUsed] =
+ Flow
+ .fromMaterializer { (mat, attr) =>
+ Flow
+ .create[AcknowledgeRequest]()
+ .mapAsync(1,
+ (req: AcknowledgeRequest) =>
+ subscriber(mat, attr).client
+ .acknowledge(req)
+ .thenApply[AcknowledgeRequest] { _ =>
+ flowControl.release(req.getAckIdsCount)
+ req
+ })
+ }
+ .mapMaterializedValue(_ => NotUsed)
+
+ /**
+ * Create a sink that acknowledges messages and releases [[FlowControl]] permits.
+ *
+ * @param parallelism controls how many acknowledgements can be in-flight at any given time
+ * @param flowControl shared flow control instance
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def acknowledge(parallelism: Int, flowControl: FlowControl): Sink[AcknowledgeRequest, CompletionStage[Done]] =
+ Sink
+ .fromMaterializer { (mat, attr) =>
+ Flow
+ .create[AcknowledgeRequest]()
+ .mapAsyncUnordered(parallelism,
+ (req: AcknowledgeRequest) =>
+ subscriber(mat, attr).client
+ .acknowledge(req)
+ .thenApply[Done] { _ =>
+ flowControl.release(req.getAckIdsCount)
+ Done.done()
+ })
+ .toMat(Sink.ignore(), Keep.right[NotUsed, CompletionStage[Done]])
+ }
+ .mapMaterializedValue(flattenCs(_))
+
+ /**
+ * Create a flow that nacks messages and releases [[FlowControl]] permits.
+ *
+ * @param flowControl shared flow control instance
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def nackFlow(flowControl: FlowControl): Flow[AcknowledgeRequest, AcknowledgeRequest, NotUsed] =
+ Flow
+ .fromMaterializer { (mat, attr) =>
+ Flow
+ .create[AcknowledgeRequest]()
+ .mapAsync(1,
+ (req: AcknowledgeRequest) =>
+ subscriber(mat, attr).client
+ .modifyAckDeadline(
+ ModifyAckDeadlineRequest.newBuilder()
+ .setSubscription(req.getSubscription)
+ .addAllAckIds(req.getAckIdsList)
+ .setAckDeadlineSeconds(0)
+ .build())
+ .thenApply[AcknowledgeRequest] { _ =>
+ flowControl.release(req.getAckIdsCount)
+ req
+ })
+ }
+ .mapMaterializedValue(_ => NotUsed)
+
+ /**
+ * Create a sink that nacks messages and releases [[FlowControl]] permits.
+ *
+ * @param parallelism controls how many nacks can be in-flight at any given time
+ * @param flowControl shared flow control instance
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def nack(parallelism: Int, flowControl: FlowControl): Sink[AcknowledgeRequest, CompletionStage[Done]] =
+ Sink
+ .fromMaterializer { (mat, attr) =>
+ Flow
+ .create[AcknowledgeRequest]()
+ .mapAsyncUnordered(parallelism,
+ (req: AcknowledgeRequest) =>
+ subscriber(mat, attr).client
+ .modifyAckDeadline(
+ ModifyAckDeadlineRequest.newBuilder()
+ .setSubscription(req.getSubscription)
+ .addAllAckIds(req.getAckIdsList)
+ .setAckDeadlineSeconds(0)
+ .build())
+ .thenApply[Done] { _ =>
+ flowControl.release(req.getAckIdsCount)
+ Done.done()
+ })
+ .toMat(Sink.ignore(), Keep.right[NotUsed, CompletionStage[Done]])
+ }
+ .mapMaterializedValue(flattenCs(_))
+
+ /**
+ * Create a flow that acknowledges messages and records processing latencies
+ * in the [[AckDeadlineDistribution]] for adaptive deadline computation.
+ *
+ * @param distribution shared distribution for adaptive deadline tracking
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def acknowledgeFlow(
+ distribution: AckDeadlineDistribution): Flow[AcknowledgeRequest, AcknowledgeRequest, NotUsed] =
+ Flow
+ .fromMaterializer { (mat, attr) =>
+ Flow
+ .create[AcknowledgeRequest]()
+ .mapAsync(1,
+ (req: AcknowledgeRequest) =>
+ subscriber(mat, attr).client
+ .acknowledge(req)
+ .thenApply[AcknowledgeRequest] { _ =>
+ distribution.recordCompletions(req.getAckIdsList.asScala)
+ req
+ })
+ }
+ .mapMaterializedValue(_ => NotUsed)
+
+ /**
+ * Create a sink that acknowledges messages and records processing latencies
+ * in the [[AckDeadlineDistribution]] for adaptive deadline computation.
+ *
+ * @param parallelism controls how many acknowledgements can be in-flight at any given time
+ * @param distribution shared distribution for adaptive deadline tracking
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def acknowledge(parallelism: Int,
+ distribution: AckDeadlineDistribution): Sink[AcknowledgeRequest, CompletionStage[Done]] =
+ Sink
+ .fromMaterializer { (mat, attr) =>
+ Flow
+ .create[AcknowledgeRequest]()
+ .mapAsyncUnordered(parallelism,
+ (req: AcknowledgeRequest) =>
+ subscriber(mat, attr).client
+ .acknowledge(req)
+ .thenApply[Done] { _ =>
+ distribution.recordCompletions(req.getAckIdsList.asScala)
+ Done.done()
+ })
+ .toMat(Sink.ignore(), Keep.right[NotUsed, CompletionStage[Done]])
+ }
+ .mapMaterializedValue(flattenCs(_))
+
+ /**
+ * Create a flow that nacks messages and records processing latencies
+ * in the [[AckDeadlineDistribution]] for adaptive deadline computation.
+ *
+ * @param distribution shared distribution for adaptive deadline tracking
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def nackFlow(
+ distribution: AckDeadlineDistribution): Flow[AcknowledgeRequest, AcknowledgeRequest, NotUsed] =
+ Flow
+ .fromMaterializer { (mat, attr) =>
+ Flow
+ .create[AcknowledgeRequest]()
+ .mapAsync(1,
+ (req: AcknowledgeRequest) =>
+ subscriber(mat, attr).client
+ .modifyAckDeadline(
+ ModifyAckDeadlineRequest.newBuilder()
+ .setSubscription(req.getSubscription)
+ .addAllAckIds(req.getAckIdsList)
+ .setAckDeadlineSeconds(0)
+ .build())
+ .thenApply[AcknowledgeRequest] { _ =>
+ distribution.recordCompletions(req.getAckIdsList.asScala)
+ req
+ })
+ }
+ .mapMaterializedValue(_ => NotUsed)
+
+ /**
+ * Create a sink that nacks messages and records processing latencies
+ * in the [[AckDeadlineDistribution]] for adaptive deadline computation.
+ *
+ * @param parallelism controls how many nacks can be in-flight at any given time
+ * @param distribution shared distribution for adaptive deadline tracking
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def nack(parallelism: Int,
+ distribution: AckDeadlineDistribution): Sink[AcknowledgeRequest, CompletionStage[Done]] =
+ Sink
+ .fromMaterializer { (mat, attr) =>
+ Flow
+ .create[AcknowledgeRequest]()
+ .mapAsyncUnordered(parallelism,
+ (req: AcknowledgeRequest) =>
+ subscriber(mat, attr).client
+ .modifyAckDeadline(
+ ModifyAckDeadlineRequest.newBuilder()
+ .setSubscription(req.getSubscription)
+ .addAllAckIds(req.getAckIdsList)
+ .setAckDeadlineSeconds(0)
+ .build())
+ .thenApply[Done] { _ =>
+ distribution.recordCompletions(req.getAckIdsList.asScala)
+ Done.done()
+ })
+ .toMat(Sink.ignore(), Keep.right[NotUsed, CompletionStage[Done]])
+ }
+ .mapMaterializedValue(flattenCs(_))
+
private def flattenCs[T](f: CompletionStage[_ <: CompletionStage[T]]): CompletionStage[T] =
f.thenCompose((t: CompletionStage[T]) => t)
diff --git a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/javadsl/GrpcSubscriber.scala b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/javadsl/GrpcSubscriber.scala
index de2e738ad..bd9991076 100644
--- a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/javadsl/GrpcSubscriber.scala
+++ b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/javadsl/GrpcSubscriber.scala
@@ -31,39 +31,35 @@ import com.google.pubsub.v1.{ SubscriberClient => JavaSubscriberClient }
/**
* Holds the gRPC java subscriber client instance.
*/
-final class GrpcSubscriber private (settings: PubSubSettings, googleSettings: GoogleSettings, sys: ActorSystem) {
-
- @ApiMayChange
- final val client =
- JavaSubscriberClient.create(PekkoGrpcSettings.fromPubSubSettings(settings, googleSettings)(sys), sys)
-
- sys.registerOnTermination(client.close())
-}
+final class GrpcSubscriber private[grpc] (@ApiMayChange final val client: JavaSubscriberClient)
object GrpcSubscriber {
/**
- * Creates a publisher with the new actors API.
+ * Creates a subscriber with the new actors API.
*/
def create(settings: PubSubSettings,
googleSettings: GoogleSettings,
sys: ClassicActorSystemProvider): GrpcSubscriber =
create(settings, googleSettings, sys.classicSystem)
- def create(settings: PubSubSettings, googleSettings: GoogleSettings, sys: ActorSystem): GrpcSubscriber =
- new GrpcSubscriber(settings, googleSettings, sys)
+ def create(settings: PubSubSettings, googleSettings: GoogleSettings, sys: ActorSystem): GrpcSubscriber = {
+ val c = JavaSubscriberClient.create(PekkoGrpcSettings.fromPubSubSettings(settings, googleSettings)(sys), sys)
+ sys.registerOnTermination(c.close())
+ new GrpcSubscriber(c)
+ }
/**
- * Creates a publisher with the new actors API.
+ * Creates a subscriber with the new actors API.
*/
def create(settings: PubSubSettings, sys: ClassicActorSystemProvider): GrpcSubscriber =
create(settings, sys.classicSystem)
def create(settings: PubSubSettings, sys: ActorSystem): GrpcSubscriber =
- new GrpcSubscriber(settings, GoogleSettings(sys), sys)
+ create(settings, GoogleSettings(sys), sys)
/**
- * Creates a publisher with the new actors API.
+ * Creates a subscriber with the new actors API.
*/
def create(sys: ClassicActorSystemProvider): GrpcSubscriber = create(sys.classicSystem)
diff --git a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/scaladsl/GooglePubSub.scala b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/scaladsl/GooglePubSub.scala
index b7bba94a8..d0746242d 100644
--- a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/scaladsl/GooglePubSub.scala
+++ b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/scaladsl/GooglePubSub.scala
@@ -15,13 +15,22 @@ package org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.scaladsl
import org.apache.pekko
import pekko.actor.Cancellable
-import pekko.stream.{ Attributes, Materializer }
-import pekko.stream.scaladsl.{ Flow, Keep, Sink, Source }
+import pekko.annotation.ApiMayChange
+import pekko.stream.{ Attributes, KillSwitches, Materializer, RestartSettings }
+import pekko.stream.scaladsl.{ Flow, Keep, RestartSource, Sink, Source }
+import pekko.stream.connectors.googlecloud.pubsub.grpc.{
+ AckDeadlineDistribution,
+ AckDeadlineExtensionException,
+ FlowControl
+}
+import pekko.stream.connectors.googlecloud.pubsub.grpc.impl.FlowControlGateStage
import pekko.{ Done, NotUsed }
import com.google.pubsub.v1.pubsub._
+import java.util.concurrent.ConcurrentHashMap
import scala.concurrent.duration._
import scala.concurrent.{ ExecutionContext, Future, Promise }
+import scala.jdk.CollectionConverters._
/**
* Google Pub/Sub Pekko Stream operator factory.
@@ -75,6 +84,29 @@ object GooglePubSub {
}
.mapMaterializedValue(_.flatMap(identity)(ExecutionContext.parasitic))
+ /**
+ * Create a source that emits messages for a given subscription using a StreamingPullRequest.
+ *
+ * Automatically reconnects when the server closes the StreamingPull connection (due to idle
+ * timeout, rebalancing, etc.), matching the behavior of Google's official client library.
+ * All errors trigger reconnection with exponential backoff, including non-retryable errors
+ * such as PERMISSION_DENIED or NOT_FOUND. Use `RestartSettings.withMaxRestarts` to bound
+ * the total number of restart attempts; when exhausted the stream fails with the last error.
+ *
+ * @param request the subscription FQRS and ack deadline fields are mandatory for the request
+ * @param pollInterval time between StreamingPullRequest messages are being sent
+ * @param restartSettings settings for the exponential backoff reconnection behavior
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def subscribe(
+ request: StreamingPullRequest,
+ pollInterval: FiniteDuration,
+ restartSettings: RestartSettings): Source[ReceivedMessage, NotUsed] =
+ RestartSource.withBackoff(restartSettings) { () =>
+ subscribe(request, pollInterval).mapMaterializedValue(_ => NotUsed)
+ }
+
/**
* Create a source that emits messages for a given subscription using a synchronous PullRequest.
*
@@ -99,6 +131,316 @@ object GooglePubSub {
}
.mapMaterializedValue(_.flatMap(identity)(ExecutionContext.parasitic))
+ /**
+ * Create a flow that accepts ack deadline modifications.
+ *
+ * This can be used to extend the acknowledgement deadline for messages that require
+ * longer processing times, preventing Pub/Sub from redelivering them.
+ *
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def modifyAckDeadlineFlow(): Flow[ModifyAckDeadlineRequest, ModifyAckDeadlineRequest, NotUsed] =
+ Flow
+ .fromMaterializer { (mat, attr) =>
+ Flow[ModifyAckDeadlineRequest]
+ .mapAsync(1)(req =>
+ subscriber(mat, attr).client
+ .modifyAckDeadline(req)
+ .map(_ => req)(mat.executionContext))
+ }
+ .mapMaterializedValue(_ => NotUsed)
+
+ /**
+ * Create a sink that accepts ack deadline modifications.
+ *
+ * This can be used to extend the acknowledgement deadline for messages that require
+ * longer processing times, preventing Pub/Sub from redelivering them.
+ *
+ * The materialized value completes on stream completion.
+ *
+ * @param parallelism controls how many modifications can be in-flight at any given time
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def modifyAckDeadline(parallelism: Int): Sink[ModifyAckDeadlineRequest, Future[Done]] =
+ Sink
+ .fromMaterializer { (mat, attr) =>
+ Flow[ModifyAckDeadlineRequest]
+ .mapAsyncUnordered(parallelism)(subscriber(mat, attr).client.modifyAckDeadline)
+ .toMat(Sink.ignore)(Keep.right)
+ }
+ .mapMaterializedValue(_.flatMap(identity)(ExecutionContext.parasitic))
+
+ /**
+ * Create a flow that automatically extends ack deadlines for messages passing through it.
+ *
+ * A background timer periodically sends `ModifyAckDeadline` RPCs for all tracked messages,
+ * preventing Pub/Sub from redelivering them while they are being processed downstream.
+ * Messages are tracked from the moment they enter the flow until the stream completes, fails,
+ * or `maxAckExtensionPeriod` elapses (default 60 minutes, matching Google's client library).
+ * Messages are NOT removed from tracking upon acknowledgment; extending the deadline of an
+ * already-acknowledged message is a safe server-side no-op. For tracking that stops on
+ * acknowledgment, use the adaptive variant with [[AckDeadlineDistribution]].
+ *
+ * If the background lease-extension ticker fails (e.g. due to a network error in the
+ * `ModifyAckDeadline` RPC), the main stream will be aborted immediately with an
+ * [[org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.AckDeadlineExtensionException]],
+ * even if the stream is idle.
+ *
+ * Usage:
+ * {{{
+ * GooglePubSub.subscribe(request, 1.second)
+ * .via(GooglePubSub.autoExtendAckDeadlines(subscriptionFqrs, 8.seconds, 30))
+ * .mapAsync(4)(processMessage)
+ * .map(msg => AcknowledgeRequest(subscriptionFqrs, Seq(msg.ackId)))
+ * .to(GooglePubSub.acknowledge(parallelism = 1))
+ * }}}
+ *
+ * @param subscription the fully qualified subscription resource string
+ * @param extensionInterval how often to extend deadlines (should be less than the ack deadline)
+ * @param ackDeadlineSeconds the new deadline to set on each extension
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def autoExtendAckDeadlines(
+ subscription: String,
+ extensionInterval: FiniteDuration,
+ ackDeadlineSeconds: Int): Flow[ReceivedMessage, ReceivedMessage, NotUsed] =
+ autoExtendAckDeadlines(subscription, extensionInterval, ackDeadlineSeconds, 60.minutes)
+
+ /**
+ * Create a flow that automatically extends ack deadlines for messages passing through it.
+ *
+ * A background timer periodically sends `ModifyAckDeadline` RPCs for all tracked messages,
+ * preventing Pub/Sub from redelivering them while they are being processed downstream.
+ * Messages are tracked from the moment they enter the flow until the stream completes, fails,
+ * or `maxAckExtensionPeriod` elapses. Messages are NOT removed from tracking upon acknowledgment;
+ * extending the deadline of an already-acknowledged message is a safe server-side no-op.
+ * For tracking that stops on acknowledgment, use the adaptive variant with [[AckDeadlineDistribution]].
+ *
+ * If the background lease-extension ticker fails (e.g. due to a network error in the
+ * `ModifyAckDeadline` RPC), the main stream will be aborted immediately with an
+ * [[org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.AckDeadlineExtensionException]],
+ * even if the stream is idle.
+ *
+ * @param subscription the fully qualified subscription resource string
+ * @param extensionInterval how often to extend deadlines (should be less than the ack deadline)
+ * @param ackDeadlineSeconds the new deadline to set on each extension
+ * @param maxAckExtensionPeriod maximum duration to keep extending a message's deadline (default 60 minutes)
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def autoExtendAckDeadlines(
+ subscription: String,
+ extensionInterval: FiniteDuration,
+ ackDeadlineSeconds: Int,
+ maxAckExtensionPeriod: FiniteDuration): Flow[ReceivedMessage, ReceivedMessage, NotUsed] =
+ Flow.fromMaterializer { (mat, attr) =>
+ val client = subscriber(mat, attr).client
+ val tracked = new ConcurrentHashMap[String, Long]()
+ val maxNanos = maxAckExtensionPeriod.toNanos
+ val killSwitch = KillSwitches.shared("autoExtendAckDeadlines")
+
+ val (ticker, tickerDone) = Source
+ .tick(extensionInterval, extensionInterval, ())
+ .mapAsync(1) { _ =>
+ val now = System.nanoTime()
+ // Remove expired entries
+ tracked.asScala.foreach { case (ackId, entryTime) =>
+ if (now - entryTime > maxNanos) tracked.remove(ackId)
+ }
+ val ids = tracked.asScala.keys.toSeq
+ if (ids.nonEmpty)
+ client
+ .modifyAckDeadline(ModifyAckDeadlineRequest(subscription, ids, ackDeadlineSeconds))
+ .map(_ => ())(ExecutionContext.parasitic)
+ else
+ Future.successful(())
+ }
+ .toMat(Sink.ignore)(Keep.both)
+ .run()(mat)
+
+ tickerDone.onComplete {
+ case scala.util.Failure(ex) =>
+ killSwitch.abort(new AckDeadlineExtensionException(
+ "Lease management ticker failed; ack deadline extensions have stopped", ex))
+ case _ =>
+ }(ExecutionContext.parasitic)
+
+ val cleanup = () => {
+ ticker.cancel()
+ tracked.clear()
+ }
+
+ Flow[ReceivedMessage]
+ .via(killSwitch.flow)
+ .map { msg =>
+ tracked.put(msg.ackId, System.nanoTime())
+ msg
+ }
+ .watchTermination((_, done: Future[Done]) => {
+ done.onComplete(_ => cleanup())(ExecutionContext.parasitic)
+ NotUsed
+ })
+ }.mapMaterializedValue(_ => NotUsed)
+
+ /**
+ * Create a flow that automatically extends ack deadlines using an adaptive deadline
+ * based on observed processing times, matching Google's official client library behavior.
+ *
+ * Instead of a fixed deadline, the extension deadline is computed from the 99th percentile
+ * (configurable) of actual ack latencies. This means:
+ * - If messages typically process in 2s, deadlines stay short (~10s) for fast redelivery on failure
+ * - If messages take 45s, deadlines adapt upward to avoid unnecessary redelivery
+ *
+ * The same [[AckDeadlineDistribution]] must be passed to the acknowledge/nack operators
+ * so that processing latencies are recorded.
+ *
+ * Usage:
+ * {{{
+ * val dist = AckDeadlineDistribution(initialDeadlineSeconds = 10)
+ *
+ * GooglePubSub.subscribe(request, 1.second, restartSettings)
+ * .via(GooglePubSub.autoExtendAckDeadlines(subscriptionFqrs, 3.seconds, dist))
+ * .mapAsync(4)(processMessage)
+ * .map(msg => AcknowledgeRequest(subscriptionFqrs, Seq(msg.ackId)))
+ * .to(GooglePubSub.acknowledge(parallelism = 1, dist))
+ * }}}
+ *
+ * @param subscription the fully qualified subscription resource string
+ * @param extensionInterval how often to extend deadlines (should be less than the ack deadline)
+ * @param distribution shared distribution that tracks processing times and computes adaptive deadlines
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def autoExtendAckDeadlines(
+ subscription: String,
+ extensionInterval: FiniteDuration,
+ distribution: AckDeadlineDistribution): Flow[ReceivedMessage, ReceivedMessage, NotUsed] =
+ Flow.fromMaterializer { (mat, attr) =>
+ val client = subscriber(mat, attr).client
+ val killSwitch = KillSwitches.shared("autoExtendAckDeadlines")
+
+ val (ticker, tickerDone) = Source
+ .tick(extensionInterval, extensionInterval, ())
+ .mapAsync(1) { _ =>
+ val now = System.nanoTime()
+ // Remove expired entries
+ distribution.deliveryTimes.asScala.foreach { case (ackId, _) =>
+ distribution.isExpired(ackId, now)
+ }
+ val ids = distribution.deliveryTimes.asScala.keys.toSeq
+ if (ids.nonEmpty) {
+ val deadline = distribution.currentDeadlineSeconds
+ client
+ .modifyAckDeadline(ModifyAckDeadlineRequest(subscription, ids, deadline))
+ .map(_ => ())(ExecutionContext.parasitic)
+ } else
+ Future.successful(())
+ }
+ .toMat(Sink.ignore)(Keep.both)
+ .run()(mat)
+
+ tickerDone.onComplete {
+ case scala.util.Failure(ex) =>
+ killSwitch.abort(new AckDeadlineExtensionException(
+ "Lease management ticker failed; ack deadline extensions have stopped", ex))
+ case _ =>
+ }(ExecutionContext.parasitic)
+
+ val cleanup = () => {
+ ticker.cancel()
+ distribution.deliveryTimes.clear()
+ }
+
+ Flow[ReceivedMessage]
+ .via(killSwitch.flow)
+ .map { msg =>
+ distribution.recordDelivery(msg.ackId)
+ msg
+ }
+ .watchTermination((_, done: Future[Done]) => {
+ done.onComplete(_ => cleanup())(ExecutionContext.parasitic)
+ NotUsed
+ })
+ }.mapMaterializedValue(_ => NotUsed)
+
+ /**
+ * Create a flow that modifies the ack deadline for each message using a dynamic function.
+ *
+ * This allows per-message deadline control — for example, setting longer deadlines for
+ * messages that are expected to take longer to process.
+ *
+ * Returning 0 from the function is equivalent to a nack (immediate redelivery).
+ *
+ * @param subscription the fully qualified subscription resource string
+ * @param parallelism controls how many modifications can be in-flight at any given time
+ * @param deadlineFn function that computes the ack deadline in seconds for each message
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def modifyAckDeadlineDynamic(
+ subscription: String,
+ parallelism: Int = 1)(deadlineFn: ReceivedMessage => Int): Flow[ReceivedMessage, ReceivedMessage, NotUsed] =
+ Flow
+ .fromMaterializer { (mat, attr) =>
+ Flow[ReceivedMessage]
+ .mapAsync(parallelism) { msg =>
+ val deadline = deadlineFn(msg)
+ subscriber(mat, attr).client
+ .modifyAckDeadline(
+ ModifyAckDeadlineRequest(subscription, Seq(msg.ackId), deadline))
+ .map(_ => msg)(mat.executionContext)
+ }
+ }
+ .mapMaterializedValue(_ => NotUsed)
+
+ /**
+ * Create a flow that nacks messages by setting the ack deadline to 0,
+ * causing Pub/Sub to immediately redeliver them.
+ *
+ * Accepts [[AcknowledgeRequest]] for consistency with the acknowledge API —
+ * the subscription and ack IDs are reused to build the underlying
+ * `ModifyAckDeadlineRequest` with `ackDeadlineSeconds = 0`.
+ *
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def nackFlow(): Flow[AcknowledgeRequest, AcknowledgeRequest, NotUsed] =
+ Flow
+ .fromMaterializer { (mat, attr) =>
+ Flow[AcknowledgeRequest]
+ .mapAsync(1)(req =>
+ subscriber(mat, attr).client
+ .modifyAckDeadline(
+ ModifyAckDeadlineRequest(req.subscription, req.ackIds, ackDeadlineSeconds = 0))
+ .map(_ => req)(mat.executionContext))
+ }
+ .mapMaterializedValue(_ => NotUsed)
+
+ /**
+ * Create a sink that nacks messages by setting the ack deadline to 0,
+ * causing Pub/Sub to immediately redeliver them.
+ *
+ * The materialized value completes on stream completion.
+ *
+ * @param parallelism controls how many nacks can be in-flight at any given time
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def nack(parallelism: Int): Sink[AcknowledgeRequest, Future[Done]] =
+ Sink
+ .fromMaterializer { (mat, attr) =>
+ Flow[AcknowledgeRequest]
+ .mapAsyncUnordered(parallelism)(req =>
+ subscriber(mat, attr).client
+ .modifyAckDeadline(
+ ModifyAckDeadlineRequest(req.subscription, req.ackIds, ackDeadlineSeconds = 0)))
+ .toMat(Sink.ignore)(Keep.right)
+ }
+ .mapMaterializedValue(_.flatMap(identity)(ExecutionContext.parasitic))
+
/**
* Create a flow that accepts consumed message acknowledgements.
*/
@@ -130,6 +472,215 @@ object GooglePubSub {
.mapMaterializedValue(_.flatMap(identity)(ExecutionContext.parasitic))
}
+ /**
+ * Create a flow that gates messages based on [[FlowControl]], applying backpressure
+ * when the number of outstanding (unacknowledged) messages reaches the limit.
+ *
+ * Pair with a [[FlowControl]]-aware acknowledge or nack operator to release permits.
+ *
+ * @param flowControl shared flow control instance
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def flowControlGate(flowControl: FlowControl): Flow[ReceivedMessage, ReceivedMessage, NotUsed] =
+ Flow.fromGraph(new FlowControlGateStage(flowControl))
+
+ /**
+ * Create a flow that acknowledges messages and releases [[FlowControl]] permits.
+ *
+ * @param flowControl shared flow control instance
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def acknowledgeFlow(flowControl: FlowControl): Flow[AcknowledgeRequest, AcknowledgeRequest, NotUsed] =
+ Flow
+ .fromMaterializer { (mat, attr) =>
+ Flow[AcknowledgeRequest]
+ .mapAsync(1) { req =>
+ subscriber(mat, attr).client
+ .acknowledge(req)
+ .map { _ =>
+ flowControl.release(req.ackIds.size)
+ req
+ }(mat.executionContext)
+ }
+ }
+ .mapMaterializedValue(_ => NotUsed)
+
+ /**
+ * Create a sink that acknowledges messages and releases [[FlowControl]] permits.
+ *
+ * @param parallelism controls how many acknowledgements can be in-flight at any given time
+ * @param flowControl shared flow control instance
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def acknowledge(parallelism: Int, flowControl: FlowControl): Sink[AcknowledgeRequest, Future[Done]] =
+ Sink
+ .fromMaterializer { (mat, attr) =>
+ Flow[AcknowledgeRequest]
+ .mapAsyncUnordered(parallelism) { req =>
+ subscriber(mat, attr).client
+ .acknowledge(req)
+ .map { _ =>
+ flowControl.release(req.ackIds.size)
+ Done
+ }(mat.executionContext)
+ }
+ .toMat(Sink.ignore)(Keep.right)
+ }
+ .mapMaterializedValue(_.flatMap(identity)(ExecutionContext.parasitic))
+
+ /**
+ * Create a flow that nacks messages and releases [[FlowControl]] permits.
+ *
+ * @param flowControl shared flow control instance
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def nackFlow(flowControl: FlowControl): Flow[AcknowledgeRequest, AcknowledgeRequest, NotUsed] =
+ Flow
+ .fromMaterializer { (mat, attr) =>
+ Flow[AcknowledgeRequest]
+ .mapAsync(1) { req =>
+ subscriber(mat, attr).client
+ .modifyAckDeadline(
+ ModifyAckDeadlineRequest(req.subscription, req.ackIds, ackDeadlineSeconds = 0))
+ .map { _ =>
+ flowControl.release(req.ackIds.size)
+ req
+ }(mat.executionContext)
+ }
+ }
+ .mapMaterializedValue(_ => NotUsed)
+
+ /**
+ * Create a sink that nacks messages and releases [[FlowControl]] permits.
+ *
+ * @param parallelism controls how many nacks can be in-flight at any given time
+ * @param flowControl shared flow control instance
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def nack(parallelism: Int, flowControl: FlowControl): Sink[AcknowledgeRequest, Future[Done]] =
+ Sink
+ .fromMaterializer { (mat, attr) =>
+ Flow[AcknowledgeRequest]
+ .mapAsyncUnordered(parallelism) { req =>
+ subscriber(mat, attr).client
+ .modifyAckDeadline(
+ ModifyAckDeadlineRequest(req.subscription, req.ackIds, ackDeadlineSeconds = 0))
+ .map { _ =>
+ flowControl.release(req.ackIds.size)
+ Done
+ }(mat.executionContext)
+ }
+ .toMat(Sink.ignore)(Keep.right)
+ }
+ .mapMaterializedValue(_.flatMap(identity)(ExecutionContext.parasitic))
+
+ /**
+ * Create a flow that acknowledges messages and records processing latencies
+ * in the [[AckDeadlineDistribution]] for adaptive deadline computation.
+ *
+ * @param distribution shared distribution for adaptive deadline tracking
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def acknowledgeFlow(
+ distribution: AckDeadlineDistribution): Flow[AcknowledgeRequest, AcknowledgeRequest, NotUsed] =
+ Flow
+ .fromMaterializer { (mat, attr) =>
+ Flow[AcknowledgeRequest]
+ .mapAsync(1) { req =>
+ subscriber(mat, attr).client
+ .acknowledge(req)
+ .map { _ =>
+ distribution.recordCompletions(req.ackIds)
+ req
+ }(mat.executionContext)
+ }
+ }
+ .mapMaterializedValue(_ => NotUsed)
+
+ /**
+ * Create a sink that acknowledges messages and records processing latencies
+ * in the [[AckDeadlineDistribution]] for adaptive deadline computation.
+ *
+ * @param parallelism controls how many acknowledgements can be in-flight at any given time
+ * @param distribution shared distribution for adaptive deadline tracking
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def acknowledge(parallelism: Int,
+ distribution: AckDeadlineDistribution): Sink[AcknowledgeRequest, Future[Done]] =
+ Sink
+ .fromMaterializer { (mat, attr) =>
+ Flow[AcknowledgeRequest]
+ .mapAsyncUnordered(parallelism) { req =>
+ subscriber(mat, attr).client
+ .acknowledge(req)
+ .map { _ =>
+ distribution.recordCompletions(req.ackIds)
+ Done
+ }(mat.executionContext)
+ }
+ .toMat(Sink.ignore)(Keep.right)
+ }
+ .mapMaterializedValue(_.flatMap(identity)(ExecutionContext.parasitic))
+
+ /**
+ * Create a flow that nacks messages and records processing latencies
+ * in the [[AckDeadlineDistribution]] for adaptive deadline computation.
+ *
+ * @param distribution shared distribution for adaptive deadline tracking
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def nackFlow(
+ distribution: AckDeadlineDistribution): Flow[AcknowledgeRequest, AcknowledgeRequest, NotUsed] =
+ Flow
+ .fromMaterializer { (mat, attr) =>
+ Flow[AcknowledgeRequest]
+ .mapAsync(1) { req =>
+ subscriber(mat, attr).client
+ .modifyAckDeadline(
+ ModifyAckDeadlineRequest(req.subscription, req.ackIds, ackDeadlineSeconds = 0))
+ .map { _ =>
+ distribution.recordCompletions(req.ackIds)
+ req
+ }(mat.executionContext)
+ }
+ }
+ .mapMaterializedValue(_ => NotUsed)
+
+ /**
+ * Create a sink that nacks messages and records processing latencies
+ * in the [[AckDeadlineDistribution]] for adaptive deadline computation.
+ *
+ * @param parallelism controls how many nacks can be in-flight at any given time
+ * @param distribution shared distribution for adaptive deadline tracking
+ * @since 2.0.0
+ */
+ @ApiMayChange
+ def nack(parallelism: Int,
+ distribution: AckDeadlineDistribution): Sink[AcknowledgeRequest, Future[Done]] =
+ Sink
+ .fromMaterializer { (mat, attr) =>
+ Flow[AcknowledgeRequest]
+ .mapAsyncUnordered(parallelism) { req =>
+ subscriber(mat, attr).client
+ .modifyAckDeadline(
+ ModifyAckDeadlineRequest(req.subscription, req.ackIds, ackDeadlineSeconds = 0))
+ .map { _ =>
+ distribution.recordCompletions(req.ackIds)
+ Done
+ }(mat.executionContext)
+ }
+ .toMat(Sink.ignore)(Keep.right)
+ }
+ .mapMaterializedValue(_.flatMap(identity)(ExecutionContext.parasitic))
+
private def publisher(mat: Materializer, attr: Attributes) =
attr
.get[PubSubAttributes.Publisher]
diff --git a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/scaladsl/GrpcSubscriber.scala b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/scaladsl/GrpcSubscriber.scala
index 0d70d861d..df3d3865e 100644
--- a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/scaladsl/GrpcSubscriber.scala
+++ b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/scaladsl/GrpcSubscriber.scala
@@ -31,28 +31,24 @@ import com.google.pubsub.v1.pubsub.{ SubscriberClient => ScalaSubscriberClient }
/**
* Holds the gRPC scala subscriber client instance.
*/
-final class GrpcSubscriber private (settings: PubSubSettings, googleSettings: GoogleSettings, sys: ActorSystem) {
-
- @ApiMayChange
- final val client =
- ScalaSubscriberClient(PekkoGrpcSettings.fromPubSubSettings(settings, googleSettings)(sys))(sys)
-
- sys.registerOnTermination(client.close())
-}
+final class GrpcSubscriber private[grpc] (@ApiMayChange final val client: ScalaSubscriberClient)
object GrpcSubscriber {
def apply(settings: PubSubSettings,
googleSettings: GoogleSettings)(implicit sys: ClassicActorSystemProvider): GrpcSubscriber =
- new GrpcSubscriber(settings, googleSettings, sys.classicSystem)
+ apply(settings, googleSettings, sys.classicSystem)
- def apply(settings: PubSubSettings, googleSettings: GoogleSettings, sys: ActorSystem): GrpcSubscriber =
- new GrpcSubscriber(settings, googleSettings, sys)
+ def apply(settings: PubSubSettings, googleSettings: GoogleSettings, sys: ActorSystem): GrpcSubscriber = {
+ val c = ScalaSubscriberClient(PekkoGrpcSettings.fromPubSubSettings(settings, googleSettings)(sys))(sys)
+ sys.registerOnTermination(c.close())
+ new GrpcSubscriber(c)
+ }
def apply(settings: PubSubSettings)(implicit sys: ClassicActorSystemProvider): GrpcSubscriber =
- new GrpcSubscriber(settings, GoogleSettings(), sys.classicSystem)
+ apply(settings, GoogleSettings())
def apply(settings: PubSubSettings, sys: ActorSystem): GrpcSubscriber =
- new GrpcSubscriber(settings, GoogleSettings(sys), sys)
+ apply(settings, GoogleSettings(sys), sys)
def apply()(implicit sys: ClassicActorSystemProvider): GrpcSubscriber =
apply(PubSubSettings(sys))
diff --git a/google-cloud-pub-sub-grpc/src/test/java/docs/javadsl/IntegrationTest.java b/google-cloud-pub-sub-grpc/src/test/java/docs/javadsl/IntegrationTest.java
index 8a172fecd..c92199b01 100644
--- a/google-cloud-pub-sub-grpc/src/test/java/docs/javadsl/IntegrationTest.java
+++ b/google-cloud-pub-sub-grpc/src/test/java/docs/javadsl/IntegrationTest.java
@@ -217,6 +217,263 @@ public class IntegrationTest {
// #acknowledge
}
+ @Test
+ public void shouldModifyAckDeadline() {
+ final String projectId = "pekko-connectors";
+ final String subscription = "simpleSubscription";
+ final String subscriptionFqrs =
+ "projects/" + projectId + "/subscriptions/" + subscription;
+
+ final StreamingPullRequest request =
+ StreamingPullRequest.newBuilder()
+ .setSubscription(subscriptionFqrs)
+ .setStreamAckDeadlineSeconds(10)
+ .build();
+
+ final Duration pollInterval = Duration.ofSeconds(1);
+ final Source<ReceivedMessage, CompletableFuture<Cancellable>> subscriptionSource =
+ GooglePubSub.subscribe(request, pollInterval);
+
+ // #modify-ack-deadline
+ subscriptionSource
+ .mapAsync(
+ 1,
+ receivedMessage -> {
+ ModifyAckDeadlineRequest modifyRequest =
+ ModifyAckDeadlineRequest.newBuilder()
+ .setSubscription(subscriptionFqrs)
+ .addAckIds(receivedMessage.getAckId())
+ .setAckDeadlineSeconds(30)
+ .build();
+ return Source.single(modifyRequest)
+ .via(GooglePubSub.modifyAckDeadlineFlow())
+ .runWith(Sink.head(), system)
+ .thenApply(ignore -> receivedMessage);
+ })
+ .map(
+ msg ->
+ AcknowledgeRequest.newBuilder()
+ .setSubscription(subscriptionFqrs)
+ .addAckIds(msg.getAckId())
+ .build())
+ .to(GooglePubSub.acknowledge(1));
+ // #modify-ack-deadline
+ }
+
+ @Test
+ public void shouldAutoExtendAckDeadlines() {
+ final String projectId = "pekko-connectors";
+ final String subscription = "simpleSubscription";
+ final String subscriptionFqrs =
+ "projects/" + projectId + "/subscriptions/" + subscription;
+
+ final StreamingPullRequest request =
+ StreamingPullRequest.newBuilder()
+ .setSubscription(subscriptionFqrs)
+ .setStreamAckDeadlineSeconds(10)
+ .build();
+
+ final Duration pollInterval = Duration.ofSeconds(1);
+ final Source<ReceivedMessage, CompletableFuture<Cancellable>> subscriptionSource =
+ GooglePubSub.subscribe(request, pollInterval);
+
+ // #subscribe-auto-extend
+ subscriptionSource
+ .via(GooglePubSub.autoExtendAckDeadlines(subscriptionFqrs, Duration.ofSeconds(3), 30))
+ .map(
+ msg ->
+ AcknowledgeRequest.newBuilder()
+ .setSubscription(subscriptionFqrs)
+ .addAckIds(msg.getAckId())
+ .build())
+ .to(GooglePubSub.acknowledge(1));
+ // #subscribe-auto-extend
+ }
+
+ @Test
+ public void shouldNackAndRedeliver()
+ throws InterruptedException, ExecutionException, TimeoutException {
+ final String projectId = "pekko-connectors";
+ final String topic = "simpleTopic";
+ final String subscription = "simpleSubscription";
+ final String subscriptionFqrs = "projects/" + projectId + "/subscriptions/" + subscription;
+
+ // publish a message
+ final ByteString msg = ByteString.copyFromUtf8("nack-java-" + System.nanoTime());
+ final PublishRequest publishRequest =
+ PublishRequest.newBuilder()
+ .setTopic("projects/" + projectId + "/topics/" + topic)
+ .addMessages(PubsubMessage.newBuilder().setData(msg).build())
+ .build();
+ Source.single(publishRequest).via(GooglePubSub.publish(1)).runWith(Sink.head(), system)
+ .toCompletableFuture().get(5, TimeUnit.SECONDS);
+
+ final StreamingPullRequest request =
+ StreamingPullRequest.newBuilder()
+ .setSubscription(subscriptionFqrs)
+ .setStreamAckDeadlineSeconds(10)
+ .build();
+
+ // receive and nack — filter for our specific message to avoid test pollution
+ ReceivedMessage received = GooglePubSub.subscribe(request, Duration.ofSeconds(1))
+ .filter(m -> m.getMessage().getData().equals(msg))
+ .runWith(Sink.head(), system)
+ .toCompletableFuture().get(10, TimeUnit.SECONDS);
+
+ Source.single(
+ AcknowledgeRequest.newBuilder()
+ .setSubscription(subscriptionFqrs)
+ .addAckIds(received.getAckId())
+ .build())
+ .via(GooglePubSub.nackFlow())
+ .runWith(Sink.head(), system)
+ .toCompletableFuture().get(5, TimeUnit.SECONDS);
+
+ // should be redelivered
+ ReceivedMessage redelivered = GooglePubSub.subscribe(request, Duration.ofSeconds(1))
+ .filter(m -> m.getMessage().getData().equals(msg))
+ .runWith(Sink.head(), system)
+ .toCompletableFuture().get(15, TimeUnit.SECONDS);
+
+ assertEquals("nacked message should be redelivered", msg, redelivered.getMessage().getData());
+ }
+
+ @Test
+ public void shouldModifyAckDeadlineDynamic()
+ throws InterruptedException, ExecutionException, TimeoutException {
+ final String projectId = "pekko-connectors";
+ final String topic = "simpleTopic";
+ final String subscription = "simpleSubscription";
+ final String subscriptionFqrs = "projects/" + projectId + "/subscriptions/" + subscription;
+
+ // publish a message
+ final ByteString msg = ByteString.copyFromUtf8("dynamic-java-" + System.nanoTime());
+ final PublishRequest publishRequest =
+ PublishRequest.newBuilder()
+ .setTopic("projects/" + projectId + "/topics/" + topic)
+ .addMessages(PubsubMessage.newBuilder().setData(msg).build())
+ .build();
+ Source.single(publishRequest).via(GooglePubSub.publish(1)).runWith(Sink.head(), system)
+ .toCompletableFuture().get(5, TimeUnit.SECONDS);
+
+ final StreamingPullRequest request =
+ StreamingPullRequest.newBuilder()
+ .setSubscription(subscriptionFqrs)
+ .setStreamAckDeadlineSeconds(10)
+ .build();
+
+ // subscribe, dynamically set deadline, then ack
+ ReceivedMessage result = GooglePubSub.subscribe(request, Duration.ofSeconds(1))
+ .take(1)
+ .via(GooglePubSub.modifyAckDeadlineDynamic(subscriptionFqrs, 1, m -> 30))
+ .runWith(Sink.head(), system)
+ .toCompletableFuture().get(5, TimeUnit.SECONDS);
+
+ // ack the message
+ Source.single(
+ AcknowledgeRequest.newBuilder()
+ .setSubscription(subscriptionFqrs)
+ .addAckIds(result.getAckId())
+ .build())
+ .runWith(GooglePubSub.acknowledge(1), system)
+ .toCompletableFuture().get(5, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void shouldFlowControl()
+ throws InterruptedException, ExecutionException, TimeoutException {
+ final String projectId = "pekko-connectors";
+ final String topic = "simpleTopic";
+ final String subscription = "simpleSubscription";
+ final String subscriptionFqrs = "projects/" + projectId + "/subscriptions/" + subscription;
+
+ // publish several messages
+ final String prefix = "fc-java-" + System.nanoTime();
+ PublishRequest.Builder publishBuilder = PublishRequest.newBuilder()
+ .setTopic("projects/" + projectId + "/topics/" + topic);
+ for (int i = 1; i <= 3; i++) {
+ publishBuilder.addMessages(
+ PubsubMessage.newBuilder()
+ .setData(ByteString.copyFromUtf8(prefix + "-" + i))
+ .build());
+ }
+ Source.single(publishBuilder.build()).via(GooglePubSub.publish(1))
+ .runWith(Sink.head(), system)
+ .toCompletableFuture().get(5, TimeUnit.SECONDS);
+
+ final StreamingPullRequest request =
+ StreamingPullRequest.newBuilder()
+ .setSubscription(subscriptionFqrs)
+ .setStreamAckDeadlineSeconds(10)
+ .build();
+
+ final org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.FlowControl fc =
+ org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.FlowControl.create(2);
+
+ List<AcknowledgeRequest> result = GooglePubSub.subscribe(request, Duration.ofSeconds(1))
+ .via(GooglePubSub.flowControlGate(fc))
+ .take(3)
+ .map(m -> AcknowledgeRequest.newBuilder()
+ .setSubscription(subscriptionFqrs)
+ .addAckIds(m.getAckId())
+ .build())
+ .via(GooglePubSub.acknowledgeFlow(fc))
+ .runWith(Sink.seq(), system)
+ .toCompletableFuture().get(15, TimeUnit.SECONDS);
+
+ assertEquals("should process all 3 messages", 3, result.size());
+ }
+
+ @Test
+ public void shouldAdaptiveDeadline()
+ throws InterruptedException, ExecutionException, TimeoutException {
+ final String projectId = "pekko-connectors";
+ final String topic = "simpleTopic";
+ final String subscription = "simpleSubscription";
+ final String subscriptionFqrs = "projects/" + projectId + "/subscriptions/" + subscription;
+
+ final org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.AckDeadlineDistribution dist =
+ org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.AckDeadlineDistribution.create();
+
+ // initially should use default
+ assertEquals("initial deadline", 60, dist.currentDeadlineSeconds());
+
+ // publish messages
+ final String prefix = "adaptive-java-" + System.nanoTime();
+ PublishRequest.Builder publishBuilder = PublishRequest.newBuilder()
+ .setTopic("projects/" + projectId + "/topics/" + topic);
+ for (int i = 1; i <= 2; i++) {
+ publishBuilder.addMessages(
+ PubsubMessage.newBuilder()
+ .setData(ByteString.copyFromUtf8(prefix + "-" + i))
+ .build());
+ }
+ Source.single(publishBuilder.build()).via(GooglePubSub.publish(1))
+ .runWith(Sink.head(), system)
+ .toCompletableFuture().get(5, TimeUnit.SECONDS);
+
+ final StreamingPullRequest request =
+ StreamingPullRequest.newBuilder()
+ .setSubscription(subscriptionFqrs)
+ .setStreamAckDeadlineSeconds(10)
+ .build();
+
+ List<AcknowledgeRequest> result = GooglePubSub.subscribe(request, Duration.ofSeconds(1))
+ .via(GooglePubSub.autoExtendAckDeadlines(subscriptionFqrs, Duration.ofSeconds(3), dist))
+ .take(2)
+ .map(m -> AcknowledgeRequest.newBuilder()
+ .setSubscription(subscriptionFqrs)
+ .addAckIds(m.getAckId())
+ .build())
+ .via(GooglePubSub.acknowledgeFlow(dist))
+ .runWith(Sink.seq(), system)
+ .toCompletableFuture().get(15, TimeUnit.SECONDS);
+
+ assertEquals("should process all messages", 2, result.size());
+ // adaptive deadline should be at least the minimum
+ assertTrue("deadline should be >= 10", dist.currentDeadlineSeconds() >= 10);
+ }
+
@Test
public void customPublisher() {
// #attributes
diff --git a/google-cloud-pub-sub-grpc/src/test/java/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/AutoExtendAckDeadlinesTest.java b/google-cloud-pub-sub-grpc/src/test/java/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/AutoExtendAckDeadlinesTest.java
new file mode 100644
index 000000000..3588e25fd
--- /dev/null
+++ b/google-cloud-pub-sub-grpc/src/test/java/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/AutoExtendAckDeadlinesTest.java
@@ -0,0 +1,294 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc;
+
+import static org.junit.Assert.*;
+
+import com.google.protobuf.ByteString;
+import com.google.pubsub.v1.*;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.*;
+import org.apache.pekko.Done;
+import org.apache.pekko.NotUsed;
+import org.apache.pekko.actor.ActorSystem;
+import org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.javadsl.GooglePubSub;
+import org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.javadsl.GrpcSubscriber;
+import org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.javadsl.PubSubAttributes;
+import org.apache.pekko.stream.javadsl.Sink;
+import org.apache.pekko.stream.javadsl.Source;
+import org.junit.AfterClass;
+import org.junit.Test;
+
+public class AutoExtendAckDeadlinesTest {
+
+ private static final ActorSystem system = ActorSystem.create("AutoExtendAckDeadlinesTest");
+ private static final String subscription = "projects/test/subscriptions/test-sub";
+
+ private static ReceivedMessage makeMsg(String id) {
+ return ReceivedMessage.newBuilder()
+ .setAckId(id)
+ .setMessage(
+ PubsubMessage.newBuilder().setData(ByteString.copyFromUtf8("test-" + id)).build())
+ .build();
+ }
+
+ /** Stub that fails modifyAckDeadline calls. */
+ static class FailingSubscriberClient extends SubscriberClient {
+ @Override
+ public CompletionStage<com.google.protobuf.Empty> modifyAckDeadline(
+ ModifyAckDeadlineRequest in) {
+ CompletableFuture<com.google.protobuf.Empty> f = new CompletableFuture<>();
+ f.completeExceptionally(new RuntimeException("simulated modifyAckDeadline failure"));
+ return f;
+ }
+
+ @Override
+ public CompletionStage<com.google.protobuf.Empty> acknowledge(AcknowledgeRequest in) {
+ return CompletableFuture.completedFuture(com.google.protobuf.Empty.getDefaultInstance());
+ }
+
+ @Override
+ public CompletionStage<Subscription> createSubscription(Subscription in) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletionStage<Subscription> getSubscription(GetSubscriptionRequest in) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletionStage<Subscription> updateSubscription(UpdateSubscriptionRequest in) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletionStage<ListSubscriptionsResponse> listSubscriptions(
+ ListSubscriptionsRequest in) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletionStage<com.google.protobuf.Empty> deleteSubscription(
+ DeleteSubscriptionRequest in) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletionStage<PullResponse> pull(PullRequest in) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Source<StreamingPullResponse, NotUsed> streamingPull(
+ Source<StreamingPullRequest, NotUsed> in) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletionStage<com.google.protobuf.Empty> modifyPushConfig(ModifyPushConfigRequest in) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletionStage<Snapshot> getSnapshot(GetSnapshotRequest in) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletionStage<ListSnapshotsResponse> listSnapshots(ListSnapshotsRequest in) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletionStage<Snapshot> createSnapshot(CreateSnapshotRequest in) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletionStage<Snapshot> updateSnapshot(UpdateSnapshotRequest in) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletionStage<com.google.protobuf.Empty> deleteSnapshot(DeleteSnapshotRequest in) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletionStage<SeekResponse> seek(SeekRequest in) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletionStage<Done> close() {
+ return CompletableFuture.completedFuture(Done.done());
+ }
+
+ @Override
+ public CompletionStage<Done> closed() {
+ return new CompletableFuture<>(); // never completes
+ }
+ }
+
+ /** Stub that succeeds for modifyAckDeadline calls. */
+ static class SucceedingSubscriberClient extends FailingSubscriberClient {
+ @Override
+ public CompletionStage<com.google.protobuf.Empty> modifyAckDeadline(
+ ModifyAckDeadlineRequest in) {
+ return CompletableFuture.completedFuture(com.google.protobuf.Empty.getDefaultInstance());
+ }
+ }
+
+ @Test
+ public void fixedDeadlineShouldFailStreamWhenTickerFails() throws Exception {
+ GrpcSubscriber testSubscriber = new GrpcSubscriber(new FailingSubscriberClient());
+
+ List<ReceivedMessage> msgs = Arrays.asList(makeMsg("1"), makeMsg("2"), makeMsg("3"));
+
+ CompletionStage<List<ReceivedMessage>> result =
+ Source.from(msgs)
+ .throttle(1, Duration.ofMillis(500))
+ .via(GooglePubSub.autoExtendAckDeadlines(subscription, Duration.ofMillis(200), 30))
+ .withAttributes(PubSubAttributes.subscriber(testSubscriber))
+ .runWith(Sink.seq(), system);
+
+ try {
+ result.toCompletableFuture().get(10, TimeUnit.SECONDS);
+ fail("Expected AckDeadlineExtensionException");
+ } catch (ExecutionException e) {
+ assertTrue(
+ "Expected AckDeadlineExtensionException but got: " + e.getCause().getClass(),
+ e.getCause() instanceof AckDeadlineExtensionException);
+ assertTrue(e.getCause().getMessage().contains("ticker failed"));
+ }
+ }
+
+ @Test
+ public void fixedDeadlineShouldFailIdleStreamWhenTickerDies() throws Exception {
+ GrpcSubscriber testSubscriber = new GrpcSubscriber(new FailingSubscriberClient());
+
+ // Emit one message then idle — the ticker should abort the stream
+ // even though no further elements arrive.
+ CompletionStage<List<ReceivedMessage>> result =
+ Source.single(makeMsg("only"))
+ .concat(Source.maybe())
+ .via(GooglePubSub.autoExtendAckDeadlines(subscription, Duration.ofMillis(200), 30))
+ .withAttributes(PubSubAttributes.subscriber(testSubscriber))
+ .runWith(Sink.seq(), system);
+
+ try {
+ result.toCompletableFuture().get(10, TimeUnit.SECONDS);
+ fail("Expected AckDeadlineExtensionException");
+ } catch (ExecutionException e) {
+ assertTrue(
+ "Expected AckDeadlineExtensionException but got: " + e.getCause().getClass(),
+ e.getCause() instanceof AckDeadlineExtensionException);
+ assertTrue(e.getCause().getMessage().contains("ticker failed"));
+ }
+ }
+
+ @Test
+ public void fixedDeadlineShouldPassMessagesThroughWhenHealthy() throws Exception {
+ GrpcSubscriber testSubscriber = new GrpcSubscriber(new SucceedingSubscriberClient());
+
+ List<ReceivedMessage> msgs = Arrays.asList(makeMsg("1"), makeMsg("2"), makeMsg("3"));
+
+ List<ReceivedMessage> result =
+ Source.from(msgs)
+ .via(GooglePubSub.autoExtendAckDeadlines(subscription, Duration.ofSeconds(1), 30))
+ .withAttributes(PubSubAttributes.subscriber(testSubscriber))
+ .runWith(Sink.seq(), system)
+ .toCompletableFuture()
+ .get(10, TimeUnit.SECONDS);
+
+ assertEquals(3, result.size());
+ assertEquals("1", result.get(0).getAckId());
+ assertEquals("2", result.get(1).getAckId());
+ assertEquals("3", result.get(2).getAckId());
+ }
+
+ @Test
+ public void adaptiveShouldFailStreamWhenTickerFails() throws Exception {
+ GrpcSubscriber testSubscriber = new GrpcSubscriber(new FailingSubscriberClient());
+ AckDeadlineDistribution dist = AckDeadlineDistribution.create();
+
+ List<ReceivedMessage> msgs = Arrays.asList(makeMsg("1"), makeMsg("2"), makeMsg("3"));
+
+ CompletionStage<List<ReceivedMessage>> result =
+ Source.from(msgs)
+ .throttle(1, Duration.ofMillis(500))
+ .via(GooglePubSub.autoExtendAckDeadlines(subscription, Duration.ofMillis(200), dist))
+ .withAttributes(PubSubAttributes.subscriber(testSubscriber))
+ .runWith(Sink.seq(), system);
+
+ try {
+ result.toCompletableFuture().get(10, TimeUnit.SECONDS);
+ fail("Expected AckDeadlineExtensionException");
+ } catch (ExecutionException e) {
+ assertTrue(e.getCause() instanceof AckDeadlineExtensionException);
+ assertTrue(e.getCause().getMessage().contains("ticker failed"));
+ }
+ }
+
+ @Test
+ public void adaptiveShouldFailIdleStreamWhenTickerDies() throws Exception {
+ GrpcSubscriber testSubscriber = new GrpcSubscriber(new FailingSubscriberClient());
+ AckDeadlineDistribution dist = AckDeadlineDistribution.create();
+
+ CompletionStage<List<ReceivedMessage>> result =
+ Source.single(makeMsg("only"))
+ .concat(Source.maybe())
+ .via(GooglePubSub.autoExtendAckDeadlines(subscription, Duration.ofMillis(200), dist))
+ .withAttributes(PubSubAttributes.subscriber(testSubscriber))
+ .runWith(Sink.seq(), system);
+
+ try {
+ result.toCompletableFuture().get(10, TimeUnit.SECONDS);
+ fail("Expected AckDeadlineExtensionException");
+ } catch (ExecutionException e) {
+ assertTrue(e.getCause() instanceof AckDeadlineExtensionException);
+ assertTrue(e.getCause().getMessage().contains("ticker failed"));
+ }
+ }
+
+ @Test
+ public void adaptiveShouldPassMessagesThroughWhenHealthy() throws Exception {
+ GrpcSubscriber testSubscriber = new GrpcSubscriber(new SucceedingSubscriberClient());
+ AckDeadlineDistribution dist = AckDeadlineDistribution.create();
+
+ List<ReceivedMessage> msgs = Arrays.asList(makeMsg("1"), makeMsg("2"), makeMsg("3"));
+
+ List<ReceivedMessage> result =
+ Source.from(msgs)
+ .via(GooglePubSub.autoExtendAckDeadlines(subscription, Duration.ofSeconds(1), dist))
+ .withAttributes(PubSubAttributes.subscriber(testSubscriber))
+ .runWith(Sink.seq(), system)
+ .toCompletableFuture()
+ .get(10, TimeUnit.SECONDS);
+
+ assertEquals(3, result.size());
+ }
+
+ @AfterClass
+ public static void tearDown() {
+ system.terminate();
+ }
+}
diff --git a/google-cloud-pub-sub-grpc/src/test/scala/docs/scaladsl/ExampleApp.scala b/google-cloud-pub-sub-grpc/src/test/scala/docs/scaladsl/ExampleApp.scala
index d76e43398..32417c6dc 100644
--- a/google-cloud-pub-sub-grpc/src/test/scala/docs/scaladsl/ExampleApp.scala
+++ b/google-cloud-pub-sub-grpc/src/test/scala/docs/scaladsl/ExampleApp.scala
@@ -15,11 +15,18 @@ package docs.scaladsl
import java.util.logging.{ Level, Logger }
import org.apache.pekko
import pekko.actor.{ ActorSystem, Cancellable }
-import pekko.stream.DelayOverflowStrategy
+import pekko.stream.{ DelayOverflowStrategy, RestartSettings }
+import pekko.stream.connectors.googlecloud.pubsub.grpc.FlowControl
import pekko.stream.connectors.googlecloud.pubsub.grpc.scaladsl.GooglePubSub
import pekko.stream.scaladsl.{ Sink, Source }
import com.google.protobuf.ByteString
-import com.google.pubsub.v1.pubsub.{ PublishRequest, PubsubMessage, StreamingPullRequest }
+import com.google.pubsub.v1.pubsub.{
+ AcknowledgeRequest,
+ PublishRequest,
+ PubsubMessage,
+ ReceivedMessage,
+ StreamingPullRequest
+}
import com.typesafe.config.ConfigFactory
import scala.annotation.nowarn
@@ -42,10 +49,15 @@ object ExampleApp {
import sys.dispatcher
val result = args.toList match {
- case "publish-single" :: rest => publishSingle(rest)
- case "publish-stream" :: rest => publishStream(rest)
- case "subscribe" :: rest => subscribeStream(rest)
- case other => Future.failed(new Error(s"unknown arguments: $other"))
+ case "publish-single" :: rest => publishSingle(rest)
+ case "publish-stream" :: rest => publishStream(rest)
+ case "subscribe" :: rest => subscribeStream(rest)
+ case "subscribe-auto-extend" :: rest => subscribeAutoExtend(rest)
+ case "subscribe-flow-control" :: rest => subscribeFlowControl(rest)
+ case "subscribe-nack-retry" :: rest => subscribeNackRetry(rest)
+ case "subscribe-dynamic-deadline" :: rest => subscribeDynamicDeadline(rest)
+ case "subscribe-adaptive" :: rest => subscribeAdaptive(rest)
+ case other => Future.failed(new Error(s"unknown arguments: $other"))
}
result.onComplete { res =>
@@ -85,7 +97,7 @@ object ExampleApp {
.map(publish(projectId, topic)(_))
.via(GooglePubSub.publish(parallelism = 1))
.to(Sink.ignore)
- .mapMaterializedValue(Future.successful(_))
+ .mapMaterializedValue(Future.successful)
.run()
}
@@ -98,6 +110,173 @@ object ExampleApp {
.run()
}
+ private def subscribeAutoExtend(args: List[String])(implicit system: ActorSystem) = {
+ val projectId :: sub :: Nil = args: @nowarn("msg=match may not be exhaustive")
+ val subscriptionFqrs = subFqrs(projectId, sub)
+
+ val restartSettings = RestartSettings(
+ minBackoff = 100.millis,
+ maxBackoff = 10.seconds,
+ randomFactor = 0.2)
+
+ GooglePubSub
+ .subscribe(subscribe(projectId, sub), 1.second, restartSettings)
+ .via(GooglePubSub.autoExtendAckDeadlines(subscriptionFqrs, 3.seconds, 30))
+ .map { msg =>
+ println(msg)
+ AcknowledgeRequest(subscriptionFqrs, Seq(msg.ackId))
+ }
+ .to(GooglePubSub.acknowledge(parallelism = 1))
+ .mapMaterializedValue(Future.successful(_))
+ .run()
+ }
+
+ /**
+ * Subscribe with flow control — limits the number of in-flight messages to prevent
+ * memory exhaustion during slow processing.
+ *
+ * Usage: subscribe-flow-control <projectId> <subscription>
+ */
+ private def subscribeFlowControl(args: List[String])(implicit system: ActorSystem) = {
+ val projectId :: sub :: Nil = args: @nowarn("msg=match may not be exhaustive")
+ val subscriptionFqrs = subFqrs(projectId, sub)
+
+ val restartSettings = RestartSettings(
+ minBackoff = 100.millis,
+ maxBackoff = 10.seconds,
+ randomFactor = 0.2)
+
+ val fc = FlowControl(maxOutstandingMessages = 100)
+
+ GooglePubSub
+ .subscribe(subscribe(projectId, sub), 1.second, restartSettings)
+ .via(GooglePubSub.flowControlGate(fc))
+ .via(GooglePubSub.autoExtendAckDeadlines(subscriptionFqrs, 8.seconds, 30))
+ .map { msg =>
+ println(s"[outstanding=${fc.outstandingCount}] ${msg.message.map(_.data.toStringUtf8).getOrElse("")}")
+ AcknowledgeRequest(subscriptionFqrs, Seq(msg.ackId))
+ }
+ .to(GooglePubSub.acknowledge(parallelism = 1, fc))
+ .mapMaterializedValue(Future.successful(_))
+ .run()
+ }
+
+ /**
+ * Subscribe and nack messages that fail processing, causing immediate redelivery.
+ * Messages that succeed are acknowledged normally.
+ *
+ * Usage: subscribe-nack-retry <projectId> <subscription>
+ */
+ private def subscribeNackRetry(args: List[String])(implicit system: ActorSystem) = {
+ import system.dispatcher
+
+ val projectId :: sub :: Nil = args: @nowarn("msg=match may not be exhaustive")
+ val subscriptionFqrs = subFqrs(projectId, sub)
+
+ val restartSettings = RestartSettings(
+ minBackoff = 100.millis,
+ maxBackoff = 10.seconds,
+ randomFactor = 0.2)
+
+ GooglePubSub
+ .subscribe(subscribe(projectId, sub), 1.second, restartSettings)
+ .via(GooglePubSub.autoExtendAckDeadlines(subscriptionFqrs, 8.seconds, 30))
+ .mapAsync(4) { msg =>
+ processMessage(msg).map(Right(_)).recover { case ex =>
+ println(s"Processing failed: ${ex.getMessage}, nacking for redelivery")
+ Left(msg)
+ }
+ }
+ .to(Sink.foreach {
+ case Right(msg) =>
+ Source.single(AcknowledgeRequest(subscriptionFqrs, Seq(msg.ackId)))
+ .runWith(GooglePubSub.acknowledge(parallelism = 1))
+ case Left(msg) =>
+ Source.single(AcknowledgeRequest(subscriptionFqrs, Seq(msg.ackId)))
+ .runWith(GooglePubSub.nack(parallelism = 1))
+ })
+ .mapMaterializedValue(Future.successful(_))
+ .run()
+ }
+
+ /**
+ * Subscribe with per-message dynamic deadline based on message size.
+ * Larger messages get longer deadlines.
+ *
+ * Usage: subscribe-dynamic-deadline <projectId> <subscription>
+ */
+ private def subscribeDynamicDeadline(args: List[String])(implicit system: ActorSystem) = {
+ val projectId :: sub :: Nil = args: @nowarn("msg=match may not be exhaustive")
+ val subscriptionFqrs = subFqrs(projectId, sub)
+
+ val restartSettings = RestartSettings(
+ minBackoff = 100.millis,
+ maxBackoff = 10.seconds,
+ randomFactor = 0.2)
+
+ GooglePubSub
+ .subscribe(subscribe(projectId, sub), 1.second, restartSettings)
+ .via(GooglePubSub.modifyAckDeadlineDynamic(subscriptionFqrs) { msg =>
+ val sizeBytes = msg.message.map(_.data.size()).getOrElse(0)
+ if (sizeBytes > 1024 * 1024) 600 // large messages: 10 min
+ else if (sizeBytes > 1024) 120 // medium messages: 2 min
+ else 30 // small messages: 30s
+ })
+ .map { msg =>
+ val deadline = msg.message.map(_.data.size()).getOrElse(0) match {
+ case s if s > 1024 * 1024 => "10min"
+ case s if s > 1024 => "2min"
+ case _ => "30s"
+ }
+ println(s"[deadline=$deadline] ${msg.message.map(_.data.toStringUtf8.take(80)).getOrElse("")}")
+ AcknowledgeRequest(subscriptionFqrs, Seq(msg.ackId))
+ }
+ .to(GooglePubSub.acknowledge(parallelism = 1))
+ .mapMaterializedValue(Future.successful(_))
+ .run()
+ }
+
+ /**
+ * Subscribe with adaptive deadline estimation — the ack extension deadline
+ * automatically adapts to the 99th percentile of observed processing times,
+ * matching the behavior of Google's official client library.
+ *
+ * Usage: subscribe-adaptive <projectId> <subscription>
+ */
+ private def subscribeAdaptive(args: List[String])(implicit system: ActorSystem) = {
+ import org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.AckDeadlineDistribution
+
+ val projectId :: sub :: Nil = args: @nowarn("msg=match may not be exhaustive")
+ val subscriptionFqrs = subFqrs(projectId, sub)
+
+ val restartSettings = RestartSettings(
+ minBackoff = 100.millis,
+ maxBackoff = 10.seconds,
+ randomFactor = 0.2)
+
+ val dist = AckDeadlineDistribution(initialDeadlineSeconds = 10)
+
+ GooglePubSub
+ .subscribe(subscribe(projectId, sub), 1.second, restartSettings)
+ .via(GooglePubSub.autoExtendAckDeadlines(subscriptionFqrs, 3.seconds, dist))
+ .map { msg =>
+ println(s"[adaptive-deadline=${dist.currentDeadlineSeconds}s] " +
+ s"${msg.message.map(_.data.toStringUtf8).getOrElse("")}")
+ AcknowledgeRequest(subscriptionFqrs, Seq(msg.ackId))
+ }
+ .to(GooglePubSub.acknowledge(parallelism = 1, dist))
+ .mapMaterializedValue(Future.successful(_))
+ .run()
+ }
+
+ /** Simulate message processing that can fail. */
+ private def processMessage(msg: ReceivedMessage): Future[ReceivedMessage] = {
+ val data = msg.message.map(_.data.toStringUtf8).getOrElse("")
+ println(s"Processing: $data")
+ if (data.contains("FAIL")) Future.failed(new RuntimeException(s"Failed to process: $data"))
+ else Future.successful(msg)
+ }
+
private def publish(projectId: String, topic: String)(msg: String) =
PublishRequest(topicFqrs(projectId, topic), Seq(PubsubMessage(ByteString.copyFromUtf8(msg))))
diff --git a/google-cloud-pub-sub-grpc/src/test/scala/docs/scaladsl/IntegrationSpec.scala b/google-cloud-pub-sub-grpc/src/test/scala/docs/scaladsl/IntegrationSpec.scala
index 3e088cecc..25e12fc0b 100644
--- a/google-cloud-pub-sub-grpc/src/test/scala/docs/scaladsl/IntegrationSpec.scala
+++ b/google-cloud-pub-sub-grpc/src/test/scala/docs/scaladsl/IntegrationSpec.scala
@@ -220,6 +220,127 @@ class IntegrationSpec
.to(Sink.ignore)
}
+ "modify ack deadline" in {
+ val projectId = "pekko-connectors"
+ val topic = "simpleTopic"
+ val subscription = "simpleSubscription"
+ val subscriptionFqrs = s"projects/$projectId/subscriptions/$subscription"
+
+ // publish a message
+ val msg = ByteString.copyFromUtf8("Hello deadline extension!")
+ val publishRequest = PublishRequest()
+ .withTopic(s"projects/$projectId/topics/$topic")
+ .addMessages(PubsubMessage().withData(msg))
+ Source.single(publishRequest).via(GooglePubSub.publish(parallelism = 1)).runWith(Sink.head).futureValue
+
+ val request = StreamingPullRequest()
+ .withSubscription(subscriptionFqrs)
+ .withStreamAckDeadlineSeconds(10)
+
+ // #modify-ack-deadline
+ // subscribe, extend deadline, then ack
+ val result = GooglePubSub.subscribe(request, pollInterval = 1.second)
+ .take(1)
+ .mapAsync(1) { receivedMessage =>
+ // extend the ack deadline
+ val modifyRequest = ModifyAckDeadlineRequest(
+ subscription = subscriptionFqrs,
+ ackIds = Seq(receivedMessage.ackId),
+ ackDeadlineSeconds = 30)
+ Source.single(modifyRequest)
+ .via(GooglePubSub.modifyAckDeadlineFlow())
+ .runWith(Sink.head)
+ .map(_ => receivedMessage)(system.dispatcher)
+ }
+ .alsoTo(
+ Flow[ReceivedMessage]
+ .map(msg => AcknowledgeRequest(subscriptionFqrs, Seq(msg.ackId)))
+ .to(GooglePubSub.acknowledge(parallelism = 1)))
+ .runWith(Sink.head)
+ // #modify-ack-deadline
+
+ result.futureValue.message.value.data shouldBe msg
+ }
+
+ "auto extend ack deadlines" in {
+ val projectId = "pekko-connectors"
+ val topic = "simpleTopic"
+ val subscription = "simpleSubscription"
+ val subscriptionFqrs = s"projects/$projectId/subscriptions/$subscription"
+
+ // publish a message
+ val msg = ByteString.copyFromUtf8("Hello auto-extend!")
+ val publishRequest = PublishRequest()
+ .withTopic(s"projects/$projectId/topics/$topic")
+ .addMessages(PubsubMessage().withData(msg))
+ Source.single(publishRequest).via(GooglePubSub.publish(parallelism = 1)).runWith(Sink.head).futureValue
+
+ val request = StreamingPullRequest()
+ .withSubscription(subscriptionFqrs)
+ .withStreamAckDeadlineSeconds(10)
+
+ // #subscribe-auto-extend
+ // subscribe with auto-extending deadlines, then ack
+ val result = GooglePubSub.subscribe(request, pollInterval = 1.second)
+ .via(GooglePubSub.autoExtendAckDeadlines(subscriptionFqrs, 3.seconds, 30))
+ .take(1)
+ .alsoTo(
+ Flow[ReceivedMessage]
+ .map(msg => AcknowledgeRequest(subscriptionFqrs, Seq(msg.ackId)))
+ .to(GooglePubSub.acknowledge(parallelism = 1)))
+ .runWith(Sink.head)
+ // #subscribe-auto-extend
+
+ result.futureValue.message.value.data shouldBe msg
+ }
+
+ "auto extend prevents redelivery during slow processing" in {
+ val projectId = "pekko-connectors"
+ val topic = "simpleTopic"
+ val subscription = "simpleSubscription"
+ val subscriptionFqrs = s"projects/$projectId/subscriptions/$subscription"
+
+ // publish a unique message
+ val msg = ByteString.copyFromUtf8(s"slow-processing-${System.nanoTime()}")
+ val publishRequest = PublishRequest()
+ .withTopic(s"projects/$projectId/topics/$topic")
+ .addMessages(PubsubMessage().withData(msg))
+ Source.single(publishRequest).via(GooglePubSub.publish(parallelism = 1)).runWith(Sink.head).futureValue
+
+ val request = StreamingPullRequest()
+ .withSubscription(subscriptionFqrs)
+ .withStreamAckDeadlineSeconds(10)
+
+ // Subscribe with auto-extend, simulate slow processing (15s > 10s ack deadline),
+ // then ack. Without the deadline extension the message would be redelivered.
+ val received = GooglePubSub.subscribe(request, pollInterval = 1.second)
+ .via(GooglePubSub.autoExtendAckDeadlines(subscriptionFqrs, 3.seconds, 10))
+ .filter(_.message.exists(_.data == msg))
+ .take(1)
+ .mapAsync(1) { receivedMessage =>
+ // simulate slow processing that exceeds the ack deadline
+ org.apache.pekko.pattern.after(15.seconds)(
+ Future.successful(receivedMessage))(system)
+ }
+ .alsoTo(
+ Flow[ReceivedMessage]
+ .map(m => AcknowledgeRequest(subscriptionFqrs, Seq(m.ackId)))
+ .to(GooglePubSub.acknowledge(parallelism = 1)))
+ .runWith(Sink.head)
+
+ received.futureValue(timeout(30.seconds)).message.value.data shouldBe msg
+
+ // verify no redelivery — subscription should be empty (for our message)
+ val redelivered = GooglePubSub.subscribe(request, pollInterval = 1.second)
+ .filter(_.message.exists(_.data == msg))
+ .idleTimeout(12.seconds)
+ .runWith(Sink.seq)
+ .failed
+ .futureValue
+
+ redelivered shouldBe a[java.util.concurrent.TimeoutException]
+ }
+
"republish" in {
val msg = "Labas!"
@@ -266,6 +387,267 @@ class IntegrationSpec
.futureValue
}
+ "reconnect after emulator restart" in {
+ import pekko.stream.RestartSettings
+ import scala.sys.process._
+
+ val projectId = "pekko-connectors"
+ val topic = "simpleTopic"
+ val subscription = "simpleSubscription"
+ val subscriptionFqrs = s"projects/$projectId/subscriptions/$subscription"
+ val topicFqrs = s"projects/$projectId/topics/$topic"
+
+ val restartSettings = RestartSettings(
+ minBackoff = 100.millis,
+ maxBackoff = 5.seconds,
+ randomFactor = 0.2)
+
+ val request = StreamingPullRequest()
+ .withSubscription(subscriptionFqrs)
+ .withStreamAckDeadlineSeconds(10)
+
+ // publish a message before restart
+ val msg1 = ByteString.copyFromUtf8(s"before-restart-${System.nanoTime()}")
+ Source.single(PublishRequest(topicFqrs, Seq(PubsubMessage().withData(msg1))))
+ .via(GooglePubSub.publish(parallelism = 1)).runWith(Sink.head).futureValue
+
+ // start reconnecting subscriber — take first matching message, ack everything
+ val first = GooglePubSub
+ .subscribe(request, 1.second, restartSettings)
+ .alsoTo(
+ Flow[ReceivedMessage]
+ .map(msg => AcknowledgeRequest(subscriptionFqrs, Seq(msg.ackId)))
+ .to(GooglePubSub.acknowledge(parallelism = 1)))
+ .filter(_.message.exists(_.data == msg1))
+ .runWith(Sink.head)
+
+ first.futureValue.message.value.data shouldBe msg1
+
+ // restart emulator — this kills all gRPC connections
+ "docker compose restart gcloud-pubsub-emulator".!
+
+ // wait for emulator to come back up
+ Thread.sleep(8000)
+
+ // recreate topic and subscription (emulator loses state on restart)
+ s"docker compose exec gcloud-pubsub-emulator curl -s -X PUT http://localhost:8538/v1/projects/$projectId/topics/$topic".!
+ s"docker compose exec gcloud-pubsub-emulator curl -s -X PUT http://localhost:8538/v1/projects/$projectId/subscriptions/$subscription -H Content-Type:application/json -d {\"topic\":\"projects/$projectId/topics/$topic\"}".!
+
+ Thread.sleep(2000)
+
+ // publish a second message after restart
+ val msg2 = ByteString.copyFromUtf8(s"after-restart-${System.nanoTime()}")
+ Source.single(PublishRequest(topicFqrs, Seq(PubsubMessage().withData(msg2))))
+ .via(GooglePubSub.publish(parallelism = 1)).runWith(Sink.head).futureValue
+
+ // start another reconnecting subscriber — should connect to the restarted emulator
+ val second = GooglePubSub
+ .subscribe(request, 1.second, restartSettings)
+ .alsoTo(
+ Flow[ReceivedMessage]
+ .map(msg => AcknowledgeRequest(subscriptionFqrs, Seq(msg.ackId)))
+ .to(GooglePubSub.acknowledge(parallelism = 1)))
+ .runWith(Sink.head)
+
+ second.futureValue.message.value.data shouldBe msg2
+ }
+
+ "nack causes immediate redelivery" in {
+ val projectId = "pekko-connectors"
+ val topic = "simpleTopic"
+ val subscription = "simpleSubscription"
+ val subscriptionFqrs = s"projects/$projectId/subscriptions/$subscription"
+
+ // publish a unique message
+ val msg = ByteString.copyFromUtf8(s"nack-test-${System.nanoTime()}")
+ val publishRequest = PublishRequest()
+ .withTopic(s"projects/$projectId/topics/$topic")
+ .addMessages(PubsubMessage().withData(msg))
+ Source.single(publishRequest).via(GooglePubSub.publish(parallelism = 1)).runWith(Sink.head).futureValue
+
+ val request = StreamingPullRequest()
+ .withSubscription(subscriptionFqrs)
+ .withStreamAckDeadlineSeconds(10)
+
+ // receive the message and nack it
+ val nacked = GooglePubSub.subscribe(request, pollInterval = 1.second)
+ .filter(_.message.exists(_.data == msg))
+ .take(1)
+ .alsoTo(
+ Flow[ReceivedMessage]
+ .map(m => AcknowledgeRequest(subscriptionFqrs, Seq(m.ackId)))
+ .to(GooglePubSub.nack(parallelism = 1)))
+ .runWith(Sink.head)
+
+ nacked.futureValue.message.value.data shouldBe msg
+
+ // the nacked message should be redelivered quickly
+ val redelivered = GooglePubSub.subscribe(request, pollInterval = 1.second)
+ .filter(_.message.exists(_.data == msg))
+ .take(1)
+ .alsoTo(
+ Flow[ReceivedMessage]
+ .map(m => AcknowledgeRequest(subscriptionFqrs, Seq(m.ackId)))
+ .to(GooglePubSub.acknowledge(parallelism = 1)))
+ .runWith(Sink.head)
+
+ redelivered.futureValue.message.value.data shouldBe msg
+ }
+
+ "dynamic deadline modification" in {
+ val projectId = "pekko-connectors"
+ val topic = "simpleTopic"
+ val subscription = "simpleSubscription"
+ val subscriptionFqrs = s"projects/$projectId/subscriptions/$subscription"
+
+ // publish a message
+ val msg = ByteString.copyFromUtf8(s"dynamic-deadline-${System.nanoTime()}")
+ val publishRequest = PublishRequest()
+ .withTopic(s"projects/$projectId/topics/$topic")
+ .addMessages(PubsubMessage().withData(msg))
+ Source.single(publishRequest).via(GooglePubSub.publish(parallelism = 1)).runWith(Sink.head).futureValue
+
+ val request = StreamingPullRequest()
+ .withSubscription(subscriptionFqrs)
+ .withStreamAckDeadlineSeconds(10)
+
+ // subscribe, dynamically set deadline based on message content, then ack
+ val result = GooglePubSub.subscribe(request, pollInterval = 1.second)
+ .filter(_.message.exists(_.data == msg))
+ .take(1)
+ .via(GooglePubSub.modifyAckDeadlineDynamic(subscriptionFqrs) { _ => 30 })
+ .alsoTo(
+ Flow[ReceivedMessage]
+ .map(m => AcknowledgeRequest(subscriptionFqrs, Seq(m.ackId)))
+ .to(GooglePubSub.acknowledge(parallelism = 1)))
+ .runWith(Sink.head)
+
+ result.futureValue.message.value.data shouldBe msg
+ }
+
+ "flow control limits outstanding messages" in {
+ import org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.FlowControl
+
+ val projectId = "pekko-connectors"
+ val topic = "simpleTopic"
+ val subscription = "simpleSubscription"
+ val subscriptionFqrs = s"projects/$projectId/subscriptions/$subscription"
+
+ // publish several messages
+ val prefix = s"fc-test-${System.nanoTime()}"
+ val messages = (1 to 5).map { i =>
+ PubsubMessage().withData(ByteString.copyFromUtf8(s"$prefix-$i"))
+ }
+ val publishRequest = PublishRequest()
+ .withTopic(s"projects/$projectId/topics/$topic")
+ .addAllMessages(messages)
+ Source.single(publishRequest).via(GooglePubSub.publish(parallelism = 1)).runWith(Sink.head).futureValue
+
+ val request = StreamingPullRequest()
+ .withSubscription(subscriptionFqrs)
+ .withStreamAckDeadlineSeconds(10)
+
+ // flow control with max 2 outstanding messages
+ val fc = FlowControl(maxOutstandingMessages = 2)
+
+ val maxSeen = new java.util.concurrent.atomic.AtomicLong(0)
+
+ val result = GooglePubSub.subscribe(request, pollInterval = 1.second)
+ .via(GooglePubSub.flowControlGate(fc))
+ .take(5)
+ .mapAsync(2) { msg =>
+ val current = fc.outstandingCount
+ maxSeen.updateAndGet(old => math.max(old, current))
+ // simulate some processing
+ org.apache.pekko.pattern.after(200.millis)(
+ Future.successful(msg))(system)
+ }
+ .map(msg => AcknowledgeRequest(subscriptionFqrs, Seq(msg.ackId)))
+ .via(GooglePubSub.acknowledgeFlow(fc))
+ .runWith(Sink.seq)
+
+ result.futureValue should have size 5
+ // max outstanding should never exceed our limit (allow +1 for race between acquire and check)
+ maxSeen.get() should be <= 3L
+ }
+
+ "auto extend with maxAckExtensionPeriod" in {
+ val projectId = "pekko-connectors"
+ val topic = "simpleTopic"
+ val subscription = "simpleSubscription"
+ val subscriptionFqrs = s"projects/$projectId/subscriptions/$subscription"
+
+ // publish a message
+ val msg = ByteString.copyFromUtf8(s"max-ext-${System.nanoTime()}")
+ val publishRequest = PublishRequest()
+ .withTopic(s"projects/$projectId/topics/$topic")
+ .addMessages(PubsubMessage().withData(msg))
+ Source.single(publishRequest).via(GooglePubSub.publish(parallelism = 1)).runWith(Sink.head).futureValue
+
+ val request = StreamingPullRequest()
+ .withSubscription(subscriptionFqrs)
+ .withStreamAckDeadlineSeconds(10)
+
+ // use auto-extend with a very short maxAckExtensionPeriod
+ val result = GooglePubSub.subscribe(request, pollInterval = 1.second)
+ .via(GooglePubSub.autoExtendAckDeadlines(subscriptionFqrs, 3.seconds, 10, 5.seconds))
+ .filter(_.message.exists(_.data == msg))
+ .take(1)
+ .alsoTo(
+ Flow[ReceivedMessage]
+ .map(m => AcknowledgeRequest(subscriptionFqrs, Seq(m.ackId)))
+ .to(GooglePubSub.acknowledge(parallelism = 1)))
+ .runWith(Sink.head)
+
+ result.futureValue.message.value.data shouldBe msg
+ }
+
+ "adaptive deadline distribution" in {
+ import org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc.AckDeadlineDistribution
+
+ val projectId = "pekko-connectors"
+ val topic = "simpleTopic"
+ val subscription = "simpleSubscription"
+ val subscriptionFqrs = s"projects/$projectId/subscriptions/$subscription"
+
+ val dist = AckDeadlineDistribution(initialDeadlineSeconds = 10)
+
+ // publish several messages
+ val prefix = s"adaptive-${System.nanoTime()}"
+ val messages = (1 to 3).map { i =>
+ PubsubMessage().withData(ByteString.copyFromUtf8(s"$prefix-$i"))
+ }
+ val publishRequest = PublishRequest()
+ .withTopic(s"projects/$projectId/topics/$topic")
+ .addAllMessages(messages)
+ Source.single(publishRequest).via(GooglePubSub.publish(parallelism = 1)).runWith(Sink.head).futureValue
+
+ val request = StreamingPullRequest()
+ .withSubscription(subscriptionFqrs)
+ .withStreamAckDeadlineSeconds(10)
+
+ // initially should use the initialDeadlineSeconds
+ dist.currentDeadlineSeconds shouldBe 10
+
+ // subscribe with adaptive deadline, process with a small delay, then ack
+ val result = GooglePubSub.subscribe(request, pollInterval = 1.second)
+ .via(GooglePubSub.autoExtendAckDeadlines(subscriptionFqrs, 3.seconds, dist))
+ .take(3)
+ .mapAsync(1) { msg =>
+ // simulate ~2 second processing
+ org.apache.pekko.pattern.after(2.seconds)(Future.successful(msg))(system)
+ }
+ .map(msg => AcknowledgeRequest(subscriptionFqrs, Seq(msg.ackId)))
+ .via(GooglePubSub.acknowledgeFlow(dist))
+ .runWith(Sink.seq)
+
+ result.futureValue should have size 3
+
+ // after processing, the adaptive deadline should have adapted
+ // with ~2s processing, p99 should be around 3s (rounded up), clamped to min 10s
+ dist.currentDeadlineSeconds should be >= 10
+ }
+
"custom publisher" in {
// #attributes
val settings = PubSubSettings(system)
diff --git a/google-cloud-pub-sub-grpc/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/AutoExtendAckDeadlinesSpec.scala b/google-cloud-pub-sub-grpc/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/AutoExtendAckDeadlinesSpec.scala
new file mode 100644
index 000000000..1038927df
--- /dev/null
+++ b/google-cloud-pub-sub-grpc/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/AutoExtendAckDeadlinesSpec.scala
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.connectors.googlecloud.pubsub.grpc
+
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.ActorSystem
+import pekko.stream.scaladsl.{ Sink, Source }
+import pekko.stream.connectors.googlecloud.pubsub.grpc.scaladsl.{ GooglePubSub, GrpcSubscriber, PubSubAttributes }
+import com.google.protobuf.ByteString
+import com.google.pubsub.v1.pubsub._
+
+import scala.concurrent.{ Future, Promise }
+import scala.concurrent.duration._
+
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.concurrent.ScalaFutures
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+class AutoExtendAckDeadlinesSpec
+ extends AnyWordSpec
+ with Matchers
+ with BeforeAndAfterAll
+ with ScalaFutures {
+
+ implicit val system: ActorSystem = ActorSystem("AutoExtendAckDeadlinesSpec")
+ implicit val patience: PatienceConfig = PatienceConfig(10.seconds, 100.millis)
+
+ val subscription = "projects/test/subscriptions/test-sub"
+
+ def makeMsg(id: String): ReceivedMessage =
+ ReceivedMessage(
+ ackId = id,
+ message = Some(PubsubMessage(data = ByteString.copyFromUtf8(s"test-$id"))))
+
+ /** Stub that fails modifyAckDeadline calls. */
+ class FailingModifyAckDeadlineClient extends TestSubscriberClientBase {
+ override def modifyAckDeadline(
+ in: ModifyAckDeadlineRequest): Future[com.google.protobuf.empty.Empty] =
+ Future.failed(new RuntimeException("simulated modifyAckDeadline failure"))
+ }
+
+ /** Stub that succeeds for modifyAckDeadline calls. */
+ class SucceedingClient extends TestSubscriberClientBase {
+ override def modifyAckDeadline(
+ in: ModifyAckDeadlineRequest): Future[com.google.protobuf.empty.Empty] =
+ Future.successful(com.google.protobuf.empty.Empty())
+ }
+
+ "autoExtendAckDeadlines (fixed deadline)" should {
+
+ "fail the main stream when the ticker's modifyAckDeadline fails" in {
+ val testSubscriber = new GrpcSubscriber(new FailingModifyAckDeadlineClient())
+
+ val result = Source(List(makeMsg("1"), makeMsg("2"), makeMsg("3")))
+ .throttle(1, 500.millis)
+ .via(GooglePubSub.autoExtendAckDeadlines(subscription, 200.millis, 30))
+ .withAttributes(PubSubAttributes.subscriber(testSubscriber))
+ .runWith(Sink.seq)
+
+ val ex = result.failed.futureValue
+ ex shouldBe an[AckDeadlineExtensionException]
+ ex.getMessage should include("ticker failed")
+ ex.getCause.getMessage should include("simulated modifyAckDeadline failure")
+ }
+
+ "fail an idle stream when the ticker dies after the last element" in {
+ val testSubscriber = new GrpcSubscriber(new FailingModifyAckDeadlineClient())
+
+ // Emit one message then idle — the ticker should abort the stream even
+ // though no further elements arrive.
+ val result = Source
+ .single(makeMsg("only"))
+ .concat(Source.maybe[ReceivedMessage]) // keeps the stream open indefinitely
+ .via(GooglePubSub.autoExtendAckDeadlines(subscription, 200.millis, 30))
+ .withAttributes(PubSubAttributes.subscriber(testSubscriber))
+ .runWith(Sink.seq)
+
+ val ex = result.failed.futureValue
+ ex shouldBe an[AckDeadlineExtensionException]
+ ex.getMessage should include("ticker failed")
+ }
+
+ "pass messages through when ticker is healthy" in {
+ val testSubscriber = new GrpcSubscriber(new SucceedingClient())
+
+ val result = Source(List(makeMsg("1"), makeMsg("2"), makeMsg("3")))
+ .via(GooglePubSub.autoExtendAckDeadlines(subscription, 1.second, 30))
+ .withAttributes(PubSubAttributes.subscriber(testSubscriber))
+ .runWith(Sink.seq)
+
+ result.futureValue should have size 3
+ result.futureValue.map(_.ackId) shouldBe Seq("1", "2", "3")
+ }
+ }
+
+ "autoExtendAckDeadlines (adaptive)" should {
+
+ "fail the main stream when the ticker's modifyAckDeadline fails" in {
+ val testSubscriber = new GrpcSubscriber(new FailingModifyAckDeadlineClient())
+ val dist = AckDeadlineDistribution(initialDeadlineSeconds = 10)
+
+ val result = Source(List(makeMsg("1"), makeMsg("2"), makeMsg("3")))
+ .throttle(1, 500.millis)
+ .via(GooglePubSub.autoExtendAckDeadlines(subscription, 200.millis, dist))
+ .withAttributes(PubSubAttributes.subscriber(testSubscriber))
+ .runWith(Sink.seq)
+
+ val ex = result.failed.futureValue
+ ex shouldBe an[AckDeadlineExtensionException]
+ ex.getMessage should include("ticker failed")
+ }
+
+ "fail an idle stream when the ticker dies after the last element" in {
+ val testSubscriber = new GrpcSubscriber(new FailingModifyAckDeadlineClient())
+ val dist = AckDeadlineDistribution(initialDeadlineSeconds = 10)
+
+ val result = Source
+ .single(makeMsg("only"))
+ .concat(Source.maybe[ReceivedMessage])
+ .via(GooglePubSub.autoExtendAckDeadlines(subscription, 200.millis, dist))
+ .withAttributes(PubSubAttributes.subscriber(testSubscriber))
+ .runWith(Sink.seq)
+
+ val ex = result.failed.futureValue
+ ex shouldBe an[AckDeadlineExtensionException]
+ ex.getMessage should include("ticker failed")
+ }
+
+ "pass messages through when ticker is healthy" in {
+ val testSubscriber = new GrpcSubscriber(new SucceedingClient())
+ val dist = AckDeadlineDistribution(initialDeadlineSeconds = 10)
+
+ val result = Source(List(makeMsg("1"), makeMsg("2"), makeMsg("3")))
+ .via(GooglePubSub.autoExtendAckDeadlines(subscription, 1.second, dist))
+ .withAttributes(PubSubAttributes.subscriber(testSubscriber))
+ .runWith(Sink.seq)
+
+ result.futureValue should have size 3
+ }
+ }
+
+ override def afterAll(): Unit =
+ system.terminate()
+}
+
+/**
+ * Base stub for [[SubscriberClient]] that provides default implementations
+ * for all methods. Tests override only the methods they need.
+ */
+trait TestSubscriberClientBase extends SubscriberClient {
+ import pekko.NotUsed
+
+ override def createSubscription(in: Subscription): Future[Subscription] = ???
+ override def getSubscription(in: GetSubscriptionRequest): Future[Subscription] = ???
+ override def updateSubscription(in: UpdateSubscriptionRequest): Future[Subscription] = ???
+ override def listSubscriptions(in: ListSubscriptionsRequest): Future[ListSubscriptionsResponse] = ???
+ override def deleteSubscription(in: DeleteSubscriptionRequest): Future[com.google.protobuf.empty.Empty] = ???
+ override def modifyAckDeadline(in: ModifyAckDeadlineRequest): Future[com.google.protobuf.empty.Empty] = ???
+ override def acknowledge(in: AcknowledgeRequest): Future[com.google.protobuf.empty.Empty] = ???
+ override def pull(in: PullRequest): Future[PullResponse] = ???
+ override def streamingPull(
+ in: pekko.stream.scaladsl.Source[StreamingPullRequest, NotUsed])
+ : pekko.stream.scaladsl.Source[StreamingPullResponse, NotUsed] = ???
+ override def modifyPushConfig(in: ModifyPushConfigRequest): Future[com.google.protobuf.empty.Empty] = ???
+ override def getSnapshot(in: GetSnapshotRequest): Future[Snapshot] = ???
+ override def listSnapshots(in: ListSnapshotsRequest): Future[ListSnapshotsResponse] = ???
+ override def createSnapshot(in: CreateSnapshotRequest): Future[Snapshot] = ???
+ override def updateSnapshot(in: UpdateSnapshotRequest): Future[Snapshot] = ???
+ override def deleteSnapshot(in: DeleteSnapshotRequest): Future[com.google.protobuf.empty.Empty] = ???
+ override def seek(in: SeekRequest): Future[SeekResponse] = ???
+
+ override def close(): Future[Done] = Future.successful(Done)
+ override def closed: Future[Done] = Promise[Done]().future
+}
diff --git a/google-common/src/main/resources/reference.conf b/google-common/src/main/resources/reference.conf
index 63e79b552..7dcef98ad 100644
--- a/google-common/src/main/resources/reference.conf
+++ b/google-common/src/main/resources/reference.conf
@@ -8,8 +8,10 @@ pekko.connectors.google {
# Override to specify custom scopes
scopes = ${pekko.connectors.google.credentials.default-scopes}
- # Options: application-default, service-account, compute-engine, access-token, none
- # application-default first tries service-account, compute-engine then access-token
+ # Options: application-default, google-application-default, service-account, compute-engine, access-token, none
+ # application-default first tries service-account, compute-engine then access-token (custom implementation)
+ # google-application-default delegates to Google's google-auth-library-java which supports all GCP environments
+ # including GKE Workload Identity
provider = application-default
service-account {
@@ -48,6 +50,12 @@ pekko.connectors.google {
path = ${?GOOGLE_APPLICATION_CREDENTIALS}
}
+ google-application-default {
+ project-id = ""
+ project-id = ${?GOOGLE_CLOUD_PROJECT}
+ project-id = ${?GCLOUD_PROJECT}
+ }
+
access-token {
project-id = ""
token = ""
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/AccessTokenCredentials.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/AccessTokenCredentials.scala
index 7e15eb3f0..9f121cac1 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/AccessTokenCredentials.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/AccessTokenCredentials.scala
@@ -42,7 +42,7 @@ private[auth] final case class AccessTokenCredentials(projectId: String, accessT
new GoogleCredentials {
override def getAuthenticationType: String = "OAuth2"
override def getRequestMetadata(uri: URI): util.Map[String, util.List[String]] =
- util.Collections.singletonMap("Authorization", util.Collections.singletonList(accessToken))
+ util.Collections.singletonMap("Authorization", util.Collections.singletonList("Bearer " + accessToken))
override def hasRequestMetadata: Boolean = true
override def hasRequestMetadataOnly: Boolean = true
override def refresh(): Unit = ()
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/Credentials.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/Credentials.scala
index 4e86c3ff2..3027cf526 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/Credentials.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/Credentials.scala
@@ -73,11 +73,18 @@ object Credentials {
}
}
}
- case "service-account" => parseServiceAccount(c)
- case "compute-engine" => parseComputeEngine(c)
- case "user-access" => parseUserAccess(c)
- case "access-token" => parseAccessToken(c)
- case "none" => parseNone(c)
+ case "google-application-default" => parseGoogleApplicationDefault(c)
+ case "service-account" => parseServiceAccount(c)
+ case "compute-engine" => parseComputeEngine(c)
+ case "user-access" => parseUserAccess(c)
+ case "access-token" => parseAccessToken(c)
+ case "none" => parseNone(c)
+ }
+
+ private def parseGoogleApplicationDefault(c: Config) = {
+ val scopes = c.getStringList("scopes").asScala.toSet
+ val projectId = c.getString("google-application-default.project-id")
+ GoogleApplicationDefaultCredentials(projectId, scopes)
}
private def parseServiceAccount(c: Config)(implicit system: ClassicActorSystemProvider) = {
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleApplicationDefaultCredentials.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleApplicationDefaultCredentials.scala
new file mode 100644
index 000000000..fa18b8823
--- /dev/null
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleApplicationDefaultCredentials.scala
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.connectors.google.auth
+
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.stream.connectors.google.RequestSettings
+import com.google.auth.oauth2.{ GoogleCredentials => OAuthGoogleCredentials }
+import com.google.auth.{ Credentials => GoogleCredentials }
+
+import scala.concurrent.ExecutionContext
+import scala.jdk.CollectionConverters._
+
+/**
+ * Credentials that delegate to Google's `google-auth-library-java` for credential resolution.
+ * This uses `GoogleCredentials.getApplicationDefault()` which supports all GCP environments
+ * including GKE Workload Identity.
+ */
+@InternalApi
+private[connectors] object GoogleApplicationDefaultCredentials {
+
+ def apply(projectId: String, scopes: Set[String]): GoogleApplicationDefaultCredentials = {
+ val googleCredentials = OAuthGoogleCredentials.getApplicationDefault
+ .createScoped(scopes.asJava)
+ new GoogleApplicationDefaultCredentials(projectId, googleCredentials)
+ }
+
+}
+
+/**
+ * Credentials backed by Google's `google-auth-library-java`.
+ */
+@InternalApi
+private[connectors] final class GoogleApplicationDefaultCredentials(
+ override val projectId: String,
+ private val googleCredentials: OAuthGoogleCredentials) extends Credentials {
+
+ override def asGoogle(implicit ec: ExecutionContext, settings: RequestSettings): GoogleCredentials =
+ googleCredentials
+}
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/ProvidedCredentials.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/ProvidedCredentials.scala
new file mode 100644
index 000000000..8d0c61c6e
--- /dev/null
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/ProvidedCredentials.scala
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.connectors.google.auth
+
+import org.apache.pekko
+import pekko.annotation.ApiMayChange
+import pekko.stream.connectors.google.RequestSettings
+import com.google.auth.{ Credentials => GoogleCredentials }
+
+import scala.concurrent.ExecutionContext
+
+/**
+ * Wraps a user-provided [[com.google.auth.Credentials]] instance for use with Google connectors.
+ * This allows users to bring their own credentials resolved through any mechanism.
+ *
+ * @param projectId the GCP project id
+ * @param credentials the user-provided Google credentials
+ * @since 2.0.0
+ */
+@ApiMayChange
+final class ProvidedCredentials(
+ override val projectId: String,
+ private val credentials: GoogleCredentials) extends Credentials {
+
+ override def asGoogle(implicit ec: ExecutionContext, settings: RequestSettings): GoogleCredentials =
+ credentials
+
+ /**
+ * Java API
+ */
+ def getProjectId: String = projectId
+}
+
+object ProvidedCredentials {
+
+ /**
+ * Scala API: Create credentials from a user-provided [[com.google.auth.Credentials]] instance.
+ */
+ def apply(projectId: String, credentials: GoogleCredentials): ProvidedCredentials =
+ new ProvidedCredentials(projectId, credentials)
+
+ /**
+ * Java API: Create credentials from a user-provided [[com.google.auth.Credentials]] instance.
+ */
+ def create(projectId: String, credentials: GoogleCredentials): ProvidedCredentials =
+ apply(projectId, credentials)
+}
diff --git a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/CredentialsSpec.scala b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/CredentialsSpec.scala
new file mode 100644
index 000000000..c6bda3783
--- /dev/null
+++ b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/CredentialsSpec.scala
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.connectors.google.auth
+
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.testkit.TestKit
+import com.typesafe.config.ConfigFactory
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpecLike
+
+class CredentialsSpec
+ extends TestKit(ActorSystem("CredentialsSpec"))
+ with AnyWordSpecLike
+ with Matchers
+ with BeforeAndAfterAll {
+
+ override def afterAll(): Unit = {
+ TestKit.shutdownActorSystem(system)
+ super.afterAll()
+ }
+
+ "Credentials" should {
+
+ "parse 'none' provider" in {
+ val config = ConfigFactory.parseString(
+ """
+ |provider = none
+ |scopes = []
+ |default-scopes = []
+ |none {
+ | project-id = "test-project"
+ |}
+ |service-account {
+ | project-id = ""
+ | client-email = ""
+ | private-key = ""
+ | path = "/tmp/fake"
+ |}
+ |compute-engine.timeout = 1s
+ |user-access {
+ | project-id = ""
+ | client-id = ""
+ | client-secret = ""
+ | refresh-token = ""
+ | path = "/tmp/fake"
+ |}
+ |access-token {
+ | project-id = ""
+ | token = ""
+ |}
+ |google-application-default {
+ | project-id = ""
+ |}
+ """.stripMargin)
+
+ val credentials = Credentials(config)
+ credentials shouldBe a[NoCredentials]
+ credentials.projectId shouldBe "test-project"
+ }
+
+ "parse 'access-token' provider" in {
+ val config = ConfigFactory.parseString(
+ """
+ |provider = access-token
+ |scopes = []
+ |default-scopes = []
+ |none {
+ | project-id = ""
+ |}
+ |service-account {
+ | project-id = ""
+ | client-email = ""
+ | private-key = ""
+ | path = "/tmp/fake"
+ |}
+ |compute-engine.timeout = 1s
+ |user-access {
+ | project-id = ""
+ | client-id = ""
+ | client-secret = ""
+ | refresh-token = ""
+ | path = "/tmp/fake"
+ |}
+ |access-token {
+ | project-id = "token-project"
+ | token = "my-token"
+ |}
+ |google-application-default {
+ | project-id = ""
+ |}
+ """.stripMargin)
+
+ val credentials = Credentials(config)
+ credentials shouldBe a[AccessTokenCredentials]
+ credentials.projectId shouldBe "token-project"
+ }
+
+ "parse 'google-application-default' provider" in {
+ val config = ConfigFactory.parseString(
+ """
+ |provider = google-application-default
+ |scopes = ["https://www.googleapis.com/auth/cloud-platform"]
+ |default-scopes = []
+ |none {
+ | project-id = ""
+ |}
+ |service-account {
+ | project-id = ""
+ | client-email = ""
+ | private-key = ""
+ | path = "/tmp/fake"
+ |}
+ |compute-engine.timeout = 1s
+ |user-access {
+ | project-id = ""
+ | client-id = ""
+ | client-secret = ""
+ | refresh-token = ""
+ | path = "/tmp/fake"
+ |}
+ |access-token {
+ | project-id = ""
+ | token = ""
+ |}
+ |google-application-default {
+ | project-id = "gad-project"
+ |}
+ """.stripMargin)
+
+ // When application-default credentials are available (e.g. via `gcloud auth application-default login`),
+ // this succeeds. In CI/test environments without credentials, getApplicationDefault() throws IOException.
+ try {
+ val credentials = Credentials(config)
+ credentials shouldBe a[GoogleApplicationDefaultCredentials]
+ credentials.projectId shouldBe "gad-project"
+ } catch {
+ case ex: java.io.IOException =>
+ ex.getMessage should include("Application Default Credentials")
+ }
+ }
+ }
+}
diff --git a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/ProvidedCredentialsSpec.scala b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/ProvidedCredentialsSpec.scala
new file mode 100644
index 000000000..5e2e4ab5d
--- /dev/null
+++ b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/ProvidedCredentialsSpec.scala
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.connectors.google.auth
+
+import org.apache.pekko.stream.connectors.google.RequestSettings
+import com.google.auth.oauth2.{ GoogleCredentials => OAuthGoogleCredentials }
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+import scala.concurrent.ExecutionContext
+
+class ProvidedCredentialsSpec extends AnyWordSpec with Matchers {
+
+ implicit val ec: ExecutionContext = ExecutionContext.global
+ implicit val settings: RequestSettings = null // not used by ProvidedCredentials
+
+ "ProvidedCredentials" should {
+
+ "wrap user-provided credentials and return them via asGoogle" in {
+ val googleCreds = OAuthGoogleCredentials.newBuilder().build()
+ val provided = ProvidedCredentials("my-project", googleCreds)
+
+ provided.projectId shouldBe "my-project"
+ provided.asGoogle shouldBe googleCreds
+ }
+
+ "expose projectId via Java API" in {
+ val googleCreds = OAuthGoogleCredentials.newBuilder().build()
+ val provided = ProvidedCredentials.create("java-project", googleCreds)
+
+ provided.getProjectId shouldBe "java-project"
+ }
+
+ "use Scala apply factory method" in {
+ val googleCreds = OAuthGoogleCredentials.newBuilder().build()
+ val provided = ProvidedCredentials("scala-project", googleCreds)
+
+ provided shouldBe a[ProvidedCredentials]
+ provided.projectId shouldBe "scala-project"
+ }
+ }
+}
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
index 4a7b66a79..0995631f4 100644
--- a/project/Dependencies.scala
+++ b/project/Dependencies.scala
@@ -241,6 +241,7 @@ object Dependencies {
"org.apache.pekko" %% "pekko-http-spray-json" % PekkoHttpVersion,
"com.github.jwt-scala" %% "jwt-json-common" % JwtScalaVersion,
"com.google.auth" % "google-auth-library-credentials" % GoogleAuthVersion,
+ "com.google.auth" % "google-auth-library-oauth2-http" % GoogleAuthVersion,
"io.specto" % "hoverfly-java" % hoverflyVersion % Test) ++ Mockito)
val GoogleBigQuery = Seq(
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org