Set WebClient Observation as current in reactor context

Prior to this commit, the `DefaultWebClient` would be instrumented for
client observations and would start/stop a `"http.client.requests"`
observation. This would not set this new observation as the current one
in the Reactor context under `ObservationThreadLocalAccessor.KEY`.
This means that potential child observations would not detect it as
their parent; this can happen if the Reactor Netty `HttpClient`
observation is enabled.

This commit ensures that the reactor context is properly populated for
upstream operators.

Fixes gh-29891

spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -478,7 +478,8 @@ class DefaultWebClient implements WebClient {
 							observationContext.setAborted(true);
 							observation.stop();
 						})
-						.doOnTerminate(observation::stop);
+						.doOnTerminate(observation::stop)
+						.contextWrite(context -> context.put(ObservationThreadLocalAccessor.KEY, observation));
 			});
 		}
 

spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientObservationTests.java

@@ -114,6 +114,25 @@ class WebClientObservationTests {
 				.hasLowCardinalityKeyValue("status", "CLIENT_ERROR");
 	}
 
+	@Test
+	void setsCurrentObservationInReactorContext() {
+		ExchangeFilterFunction assertionFilter = new ExchangeFilterFunction() {
+			@Override
+			public Mono<ClientResponse> filter(ClientRequest request, ExchangeFunction chain) {
+				return chain.exchange(request).contextWrite(context -> {
+					Observation currentObservation = context.get(ObservationThreadLocalAccessor.KEY);
+					assertThat(currentObservation).isNotNull();
+					assertThat(currentObservation.getContext()).isInstanceOf(ClientRequestObservationContext.class);
+					return context;
+				});
+			}
+		};
+		this.builder.filter(assertionFilter).build().get().uri("/resource/{id}", 42)
+				.retrieve().bodyToMono(Void.class)
+				.block(Duration.ofSeconds(10));
+			verifyAndGetRequest();
+	}
+
 	private TestObservationRegistryAssert.TestObservationRegistryAssertReturningObservationContextAssert assertThatHttpObservation() {
 		return TestObservationRegistryAssert.assertThat(this.observationRegistry)
 				.hasObservationWithNameEqualTo("http.client.requests").that();