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-management.git
The following commit(s) were added to refs/heads/main by this push:
new 3d8ae24c better handling of resources when responses are handled (#905)
3d8ae24c is described below
commit 3d8ae24c88bd8d6d2232037678b5f7b15fee7ad6
Author: PJ Fanning <[email protected]>
AuthorDate: Tue Aug 4 18:44:07 2026 +0100
better handling of resources when responses are handled (#905)
---
.../discovery/consul/ConsulServiceDiscovery.scala | 17 +++++++----
.../internal/AbstractKubernetesApiImpl.scala | 30 ++++++++++++++------
.../internal/HttpContactPointBootstrap.scala | 33 ++++++++++++++++------
.../management/internal/HealthChecksImpl.scala | 15 ++++------
.../kubernetes/KubernetesApiImpl.scala | 32 ++++++++++++++-------
5 files changed, 85 insertions(+), 42 deletions(-)
diff --git
a/discovery-consul/src/main/scala/org/apache/pekko/discovery/consul/ConsulServiceDiscovery.scala
b/discovery-consul/src/main/scala/org/apache/pekko/discovery/consul/ConsulServiceDiscovery.scala
index a9294d9d..0cb26734 100644
---
a/discovery-consul/src/main/scala/org/apache/pekko/discovery/consul/ConsulServiceDiscovery.scala
+++
b/discovery-consul/src/main/scala/org/apache/pekko/discovery/consul/ConsulServiceDiscovery.scala
@@ -20,7 +20,6 @@ import pekko.annotation.ApiMayChange
import pekko.discovery.ServiceDiscovery.{ Resolved, ResolvedTarget }
import pekko.discovery.consul.ConsulServiceDiscovery._
import pekko.discovery.{ Lookup, ServiceDiscovery }
-import pekko.pattern.after
import org.kiwiproject.consul.Consul
import org.kiwiproject.consul.async.ConsulResponseCallback
import org.kiwiproject.consul.model.ConsulResponse
@@ -45,11 +44,17 @@ class ConsulServiceDiscovery(system: ActorSystem) extends
ServiceDiscovery {
override def lookup(lookup: Lookup, resolveTimeout: FiniteDuration):
Future[Resolved] = {
implicit val ec: ExecutionContext = system.dispatcher
- Future.firstCompletedOf(
- Seq(
- after(resolveTimeout, using = system.scheduler)(
- Future.failed(new TimeoutException(s"Lookup for [$lookup] timed-out,
within [$resolveTimeout]!"))),
- lookupInConsul(lookup.serviceName)))
+ // Use a Promise-based pattern instead of Future.firstCompletedOf to avoid
leaking
+ // the underlying Consul HTTP connections when the timeout fires first.
+ val promise = Promise[Resolved]()
+ val timeoutCancellable = system.scheduler.scheduleOnce(resolveTimeout) {
+ promise.tryFailure(new TimeoutException(s"Lookup for [$lookup]
timed-out, within [$resolveTimeout]!"))
+ }
+ lookupInConsul(lookup.serviceName).onComplete { result =>
+ timeoutCancellable.cancel()
+ promise.tryComplete(result)
+ }
+ promise.future
}
private def lookupInConsul(name: String)(implicit executionContext:
ExecutionContext): Future[Resolved] = {
diff --git
a/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/internal/AbstractKubernetesApiImpl.scala
b/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/internal/AbstractKubernetesApiImpl.scala
index 49fb33ab..9cb40e1f 100644
---
a/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/internal/AbstractKubernetesApiImpl.scala
+++
b/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/internal/AbstractKubernetesApiImpl.scala
@@ -24,7 +24,7 @@ import pekko.http.scaladsl.model._
import pekko.http.scaladsl.model.headers.{ Authorization, OAuth2BearerToken }
import pekko.http.scaladsl.unmarshalling.Unmarshal
import pekko.http.scaladsl.{ ConnectionContext, Http, HttpExt,
HttpsConnectionContext }
-import pekko.pattern.{ after, RetrySupport }
+import pekko.pattern.RetrySupport
import pekko.pki.kubernetes.PemManagersProvider
import pekko.stream.scaladsl.{ FileIO, Keep, Sink }
import pekko.util.ByteString
@@ -32,7 +32,7 @@ import pekko.util.ByteString
import java.nio.file.{ Files, Paths }
import javax.net.ssl.SSLContext
import scala.collection.immutable
-import scala.concurrent.{ ExecutionContext, Future }
+import scala.concurrent.{ ExecutionContext, Future, Promise }
import scala.util.control.NonFatal
/**
@@ -177,13 +177,25 @@ import scala.util.control.NonFatal
settings.tokenRetrySettings.randomFactor
)
- // make sure we always consume response body (in case of timeout)
- val strictResponse = response.flatMap(_.toStrict(settings.bodyReadTimeout))
-
- val timeout = after(settings.apiServerRequestTimeout, using =
system.scheduler)(
- Future.failed(new LeaseTimeoutException(s"$timeoutMsg. Is the API server
up?")))
-
- Future.firstCompletedOf(Seq(strictResponse, timeout))
+ // Use a Promise-based pattern instead of Future.firstCompletedOf so that
when the
+ // timeout fires first, we properly discard the in-flight response entity
to release
+ // the HTTP connection back to the pool.
+ val promise = Promise[HttpResponse]()
+ val timeoutCancellable =
system.scheduler.scheduleOnce(settings.apiServerRequestTimeout) {
+ promise.tryFailure(new LeaseTimeoutException(s"$timeoutMsg. Is the API
server up?"))
+ }
+ response.onComplete {
+ case scala.util.Success(resp) =>
+ timeoutCancellable.cancel()
+ if (!promise.trySuccess(resp)) {
+ // Timeout already fired — discard the response to release the
connection
+
resp.discardEntityBytes()(pekko.stream.Materializer.matFromSystem(system))
+ }
+ case scala.util.Failure(ex) =>
+ timeoutCancellable.cancel()
+ promise.tryFailure(ex)
+ }(system.dispatcher)
+ promise.future.flatMap(_.toStrict(settings.bodyReadTimeout))
}
protected def readConfigVarFromFilesystem(path: String, name: String):
Future[Option[String]] = {
diff --git
a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/internal/HttpContactPointBootstrap.scala
b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/internal/HttpContactPointBootstrap.scala
index 85d483ab..195c97d6 100644
---
a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/internal/HttpContactPointBootstrap.scala
+++
b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/internal/HttpContactPointBootstrap.scala
@@ -18,7 +18,7 @@ import java.security.{ KeyStore, SecureRandom }
import java.util.concurrent.ThreadLocalRandom
import java.util.concurrent.TimeoutException
import javax.net.ssl.{ KeyManager, KeyManagerFactory, SSLContext, TrustManager
}
-import scala.concurrent.Future
+import scala.concurrent.{ Future, Promise }
import scala.concurrent.duration._
import org.apache.pekko
@@ -44,7 +44,6 @@ import pekko.http.scaladsl.unmarshalling.Unmarshal
import pekko.management.cluster.bootstrap.ClusterBootstrapSettings
import
pekko.management.cluster.bootstrap.contactpoint.HttpBootstrapJsonProtocol.SeedNodes
import pekko.management.cluster.bootstrap.contactpoint.{
ClusterBootstrapRequests, HttpBootstrapJsonProtocol }
-import pekko.pattern.after
import pekko.pattern.pipe
import pekko.pki.kubernetes.PemManagersProvider
@@ -133,7 +132,6 @@ private[bootstrap] class HttpContactPointBootstrap(
private val probeInterval = settings.contactPoint.probeInterval
private val probeRequest =
ClusterBootstrapRequests.bootstrapSeedNodes(baseUri)
- private val replyTimeout = Future.failed(new TimeoutException(s"Probing
timeout of [$baseUri]"))
/**
* If probing keeps failing until the deadline triggers, we notify the
parent,
@@ -150,15 +148,34 @@ private[bootstrap] class HttpContactPointBootstrap(
override def receive = {
case ProbeTick =>
log.debug("Probing [{}] for seed nodes...", probeRequest.uri)
- val reply = if (probeRequest.uri.scheme == "https" &&
useCustomSslContext) {
+ val response = if (probeRequest.uri.scheme == "https" &&
useCustomSslContext) {
http.singleRequest(probeRequest, settings =
connectionPoolWithoutRetries,
connectionContext = clientSslContext)
} else {
http.singleRequest(probeRequest, settings =
connectionPoolWithoutRetries)
- }.flatMap(handleResponse)
-
- val afterTimeout = after(settings.contactPoint.probingFailureTimeout,
context.system.scheduler)(replyTimeout)
- Future.firstCompletedOf(List(reply, afterTimeout)).pipeTo(self)
+ }
+ // Use a Promise-based pattern instead of Future.firstCompletedOf so
that when the
+ // timeout fires first, we properly discard the in-flight response
entity to release
+ // the HTTP connection back to the pool.
+ val promise = Promise[SeedNodes]()
+ val timeoutCancellable =
context.system.scheduler.scheduleOnce(settings.contactPoint.probingFailureTimeout)
{
+ promise.tryFailure(new TimeoutException(s"Probing timeout of
[$baseUri]"))
+ }
+ response.flatMap(handleResponse).onComplete {
+ case scala.util.Success(seedNodes) =>
+ timeoutCancellable.cancel()
+ promise.trySuccess(seedNodes)
+ case scala.util.Failure(ex) =>
+ timeoutCancellable.cancel()
+ promise.tryFailure(ex)
+ }(context.dispatcher)
+ // If timeout fires first, discard the in-flight response to release the
connection
+ promise.future.failed.foreach { _ =>
+ response.foreach { resp =>
+
resp.discardEntityBytes()(pekko.stream.Materializer.matFromSystem(context.system))
+ }(context.dispatcher)
+ }(context.dispatcher)
+ promise.future.pipeTo(self)
case Status.Failure(cause) =>
log.warning("Probing [{}] failed due to: {}", probeRequest.uri,
cause.getMessage)
diff --git
a/management/src/main/scala/org/apache/pekko/management/internal/HealthChecksImpl.scala
b/management/src/main/scala/org/apache/pekko/management/internal/HealthChecksImpl.scala
index 5b8bb081..b6e0f8a3 100644
---
a/management/src/main/scala/org/apache/pekko/management/internal/HealthChecksImpl.scala
+++
b/management/src/main/scala/org/apache/pekko/management/internal/HealthChecksImpl.scala
@@ -203,19 +203,16 @@ final private[pekko] class HealthChecksImpl(system:
ExtendedActorSystem, setting
}
private def check(checks: immutable.Seq[HealthCheck]): Future[Either[String,
Unit]] = {
- val timeout = pekko.pattern.after(settings.checkTimeout, system.scheduler)(
- Future.failed(new RuntimeException) // will be enriched with which check
timed out below
- )
-
val spawnedChecks: Seq[Future[Either[String, Unit]]] = checks.map { check
=>
val checkName = check.getClass.getName
+ // Create a per-check timeout so each check gets its own timer,
+ // avoiding the shared timeout that leaks scheduler resources.
+ val timeout = pekko.pattern.after(settings.checkTimeout,
system.scheduler)(
+ Future.failed(CheckTimeoutException(s"Check [$checkName] timed out
after ${settings.checkTimeout}"))
+ )
Future.firstCompletedOf(
Seq(
- timeout.recoverWith {
- case _: Throwable =>
- Future.failed(
- CheckTimeoutException(s"Check [$checkName] timed out after
${settings.checkTimeout}"))
- },
+ timeout,
runCheck(check)
.map {
case true => Right(())
diff --git
a/rolling-update-kubernetes/src/main/scala/org/apache/pekko/rollingupdate/kubernetes/KubernetesApiImpl.scala
b/rolling-update-kubernetes/src/main/scala/org/apache/pekko/rollingupdate/kubernetes/KubernetesApiImpl.scala
index 188f74fd..2a60d99a 100644
---
a/rolling-update-kubernetes/src/main/scala/org/apache/pekko/rollingupdate/kubernetes/KubernetesApiImpl.scala
+++
b/rolling-update-kubernetes/src/main/scala/org/apache/pekko/rollingupdate/kubernetes/KubernetesApiImpl.scala
@@ -18,6 +18,7 @@ import java.nio.charset.StandardCharsets
import scala.collection.immutable
import scala.concurrent.ExecutionContext
import scala.concurrent.Future
+import scala.concurrent.Promise
import scala.util.control.NonFatal
import org.apache.pekko
@@ -41,7 +42,7 @@ import pekko.http.scaladsl.model._
import pekko.http.scaladsl.model.headers.Authorization
import pekko.http.scaladsl.model.headers.OAuth2BearerToken
import pekko.http.scaladsl.unmarshalling.Unmarshal
-import pekko.pattern.after
+
import pekko.pki.kubernetes.PemManagersProvider
import pekko.util.ByteString
@@ -274,13 +275,25 @@ PUTs must contain resourceVersions. Response:
}
}
- // make sure we always consume response body (in case of timeout)
- val strictResponse = response.flatMap(_.toStrict(settings.bodyReadTimeout))
-
- val timeout = after(settings.apiServiceRequestTimeout, using =
system.scheduler)(
- Future.failed(new PodCostTimeoutException(s"$timeoutMsg. Is the API
server up?")))
-
- Future.firstCompletedOf(Seq(strictResponse, timeout))
+ // Use a Promise-based pattern instead of Future.firstCompletedOf so that
when the
+ // timeout fires first, we properly discard the in-flight response entity
to release
+ // the HTTP connection back to the pool.
+ val promise = Promise[HttpResponse]()
+ val timeoutCancellable =
system.scheduler.scheduleOnce(settings.apiServiceRequestTimeout) {
+ promise.tryFailure(new PodCostTimeoutException(s"$timeoutMsg. Is the API
server up?"))
+ }
+ response.onComplete {
+ case scala.util.Success(resp) =>
+ timeoutCancellable.cancel()
+ if (!promise.trySuccess(resp)) {
+ // Timeout already fired — discard the response to release the
connection
+
resp.discardEntityBytes()(pekko.stream.Materializer.matFromSystem(system))
+ }
+ case scala.util.Failure(ex) =>
+ timeoutCancellable.cancel()
+ promise.tryFailure(ex)
+ }(system.dispatcher)
+ promise.future.flatMap(_.toStrict(settings.bodyReadTimeout))
}
private def toPodCostResource(cr: PodCostCustomResource) = {
@@ -305,8 +318,7 @@ PUTs must contain resourceVersions. Response:
Unmarshal(responseEntity).to[PodCostCustomResource].map(cr =>
Some(toPodCostResource(cr)))
case StatusCodes.Conflict =>
log.debug("creation of PodCost resource failed as already exists.
Will attempt to read again")
- entity.discardBytes()
- // someone else has created it
+ // response entity already consumed by toStrict above; someone else
has created it
Future.successful(None)
case StatusCodes.Unauthorized =>
handleUnauthorized(response)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
