Polish "Replace calls to verifyComplete() to avoid indefinite blocking"

See gh-35915

Author: wilkinsona

buildSrc/src/main/java/org/springframework/boot/build/architecture/ArchitectureCheck.java

@@ -74,7 +74,9 @@ void checkArchitecture() throws IOException {
 			.importPaths(this.classes.getFiles().stream().map(File::toPath).collect(Collectors.toList()));
 		List<EvaluationResult> violations = Stream.of(allPackagesShouldBeFreeOfTangles(),
 				allBeanPostProcessorBeanMethodsShouldBeStaticAndHaveParametersThatWillNotCausePrematureInitialization(),
-				allBeanFactoryPostProcessorBeanMethodsShouldBeStaticAndHaveNoParameters())
+				allBeanFactoryPostProcessorBeanMethodsShouldBeStaticAndHaveNoParameters(),
+				noClassesShouldCallStepVerifierStepVerifyComplete(),
+				noClassesShouldConfigureDefaultStepVerifierTimeout())
 			.map((rule) -> rule.evaluate(javaClasses))
 			.filter(EvaluationResult::hasViolation)
 			.collect(Collectors.toList());
@@ -163,6 +165,20 @@ public void check(JavaMethod item, ConditionEvents events) {
 		};
 	}
 
+	private ArchRule noClassesShouldCallStepVerifierStepVerifyComplete() {
+		return ArchRuleDefinition.noClasses()
+			.should()
+			.callMethod("reactor.test.StepVerifier$Step", "verifyComplete")
+			.because("it can block indefinitely and expectComplete().verify(Duration) should be used instead");
+	}
+
+	private ArchRule noClassesShouldConfigureDefaultStepVerifierTimeout() {
+		return ArchRuleDefinition.noClasses()
+			.should()
+			.callMethod("reactor.test.StepVerifier", "setDefaultTimeout", "java.time.Duration")
+			.because("expectComplete().verify(Duration) should be used instead");
+	}
+
 	public void setClasses(FileCollection classes) {
 		this.classes = classes;
 	}

spring-boot-project/spring-boot-actuator-autoconfigure/src/test/java/org/springframework/boot/actuate/autoconfigure/cloudfoundry/reactive/ReactiveCloudFoundrySecurityInterceptorTests.java

@@ -16,6 +16,8 @@
 
 package org.springframework.boot.actuate.autoconfigure.cloudfoundry.reactive;
 
+import java.time.Duration;
+
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -33,8 +35,6 @@
 import org.springframework.mock.web.server.MockServerWebExchange;
 import org.springframework.util.Base64Utils;
 
-import java.time.Duration;
-
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.BDDMockito.given;
@@ -68,7 +68,8 @@ void preHandleWhenRequestIsPreFlightShouldBeOk() {
 			.build());
 		StepVerifier.create(this.interceptor.preHandle(request, "/a"))
 			.consumeNextWith((response) -> assertThat(response.getStatus()).isEqualTo(HttpStatus.OK))
-			.expectComplete().verify(Duration.ofSeconds(5));
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 	}
 
 	@Test
@@ -77,7 +78,8 @@ void preHandleWhenTokenIsMissingShouldReturnMissingAuthorization() {
 		StepVerifier.create(this.interceptor.preHandle(request, "/a"))
 			.consumeNextWith(
 					(response) -> assertThat(response.getStatus()).isEqualTo(Reason.MISSING_AUTHORIZATION.getStatus()))
-			.expectComplete().verify(Duration.ofSeconds(5));
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 	}
 
 	@Test
@@ -87,7 +89,8 @@ void preHandleWhenTokenIsNotBearerShouldReturnMissingAuthorization() {
 		StepVerifier.create(this.interceptor.preHandle(request, "/a"))
 			.consumeNextWith(
 					(response) -> assertThat(response.getStatus()).isEqualTo(Reason.MISSING_AUTHORIZATION.getStatus()))
-			.expectComplete().verify(Duration.ofSeconds(5));
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 	}
 
 	@Test
@@ -123,7 +126,8 @@ void preHandleWhenAccessIsNotAllowedShouldReturnAccessDenied() {
 			.build());
 		StepVerifier.create(this.interceptor.preHandle(request, "/a"))
 			.consumeNextWith((response) -> assertThat(response.getStatus()).isEqualTo(Reason.ACCESS_DENIED.getStatus()))
-			.expectComplete().verify(Duration.ofSeconds(5));
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 	}
 
 	@Test
