fix tests + code adjustment

Signed-off-by: wind57 <[email protected]>

Author: wind57

spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionInitiator.java

@@ -31,9 +31,10 @@
 import org.springframework.core.log.LogAccessor;
 
 import static java.util.concurrent.Executors.newSingleThreadExecutor;
-import static org.springframework.cloud.kubernetes.fabric8.leader.election.Fabric8LeaderElectionInitiatorUtil.attachStatusLoggerPipeline;
+import static org.springframework.cloud.kubernetes.fabric8.leader.election.Fabric8LeaderElectionInitiatorUtil.attachReadinessLoggerPipeline;
 import static org.springframework.cloud.kubernetes.fabric8.leader.election.Fabric8LeaderElectionInitiatorUtil.blockReadinessCheck;
 import static org.springframework.cloud.kubernetes.fabric8.leader.election.Fabric8LeaderElectionInitiatorUtil.leaderElector;
+import static org.springframework.cloud.kubernetes.fabric8.leader.election.Fabric8LeaderElectionInitiatorUtil.shutDownExecutor;
 import static org.springframework.cloud.kubernetes.fabric8.leader.election.Fabric8LeaderElectionInitiatorUtil.sleep;
 
 /**
@@ -103,13 +104,18 @@ void postConstruct() {
 		// wait in a different thread until the pod is ready
 		// and don't block the main application from starting
 		podReadyWaitingExecutor.execute(() -> {
-			if (waitForPodReady) {
-				CompletableFuture<?> ready = attachStatusLoggerPipeline(podReadyFuture, candidateIdentity);
-				blockReadinessCheck(ready);
-				startLeaderElection();
+			try {
+				if (waitForPodReady) {
+					CompletableFuture<?> ready = attachReadinessLoggerPipeline(podReadyFuture, candidateIdentity);
+					blockReadinessCheck(ready);
+					startLeaderElection();
+				}
+				else {
+					startLeaderElection();
+				}
 			}
-			else {
-				startLeaderElection();
+			catch (Exception e) {
+				LOG.error(e, () -> "failure : " + e.getMessage());
 			}
 		});
 
@@ -119,11 +125,6 @@ void postConstruct() {
 	void preDestroy() {
 		LOG.info(() -> "preDestroy called on the leader initiator : " + candidateIdentity);
 
-		if (!podReadyWaitingExecutor.isShutdown()) {
-			LOG.debug(() -> "podReadyWaitingExecutor will be shutdown for : " + candidateIdentity);
-			podReadyWaitingExecutor.shutdownNow();
-		}
-
 		if (podReadyFuture != null && !podReadyFuture.isDone()) {
 			// if the task is not running, this has no effect.
 			// if the task is running, calling this will also make sure
@@ -132,6 +133,10 @@ void preDestroy() {
 			podReadyFuture.cancel(true);
 		}
 
+		if (!podReadyWaitingExecutor.isShutdown()) {
+			shutDownExecutor(podReadyWaitingExecutor, candidateIdentity);
+		}
+
 		if (leaderFuture != null) {
 			// needed to release the lock, in case we are holding it.
 			// fabric8 internally expects this one to be called
@@ -153,8 +158,7 @@ private void startLeaderElection() {
 
 			if (error != null) {
 				// only we have a reference to leaderFuture; and if it is canceled, it is
-				// only possible
-				// from our own preDestroy
+				// only possible from our own preDestroy
 				if (error instanceof CancellationException) {
 					LOG.info(() -> "cancel was called on the leader initiator : " + candidateIdentity);
 					LOG.info(() -> "terminating leadership for : " + candidateIdentity);

spring-cloud-kubernetes-fabric8-leader/src/main/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionInitiatorUtil.java

@@ -17,6 +17,7 @@
 package org.springframework.cloud.kubernetes.fabric8.leader.election;
 
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 
 import io.fabric8.kubernetes.client.KubernetesClient;
@@ -38,7 +39,7 @@ private Fabric8LeaderElectionInitiatorUtil() {
 	 * if 'ready' is already completed at this point, thread will run this, otherwise it
 	 * will attach the pipeline and move on to 'blockReadinessCheck'.
 	 */
