This is an automated email from the ASF dual-hosted git repository.
pjfanning 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 24f190e56 Google: allow OAuth2 credentials to be closed (#1927)
24f190e56 is described below
commit 24f190e56b5423680af1f070d4bc0735dcae3271
Author: PJ Fanning <[email protected]>
AuthorDate: Tue Sep 8 09:23:07 2026 +0100
Google: allow OAuth2 credentials to be closed (#1927)
OAuth2Credentials materialises a Source.actorRef in its constructor to cache
and refresh access tokens, and both the completion and failure matchers were
PartialFunction.empty, so no message could ever stop it. Credentials read
through GoogleExt are cached per ActorSystem and go away with it, but anyone
building GoogleSettings directly from a Config - one set of credentials per
tenant, say - leaked an actor and a materialised stream per instance with no
way to reclaim them.
Add Credentials.close(), a no-op by default, overridden in
OAuth2Credentials to
send a new Close command that the completion matcher turns into
CompletionStrategy.draining, so requests already queued are still served.
Requests made after close fail with an IllegalStateException rather than
hanging on a promise nothing will complete, and closing twice is a no-op.
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
docs/src/main/paradox/google-common.md | 8 +++++
.../connectors/google/auth/Credentials.scala | 12 ++++++++
.../connectors/google/auth/OAuth2Credentials.scala | 34 ++++++++++++++++------
.../google/auth/OAuth2CredentialsSpec.scala | 32 ++++++++++++++++++++
4 files changed, 77 insertions(+), 9 deletions(-)
diff --git a/docs/src/main/paradox/google-common.md
b/docs/src/main/paradox/google-common.md
index cde562891..30d29dcbe 100644
--- a/docs/src/main/paradox/google-common.md
+++ b/docs/src/main/paradox/google-common.md
@@ -36,6 +36,14 @@ Credentials will be loaded automatically:
Credentials can also be specified manually in your configuration file.
+Credentials that refresh OAuth2 access tokens keep a stream running to cache
and renew the token.
+
+Credentials read from an `ActorSystem` are cached per system and released when
it terminates, so
+they need no special handling. If you build @apidoc[GoogleSettings] directly
from a
+@javadoc[Config](com.typesafe.config.Config) — one set of credentials per
tenant, for example — call
+`close()` on the credentials when you are done with them to release that
stream. Credentials must
+not be used after they have been closed.
+
## Project id
The project id used for requests is resolved in this order:
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 83a843404..e1552372f 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
@@ -117,6 +117,18 @@ abstract class Credentials private[auth] () {
private[google] def projectId: String
+ /**
+ * Releases any resources held by these credentials, such as the stream that
caches and refreshes
+ * OAuth2 access tokens. The credentials must not be used again once closed,
and closing more than
+ * once has no further effect.
+ *
+ * Credentials obtained from
[[org.apache.pekko.stream.connectors.google.GoogleSettings GoogleSettings]]
+ * via an `ActorSystem` are cached per system and their resources are
released when that system
+ * terminates, so only credentials that are built directly, for example one
set of credentials per
+ * tenant, need to be closed.
+ */
+ def close(): Unit = ()
+
/**
* Wraps these credentials as a [[com.google.auth.Credentials]] for interop
with Google's Java client libraries.
* @param ec the [[scala.concurrent.ExecutionContext]] to use for blocking
requests if credentials are requested synchronously
diff --git
a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/OAuth2Credentials.scala
b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/OAuth2Credentials.scala
index 3aa2aa4d0..e180a2404 100644
---
a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/OAuth2Credentials.scala
+++
b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/OAuth2Credentials.scala
@@ -17,12 +17,13 @@ import org.apache.pekko
import pekko.annotation.InternalApi
import pekko.http.scaladsl.model.headers.OAuth2BearerToken
import pekko.stream.connectors.google.RequestSettings
-import pekko.stream.connectors.google.auth.OAuth2Credentials.{ ForceRefresh,
TokenRequest }
+import pekko.stream.connectors.google.auth.OAuth2Credentials.{ Close,
ForceRefresh, TokenRequest }
import pekko.stream.scaladsl.{ Sink, Source }
import pekko.stream.{ CompletionStrategy, Materializer, OverflowStrategy }
import com.google.auth.{ Credentials => GoogleCredentials }
import java.time.Clock
+import java.util.concurrent.atomic.AtomicBoolean
import scala.concurrent.{ ExecutionContext, Future, Promise }
@InternalApi
@@ -30,21 +31,33 @@ private[auth] object OAuth2Credentials {
sealed abstract class Command
final case class TokenRequest(promise: Promise[OAuth2BearerToken], settings:
RequestSettings) extends Command
case object ForceRefresh extends Command
+ case object Close extends Command
}
@InternalApi
private[auth] abstract class OAuth2Credentials(val projectId: String)(implicit
mat: Materializer) extends Credentials
with RetrievableCredentials {
- private val tokenStream = stream.run()
+ private[auth] val tokenStream = stream.run()
+ private val closed = new AtomicBoolean(false)
- override def get()(implicit ec: ExecutionContext, settings:
RequestSettings): Future[OAuth2BearerToken] = {
- val token = Promise[OAuth2BearerToken]()
- tokenStream ! TokenRequest(token, settings)
- token.future
- }
+ override def get()(implicit ec: ExecutionContext, settings:
RequestSettings): Future[OAuth2BearerToken] =
+ if (closed.get())
+ Future.failed(new IllegalStateException("These credentials have been
closed"))
+ else {
+ val token = Promise[OAuth2BearerToken]()
+ tokenStream ! TokenRequest(token, settings)
+ token.future
+ }
- def refresh(): Unit = tokenStream ! ForceRefresh
+ def refresh(): Unit = if (!closed.get()) tokenStream ! ForceRefresh
+
+ /**
+ * Completes the token stream, releasing the actor and the materialized
stages behind it. Requests
+ * already queued are still served; requests made after closing fail with an
+ * [[java.lang.IllegalStateException]].
+ */
+ override def close(): Unit = if (closed.compareAndSet(false, true))
tokenStream ! Close
override def asGoogle(implicit ec: ExecutionContext, settings:
RequestSettings): GoogleCredentials =
new GoogleOAuth2Credentials(this)(ec, settings)
@@ -56,7 +69,7 @@ private[auth] abstract class OAuth2Credentials(val projectId:
String)(implicit m
private def stream =
Source
.actorRef[OAuth2Credentials.Command](
- PartialFunction.empty[Any, CompletionStrategy],
+ { case Close => CompletionStrategy.draining },
PartialFunction.empty[Any, Throwable],
Int.MaxValue,
OverflowStrategy.fail)
@@ -76,6 +89,9 @@ private[auth] abstract class OAuth2Credentials(val projectId:
String)(implicit m
.recover { case _ => None }(ExecutionContext.parasitic)
case (_, ForceRefresh) =>
Future.successful(None)
+ case (cachedToken, Close) =>
+ // consumed by the completion matcher above, here only to keep
the match exhaustive
+ Future.successful(cachedToken)
}
})
}
diff --git
a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/OAuth2CredentialsSpec.scala
b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/OAuth2CredentialsSpec.scala
index c6e183871..9d1b48877 100644
---
a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/OAuth2CredentialsSpec.scala
+++
b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/OAuth2CredentialsSpec.scala
@@ -107,6 +107,38 @@ class OAuth2CredentialsSpec
request3.futureValue shouldEqual OAuth2BearerToken("second token")
}
+ "serve queued requests and then stop the token stream when closed" in {
+
+ val accessTokenPromise = Promise[AccessToken]()
+ val credentials = new OAuth2Credentials("dummyProject") {
+ override protected def getAccessToken()(implicit mat: Materializer,
+ settings: RequestSettings,
+ clock: Clock): Future[AccessToken] = accessTokenPromise.future
+ }
+ watch(credentials.tokenStream)
+
+ val request = credentials.get()
+ credentials.close()
+ accessTokenPromise.success(AccessToken("a token", JwtTime.nowSeconds +
120))
+
+ request.futureValue shouldEqual OAuth2BearerToken("a token")
+ expectTerminated(credentials.tokenStream)
+ }
+
+ "fail requests made after close, and tolerate closing twice" in {
+
+ val credentials = new OAuth2Credentials("dummyProject") {
+ override protected def getAccessToken()(implicit mat: Materializer,
+ settings: RequestSettings,
+ clock: Clock): Future[AccessToken] = Promise[AccessToken]().future
+ }
+
+ credentials.close()
+ credentials.close()
+
+ credentials.get().failed.futureValue shouldBe an[IllegalStateException]
+ }
+
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
