pjfanning opened a new pull request, #905:
URL: https://github.com/apache/pekko-management/pull/905
Summary of Changes
5 files changed, 85 insertions, 42 deletions.
1. rolling-update-kubernetes/.../KubernetesApiImpl.scala
Timeout leak fix: Replaced Future.firstCompletedOf(Seq(strictResponse,
timeout)) in makeRequest() with a Promise-based pattern. When the timeout fires
first, the onComplete callback on the HTTP response discards the entity via
discardEntityBytes(), releasing the connection back to Pekko HTTP's pool.
Removed unused pekko.pattern.after import.
Wrong entity fix: Removed entity.discardBytes() in the StatusCodes.Conflict
branch of createPodCostResource() — it was discarding the request entity
(already sent), not the response (already consumed by toStrict).
2. lease-kubernetes/.../AbstractKubernetesApiImpl.scala
Timeout leak fix: Same Promise-based pattern in makeRequest(). The retry
wrapper (RetrySupport.retry for 401 token rotation) is preserved — the cleanup
runs after the retry chain completes. Removed unused pekko.pattern.after import.
3. discovery-consul/.../ConsulServiceDiscovery.scala
Timeout leak fix: Replaced Future.firstCompletedOf with a Promise-based
pattern using scheduleOnce for the timeout and onComplete to cancel it. When
the Consul client responds, the timer is cancelled; when the timer fires, the
promise fails. Removed unused pekko.pattern.after import.
4. management-cluster-bootstrap/.../HttpContactPointBootstrap.scala
Timeout leak fix: Replaced Future.firstCompletedOf(List(reply,
afterTimeout)) in the ProbeTick handler with a Promise-based pattern. When the
timeout fires first, a promise.future.failed.foreach callback discards the
in-flight response entity. Removed unused pekko.pattern.after import and
replyTimeout val.
5. management/.../HealthChecksImpl.scala
Shared timeout fix: Moved the pekko.pattern.after timeout creation inside
the checks.map loop so each health check gets its own timer instead of sharing
one. Eliminates the timeout.recoverWith wrapper that enriched the generic
RuntimeException — the timeout now throws CheckTimeoutException directly.
Pattern Used
All HTTP-based modules use the same cleanup pattern:
```
val promise = Promise[T]()
val timeoutCancellable = system.scheduler.scheduleOnce(timeout) {
promise.tryFailure(new TimeoutException(...))
}
response.onComplete {
case Success(resp) =>
timeoutCancellable.cancel()
if (!promise.trySuccess(resp)) {
// Timeout already fired — discard the response to release the
connection
resp.discardEntityBytes()(Materializer.matFromSystem(system))
}
case Failure(ex) =>
timeoutCancellable.cancel()
promise.tryFailure(ex)
}(system.dispatcher)
promise.future
```
This ensures the HTTP response entity is always consumed — either by the
normal processing path, or by discardEntityBytes() if the timeout won the race.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