@@ -137,7 +141,7 @@ void preHandleSuccessfulWithFullAccess() {
 		StepVerifier.create(this.interceptor.preHandle(exchange, "/a")).consumeNextWith((response) -> {
 			assertThat(response.getStatus()).isEqualTo(HttpStatus.OK);
 			assertThat((AccessLevel) exchange.getAttribute("cloudFoundryAccessLevel")).isEqualTo(AccessLevel.FULL);
-		}).expectComplete().verify(Duration.ofSeconds(5));
+		}).expectComplete().verify(Duration.ofSeconds(30));
 	}
 
 	@Test
@@ -153,7 +157,7 @@ void preHandleSuccessfulWithRestrictedAccess() {
 			assertThat(response.getStatus()).isEqualTo(HttpStatus.OK);
 			assertThat((AccessLevel) exchange.getAttribute("cloudFoundryAccessLevel"))
 				.isEqualTo(AccessLevel.RESTRICTED);
-		}).expectComplete().verify(Duration.ofSeconds(5));
+		}).expectComplete().verify(Duration.ofSeconds(30));
 	}
 
 	private String mockAccessToken() {

spring-boot-project/spring-boot-actuator-autoconfigure/src/test/java/org/springframework/boot/actuate/autoconfigure/cloudfoundry/reactive/ReactiveTokenValidatorTests.java

@@ -122,7 +122,8 @@ void validateTokenWhenKidValidationSucceedsInTheSecondAttempt() throws Exception
 		String claims = "{\"exp\": 2147483647, \"iss\": \"http://localhost:8080/uaa/oauth/token\", \"scope\": [\"actuator.read\"]}";
 		StepVerifier
 			.create(this.tokenValidator.validate(new Token(getSignedToken(header.getBytes(), claims.getBytes()))))
-			.expectComplete().verify(Duration.ofSeconds(5));
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 		assertThat(this.tokenValidator).hasFieldOrPropertyWithValue("cachedTokenKeys", VALID_KEYS);
 		fetchTokenKeys.assertWasSubscribed();
 	}
@@ -136,7 +137,8 @@ void validateTokenWhenCacheIsEmptyShouldFetchTokenKeys() throws Exception {
 		String claims = "{\"exp\": 2147483647, \"iss\": \"http://localhost:8080/uaa/oauth/token\", \"scope\": [\"actuator.read\"]}";
 		StepVerifier
 			.create(this.tokenValidator.validate(new Token(getSignedToken(header.getBytes(), claims.getBytes()))))
-			.expectComplete().verify(Duration.ofSeconds(5));
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 		assertThat(this.tokenValidator).hasFieldOrPropertyWithValue("cachedTokenKeys", VALID_KEYS);
 		fetchTokenKeys.assertWasSubscribed();
 	}
@@ -168,7 +170,8 @@ void validateTokenWhenCacheValidShouldNotFetchTokenKeys() throws Exception {
 		String claims = "{\"exp\": 2147483647, \"iss\": \"http://localhost:8080/uaa/oauth/token\", \"scope\": [\"actuator.read\"]}";
 		StepVerifier
 			.create(this.tokenValidator.validate(new Token(getSignedToken(header.getBytes(), claims.getBytes()))))
-			.expectComplete().verify(Duration.ofSeconds(5));
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 		fetchTokenKeys.assertWasNotSubscribed();
 	}
 

spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/cassandra/CassandraDriverReactiveHealthIndicatorTests.java

@@ -62,7 +62,8 @@ void healthWithOneHealthyNodeShouldReturnUp() {
 		Mono<Health> health = healthIndicator.health();
 		StepVerifier.create(health)
 			.consumeNextWith((h) -> assertThat(h.getStatus()).isEqualTo(Status.UP))
-			.expectComplete().verify(Duration.ofSeconds(5));
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 	}
 
 	@Test
@@ -72,7 +73,8 @@ void healthWithOneUnhealthyNodeShouldReturnDown() {
 		Mono<Health> health = healthIndicator.health();
 		StepVerifier.create(health)
 			.consumeNextWith((h) -> assertThat(h.getStatus()).isEqualTo(Status.DOWN))
-			.expectComplete().verify(Duration.ofSeconds(5));
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 	}
 
 	@Test
@@ -82,7 +84,8 @@ void healthWithOneUnknownNodeShouldReturnDown() {
 		Mono<Health> health = healthIndicator.health();
 		StepVerifier.create(health)
 			.consumeNextWith((h) -> assertThat(h.getStatus()).isEqualTo(Status.DOWN))
-			.expectComplete().verify(Duration.ofSeconds(5));
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 	}
 
 	@Test
@@ -92,7 +95,8 @@ void healthWithOneForcedDownNodeShouldReturnDown() {
 		Mono<Health> health = healthIndicator.health();
 		StepVerifier.create(health)
 			.consumeNextWith((h) -> assertThat(h.getStatus()).isEqualTo(Status.DOWN))
-			.expectComplete().verify(Duration.ofSeconds(5));
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 	}
 
 	@Test
