Merge branch '3.1.x' into 3.2.x

Closes gh-39444

Author: scottfrederick

spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/endpoint/web/reactive/AbstractWebFluxEndpointHandlerMapping.java

@@ -83,6 +83,7 @@
  * @author Madhura Bhave
  * @author Phillip Webb
  * @author Brian Clozel
+ * @author Scott Frederick
  * @since 2.0.0
  */
 @ImportRuntimeHints(AbstractWebFluxEndpointHandlerMappingRuntimeHints.class)
@@ -260,6 +261,26 @@ public Object invoke(InvocationContext context) {
 
 	}
 
+	protected static final class ExceptionCapturingInvoker implements OperationInvoker {
+
+		private final OperationInvoker invoker;
+
+		public ExceptionCapturingInvoker(OperationInvoker invoker) {
+			this.invoker = invoker;
+		}
+
+		@Override
+		public Object invoke(InvocationContext context) {
+			try {
+				return this.invoker.invoke(context);
+			}
+			catch (Exception ex) {
+				return Mono.error(ex);
+			}
+		}
+
+	}
+
 	/**
 	 * Reactive handler providing actuator links at the root endpoint.
 	 */
@@ -303,9 +324,9 @@ private ReactiveWebOperationAdapter(WebOperation operation) {
 		private OperationInvoker getInvoker(WebOperation operation) {
 			OperationInvoker invoker = operation::invoke;
 			if (operation.isBlocking()) {
-				invoker = new ElasticSchedulerInvoker(invoker);
+				return new ElasticSchedulerInvoker(invoker);
 			}
-			return invoker;
+			return new ExceptionCapturingInvoker(invoker);
 		}
 
 		private Supplier<Mono<? extends SecurityContext>> getSecurityContextSupplier() {

spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/endpoint/web/annotation/AbstractWebEndpointIntegrationTests.java

@@ -226,6 +226,31 @@ void readOperationWithSingleQueryParameters() {
 					.isEqualTo("1 2"));
 	}
 
+	@Test
+	void readOperationWithQueryParametersMissing() {
+		load(QueryEndpointConfiguration.class,
+				(client) -> client.get().uri("/query").exchange().expectStatus().isBadRequest());
+	}
+
+	@Test
+	void reactiveReadOperationWithSingleQueryParameters() {
+		load(ReactiveQueryEndpointConfiguration.class,
+				(client) -> client.get()
+					.uri("/query?param=test")
+					.exchange()
+					.expectStatus()
+					.isOk()
+					.expectBody()
+					.jsonPath("query")
+					.isEqualTo("test"));
+	}
+
+	@Test
+	void reactiveReadOperationWithQueryParametersMissing() {
+		load(ReactiveQueryEndpointConfiguration.class,
+				(client) -> client.get().uri("/query").exchange().expectStatus().isBadRequest());
+	}
+
 	@Test
 	void readOperationWithSingleQueryParametersAndMultipleValues() {
 		load(QueryEndpointConfiguration.class,
@@ -732,6 +757,17 @@ QueryWithListEndpoint queryEndpoint() {
 
 	}
 
+	@Configuration(proxyBeanMethods = false)
+	@Import(BaseConfiguration.class)
+	static class ReactiveQueryEndpointConfiguration {
+
+		@Bean
+		ReactiveQueryEndpoint reactiveQueryEndpoint() {
+			return new ReactiveQueryEndpoint();
+		}
+
+	}
+
 	@Configuration(proxyBeanMethods = false)
 	@Import(BaseConfiguration.class)
 	static class VoidWriteResponseEndpointConfiguration {
@@ -974,6 +1010,16 @@ Map<String, String> queryWithParameterList(String one, List<String> two) {
 
 	}
 
+	@Endpoint(id = "query")
+	static class ReactiveQueryEndpoint {
+
+		@ReadOperation
+		Mono<Map<String, String>> query(String param) {
+			return Mono.just(Collections.singletonMap("query", param));
+		}
+
+	}
+
 	@Endpoint(id = "voidwrite")
 	static class VoidWriteResponseEndpoint {