Clear context before flux retry #8456
Conversation
| Flux.defer(() -> publish.delaySubscription(Duration.ofMillis(1))) | ||
| .doOnNext( | ||
| message -> { | ||
| tracer.spanBuilder("process " + message).startSpan().end(); | ||
| }) | ||
| .handle( | ||
| (message, sink) -> { | ||
| if (message == 1 && beforeRetry.compareAndSet(true, false)) { | ||
| sink.error(new RuntimeException("Message has error")); | ||
| } else { | ||
| sink.next(message); | ||
| } | ||
| }) | ||
| .retry() | ||
| .subscribe(); |
There was a problem hiding this comment.
can you add a test where there's an active parent span when .subscribe() is called?
There was a problem hiding this comment.
I added a test with parent span, all child spans are under that parent.
|
cc @lmolkova |
There was a problem hiding this comment.
Thank you!
On a side note, I think we need to revisit context propagation in Reactor. It seems by propagating context beyond current operation (into consequent operations such as flatMap), we're leaking it.
I did not have a chance to investigate deeply, but we're seeing issues where all operations happening in background threads (where there is no incoming request or another parent) are correlated to something random happening on this thread (e.g. Azure/azure-sdk-for-java#32807).
I want to look into new reactor context propagation helpers once I get a chance, but it does not block this PR :)
Resolves #8454
Flux retry calls subscribe from
onErrorwhere we might have active context. We need to clear the context before subscribe is called to avoid having retried operations run with that context. This pr handles handlesFlux.retryandFlux.retryWhenit is possible that there are more methods with he same issue.