@@ -102,7 +106,8 @@ void healthWithOneHealthyNodeAndOneUnhealthyNodeShouldReturnUp() {
 		Mono<Health> health = healthIndicator.health();
 		StepVerifier.create(health)
 			.consumeNextWith((h) -> assertThat(h.getStatus()).isEqualTo(Status.UP))
-			.expectComplete().verify(Duration.ofSeconds(5));
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 	}
 
 	@Test
@@ -112,7 +117,8 @@ void healthWithOneHealthyNodeAndOneUnknownNodeShouldReturnUp() {
 		Mono<Health> health = healthIndicator.health();
 		StepVerifier.create(health)
 			.consumeNextWith((h) -> assertThat(h.getStatus()).isEqualTo(Status.UP))
-			.expectComplete().verify(Duration.ofSeconds(5));
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 	}
 
 	@Test
@@ -122,7 +128,8 @@ void healthWithOneHealthyNodeAndOneForcedDownNodeShouldReturnUp() {
 		Mono<Health> health = healthIndicator.health();
 		StepVerifier.create(health)
 			.consumeNextWith((h) -> assertThat(h.getStatus()).isEqualTo(Status.UP))
-			.expectComplete().verify(Duration.ofSeconds(5));
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 	}
 
 	@Test
@@ -140,7 +147,7 @@ void healthWithNodeVersionShouldAddVersionDetail() {
 			assertThat(h.getStatus()).isEqualTo(Status.UP);
 			assertThat(h.getDetails()).containsOnlyKeys("version");
 			assertThat(h.getDetails().get("version")).isEqualTo(Version.V4_0_0);
-		}).expectComplete().verify(Duration.ofSeconds(5));
+		}).expectComplete().verify(Duration.ofSeconds(30));
 	}
 
 	@Test
@@ -151,7 +158,7 @@ void healthWithoutNodeVersionShouldNotAddVersionDetail() {
 		StepVerifier.create(health).consumeNextWith((h) -> {
 			assertThat(h.getStatus()).isEqualTo(Status.UP);
 			assertThat(h.getDetails().get("version")).isNull();
-		}).expectComplete().verify(Duration.ofSeconds(5));
+		}).expectComplete().verify(Duration.ofSeconds(30));
 	}
 
 	@Test
@@ -166,7 +173,7 @@ void healthWithCassandraDownShouldReturnDown() {
 			assertThat(h.getDetails()).containsOnlyKeys("error");
 			assertThat(h.getDetails().get("error"))
 				.isEqualTo(DriverTimeoutException.class.getName() + ": Test Exception");
-		}).expectComplete().verify(Duration.ofSeconds(5));
+		}).expectComplete().verify(Duration.ofSeconds(30));
 	}
 
 	private CqlSession mockCqlSessionWithNodeState(NodeState... nodeStates) {

spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/health/HealthIndicatorReactiveAdapterTests.java

@@ -37,7 +37,7 @@ void delegateReturnsHealth() {
 		HealthIndicatorReactiveAdapter adapter = new HealthIndicatorReactiveAdapter(delegate);
 		Health status = Health.up().build();
 		given(delegate.health()).willReturn(status);
-		StepVerifier.create(adapter.health()).expectNext(status).expectComplete().verify(Duration.ofSeconds(5));
+		StepVerifier.create(adapter.health()).expectNext(status).expectComplete().verify(Duration.ofSeconds(30));
 	}
 
 	@Test
@@ -55,7 +55,10 @@ void delegateRunsOnTheElasticScheduler() {
 			.status(Thread.currentThread().getName().equals(currentThread) ? Status.DOWN : Status.UP)
 			.build();
 		HealthIndicatorReactiveAdapter adapter = new HealthIndicatorReactiveAdapter(delegate);
-		StepVerifier.create(adapter.health()).expectNext(Health.status(Status.UP).build()).expectComplete().verify(Duration.ofSeconds(5));
+		StepVerifier.create(adapter.health())
+			.expectNext(Health.status(Status.UP).build())
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 	}
 
 }

spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/health/ReactiveHealthIndicatorImplementationTests.java

@@ -16,6 +16,8 @@
 
 package org.springframework.boot.actuate.health;
 
+import java.time.Duration;
+
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import reactor.core.publisher.Mono;
@@ -25,8 +27,6 @@
 import org.springframework.boot.test.system.CapturedOutput;
 import org.springframework.boot.test.system.OutputCaptureExtension;
 
