Stop HealthPoller from polling a closed client after cancellation

[atoivonen]

Issue

The Flux in LoadBalancerBuilder is wrapped in Flux.using, which successfully closes the shared AsyncHttpClient when the subscription is cancelled or terminated. However, HealthPoller.start's recursive poll loop was driven entirely by CompletableFuture callbacks and an independent Mono.delay().subscribe(), meaning it never observed downstream cancellation.

Because of this, when a cancellation occurred, the client closed but the orphaned loop kept polling. This hit the async-http-client's closed-client guard, resulting in an infinite loop of IllegalStateException: Closed errors.

Solution

This PR ties the recursive loop directly to the subscription lifecycle to prevent orphaned polling:

• Short-circuit recursion: Checks emitter.isCancelled() before each request to halt the loop.
• Halt on completion: Returns immediately without logging or rescheduling if the subscription is cancelled in whenComplete.
• Clear dangling timers: Routes the 5-second exception-backoff (Mono.delay) through a Disposable.Swap registered with emitter.onDispose. If a cancellation occurs during the backoff window, it cancels the pending restart instead of leaving a dangling timer.

Testing

• Added test coverage for cancellation while a request is in flight.
• Added test coverage for cancellation during the exception backoff (driven with VirtualTimeScheduler).
• Added reactor-test as a test dependency.

[bpholt]

Nit: could this be final?

[bpholt]

Just to confirm my understanding: if an issue comes up while long polling Consul, the retry delay will be registered with this backoff. Normally it will be retried after the delay; however, if the overall Flux is closed before the delay completes, the retry will also be canceled (via a call to backoff.dispose() inside onDispose).

Sanity check: what happens if multiple retries are necessary? It will just keep updating backup with new delays?

[atoivonen]

Each poll failure registers a 5s retry via backoff.update(...). If the Flux is cancelled or terminated early, emitter.onDispose(backoff) handles disposing of it so the pending retry never fires.

Great question regarding the multiple retries. Yeah, it will keep updating backoff with the newest delay. A new delay is only scheduled after the previous one fires and fails again, so we will never have more than one pending delay at a time. Disposable.Swap holds a single slot, .update() safely clears out the old (already completed) delay and stores the new one. backoff is always tracking just that one in-flight retry in case it needs to be cancelled.

[bpholt]

Why do we return a completed future containing index and not e.g. a failed future with a CancelationException?

[atoivonen]

I went with a completed future because the returned future is never actually used. The recursion is driven by the whenComplete callback calling r.func.apply(...), and all three call discard the returned CompletableFuture. It really only exists to satisfy the Function<Integer, CompletableFuture<Integer>> signature.

It felt like the most straightforward no-op: basically cleanly stopping because the subscription is gone.

[bpholt]

Gotcha, that makes sense.

I think it might be more clear to use CompletableFuture.failedFuture(new CancellationException()) instead just to avoid any doubt that the future is not used for anything in the normal code path, but it's not a huge deal either way.

[bpholt]

Overall I think this looks good but I added a couple of questions just to make sure I'm understanding it correctly, since I'm not super familiar with the Flux/Reactor API.

[bpholt]

Couple of nits that could be picked, but otherwise it looks good. I'll approve and leave the decision of whether to adjust those things up to you.

[mergify]

Tick the box to add this pull request to the merge queue (same as @mergifyio queue).

• Queue this pull request