-	static CompletableFuture<?> attachStatusLoggerPipeline(CompletableFuture<?> innerPodReadyFuture,
+	static CompletableFuture<?> attachReadinessLoggerPipeline(CompletableFuture<?> innerPodReadyFuture,
 			String candidateIdentity) {
 		return innerPodReadyFuture.whenComplete((ok, error) -> {
 			if (error != null) {
@@ -69,6 +70,18 @@ static void blockReadinessCheck(CompletableFuture<?> ready) {
 		}
 		catch (Exception e) {
 			LOG.error(e, () -> "block readiness check failed with : " + e.getMessage());
+			throw new RuntimeException(e);
+		}
+	}
+
+	static void shutDownExecutor(ExecutorService podReadyWaitingExecutor, String candidateIdentity) {
+		LOG.debug(() -> "podReadyWaitingExecutor will be shutdown for : " + candidateIdentity);
+		podReadyWaitingExecutor.shutdownNow();
+		try {
+			podReadyWaitingExecutor.awaitTermination(3, TimeUnit.SECONDS);
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
 		}
 	}
 

spring-cloud-kubernetes-integration-tests/spring-cloud-kubernetes-fabric8-leader-election/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/AbstractLeaderElection.java

@@ -24,6 +24,7 @@
 import io.fabric8.kubernetes.client.Config;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.kubernetes.client.KubernetesClientBuilder;
+import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
@@ -56,7 +57,7 @@ abstract class AbstractLeaderElection {
 
 	private static K3sContainer container;
 
-	private static final MockedStatic<LeaderUtils> LEADER_UTILS_MOCKED_STATIC = Mockito.mockStatic(LeaderUtils.class);
+	private static MockedStatic<LeaderUtils> LEADER_UTILS_MOCKED_STATIC;
 
 	@Autowired
 	KubernetesClient kubernetesClient;
@@ -65,9 +66,15 @@ static void beforeAll(String candidateIdentity) {
 		container = Commons.container();
 		container.start();
 
+		LEADER_UTILS_MOCKED_STATIC = Mockito.mockStatic(LeaderUtils.class);
 		LEADER_UTILS_MOCKED_STATIC.when(LeaderUtils::hostName).thenReturn(candidateIdentity);
 	}
 
+	@AfterAll
+	static void afterAll() {
+		LEADER_UTILS_MOCKED_STATIC.close();
+	}
+
 	void stopFutureAndDeleteLease(Fabric8LeaderElectionInitiator initiator) {
 
 		initiator.preDestroy();
@@ -124,11 +131,11 @@ BooleanSupplier readinessSupplierFails() {
 			};
 		}
 
-		// readiness fails after 2 retries
+		// readiness always fails
 		@Bean
 		@Primary
-		@ConditionalOnProperty(value = "readiness.cycle.false", havingValue = "true", matchIfMissing = false)
-		BooleanSupplier readinessCycleFalse() {
+		@ConditionalOnProperty(value = "readiness.never.finishes", havingValue = "true", matchIfMissing = false)
+		BooleanSupplier readinessNeverFinishes() {
 			return () -> false;
 		}
 

spring-cloud-kubernetes-integration-tests/spring-cloud-kubernetes-fabric8-leader-election/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Assertions.java

@@ -53,12 +53,14 @@ static void assertAcquireAndRenew(CapturedOutput output, Supplier<Lease> leaseSu
 					+ candidateIdentity + ")'"));
 
 		// 4. lease has been acquired
-		awaitUntil(5, 100, () -> output.getOut().contains("Acquired lease 'LeaseLock: default - spring-k8s-leader-election-lock "
-				+ "(" + candidateIdentity + ")'"));
+		awaitUntil(5, 100,
+				() -> output.getOut()
+					.contains("Acquired lease 'LeaseLock: default - spring-k8s-leader-election-lock " + "("
+							+ candidateIdentity + ")'"));
 
 		// 5. we are the leader (comes from fabric8 code)
-		awaitUntil(5, 100, () -> output.getOut()
-			.matches("(?s).*Leader changed from (|null) to " + candidateIdentity + ".*"));
+		awaitUntil(5, 100,
+				() -> output.getOut().matches("(?s).*Leader changed from (|null) to " + candidateIdentity + ".*"));
 
 		// 6. wait until a renewal happens (comes from fabric code)
 		// this one means that we have extended our leadership

spring-cloud-kubernetes-integration-tests/spring-cloud-kubernetes-fabric8-leader-election/src/test/java/org/springframework/cloud/kubernetes/fabric8/leader/election/Fabric8LeaderElectionReadinessCanceledIT.java

@@ -32,7 +32,7 @@
  *
  * @author wind57
  */
-@TestPropertySource(properties = { "readiness.cycle.false=true",
+@TestPropertySource(properties = { "readiness.never.finishes=true",
 		"spring.cloud.kubernetes.leader.election.wait-for-pod-ready=true" })
 class Fabric8LeaderElectionReadinessCanceledIT extends AbstractLeaderElection {
 
@@ -59,13 +59,6 @@ void test(CapturedOutput output) {
 
 		initiator.preDestroy();
 
-		try {
-			Thread.sleep(2_000);
-		}
-		catch (InterruptedException e) {
-			throw new RuntimeException(e);
-		}
-
 		// 1. preDestroy method logs what it will do
 		assertThat(output.getOut()).contains("podReadyFuture will be canceled for : canceled-readiness-it");
 
@@ -82,6 +75,10 @@ void test(CapturedOutput output) {
 		// 5. the scheduled executor where pod readiness is checked is shut down also
 		awaitUntil(2, 100, () -> output.getOut().contains("Shutting down executor : podReadyExecutor"));
 
+		// we need to call preDestroy again, to make sure that leaderFuture was not
+		// started
+		initiator.preDestroy();
+
 		// 6. leader election is not started, since readiness does not finish
 		assertThat(output.getOut()).doesNotContain("leaderFuture will be canceled for");