-import java.time.Duration;
-
 import static org.assertj.core.api.Assertions.assertThat;
 
 /**
@@ -42,7 +42,8 @@ class ReactiveHealthIndicatorImplementationTests {
 	void healthUp(CapturedOutput output) {
 		StepVerifier.create(new SimpleReactiveHealthIndicator().health())
 			.consumeNextWith((health) -> assertThat(health).isEqualTo(Health.up().build()))
-			.expectComplete().verify(Duration.ofSeconds(5));
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 		assertThat(output).doesNotContain("Health check failed for simple");
 	}
 
@@ -51,15 +52,17 @@ void healthDownWithCustomErrorMessage(CapturedOutput output) {
 		StepVerifier.create(new CustomErrorMessageReactiveHealthIndicator().health())
 			.consumeNextWith(
 					(health) -> assertThat(health).isEqualTo(Health.down(new UnsupportedOperationException()).build()))
-			.expectComplete().verify(Duration.ofSeconds(5));
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 		assertThat(output).contains("Health check failed for custom");
 	}
 
 	@Test
 	void healthDownWithCustomErrorMessageFunction(CapturedOutput output) {
 		StepVerifier.create(new CustomErrorMessageFunctionReactiveHealthIndicator().health())
 			.consumeNextWith((health) -> assertThat(health).isEqualTo(Health.down(new RuntimeException()).build()))
-			.expectComplete().verify(Duration.ofSeconds(5));
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 		assertThat(output).contains("Health check failed with RuntimeException");
 	}
 

spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/metrics/r2dbc/ConnectionPoolMetricsTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2021 the original author or authors.
+ * Copyright 2012-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.
@@ -60,7 +60,7 @@ void init() {
 	@AfterEach
 	void close() {
 		if (this.connectionFactory != null) {
-			StepVerifier.create(this.connectionFactory.close()).expectComplete().verify(Duration.ofSeconds(5));
+			StepVerifier.create(this.connectionFactory.close()).expectComplete().verify(Duration.ofSeconds(30));
 		}
 	}
 
@@ -73,8 +73,16 @@ void connectionFactoryIsInstrumented() {
 				Tags.of(testTag, regionTag));
 		metrics.bindTo(registry);
 		// acquire two connections
-		connectionPool.create().as(StepVerifier::create).expectNextCount(1).expectComplete().verify(Duration.ofSeconds(5));
-		connectionPool.create().as(StepVerifier::create).expectNextCount(1).expectComplete().verify(Duration.ofSeconds(5));
+		connectionPool.create()
+			.as(StepVerifier::create)
+			.expectNextCount(1)
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
+		connectionPool.create()
+			.as(StepVerifier::create)
+			.expectNextCount(1)
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
 		assertGauge(registry, "r2dbc.pool.acquired", 2);
 		assertGauge(registry, "r2dbc.pool.allocated", 3);
 		assertGauge(registry, "r2dbc.pool.idle", 1);

spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/metrics/web/reactive/server/MetricsWebFilterTests.java

@@ -162,7 +162,7 @@ void disconnectedExceptionShouldProduceMetrics() {
 				exchange.getResponse().setRawStatusCode(500);
 				return exchange.getResponse().setComplete();
 			});
-		StepVerifier.create(processing).expectComplete().verify(Duration.ofSeconds(5));
+		StepVerifier.create(processing).expectComplete().verify(Duration.ofSeconds(30));
 		assertMetricsContainsTag("uri", "/projects/{project}");
 		assertMetricsContainsTag("status", "500");
 		assertMetricsContainsTag("outcome", "UNKNOWN");

spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/mongo/MongoReactiveHealthIndicatorTests.java

@@ -16,6 +16,8 @@
 
 package org.springframework.boot.actuate.mongo;
 
+import java.time.Duration;
+
 import com.mongodb.MongoException;
 import org.bson.Document;
 import org.junit.jupiter.api.Test;
@@ -26,8 +28,6 @@
 import org.springframework.boot.actuate.health.Status;
 import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
 
-import java.time.Duration;
-
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.Mockito.mock;
@@ -52,7 +52,7 @@ void testMongoIsUp() {
 			assertThat(h.getStatus()).isEqualTo(Status.UP);
 			assertThat(h.getDetails()).containsOnlyKeys("version");
 			assertThat(h.getDetails().get("version")).isEqualTo("2.6.4");
-		}).expectComplete().verify(Duration.ofSeconds(5));
+		}).expectComplete().verify(Duration.ofSeconds(30));
 	}
 
 	@Test
@@ -67,7 +67,7 @@ void testMongoIsDown() {
 			assertThat(h.getStatus()).isEqualTo(Status.DOWN);
 			assertThat(h.getDetails()).containsOnlyKeys("error");
 			assertThat(h.getDetails().get("error")).isEqualTo(MongoException.class.getName() + ": Connection failed");
-		}).expectComplete().verify(Duration.ofSeconds(5));
+		}).expectComplete().verify(Duration.ofSeconds(30));
 	}
 
 }