Skip to content

Commit 29af71d

Browse files
wind57wind57
committed
review comments
Signed-off-by: wind57 <[email protected]>
1 parent ee348f4 commit 29af71d

File tree

3 files changed

+179
-26
lines changed

3 files changed

+179
-26
lines changed

‎spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/CachedSingleThreadScheduler.java‎

Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,102 @@
1+
/*
2+
* Copyright 2013-2025 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.cloud.kubernetes.commons;
18+
19+
import java.util.concurrent.Executors;
20+
import java.util.concurrent.ScheduledFuture;
21+
import java.util.concurrent.ScheduledThreadPoolExecutor;
22+
import java.util.concurrent.ThreadFactory;
23+
import java.util.concurrent.TimeUnit;
24+
import java.util.concurrent.locks.ReentrantLock;
25+
26+
/**
27+
* This is taken from fabric8 with some minor changes
28+
* (we need it, so it could be placed in the common package)
29+
* @author wind57
30+
*/
31+
public final class CachedSingleThreadScheduler {
32+
33+
private final ReentrantLock lock = new ReentrantLock();
34+
35+
private final long ttlMillis;
36+
private ScheduledThreadPoolExecutor executor;
37+
38+
public CachedSingleThreadScheduler(long ttlMillis) {
39+
this.ttlMillis = ttlMillis;
40+
}
41+
42+
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay,
43+
long delay, TimeUnit unit) {
44+
try {
45+
lock.lock();
46+
this.startExecutor();
47+
return this.executor.scheduleWithFixedDelay(command, initialDelay, delay, unit);
48+
}
49+
finally {
50+
lock.unlock();
51+
}
52+
}
53+
54+
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
55+
try {
56+
lock.lock();
57+
this.startExecutor();
58+
return this.executor.schedule(command, delay, unit);
59+
}
60+
finally {
61+
lock.unlock();
62+
}
63+
}
64+
65+
private void startExecutor() {
66+
if (this.executor == null) {
67+
this.executor = new ScheduledThreadPoolExecutor(1, threadFactory());
68+
this.executor.setRemoveOnCancelPolicy(true);
69+
this.executor.scheduleWithFixedDelay(this::shutdownCheck, this.ttlMillis, this.ttlMillis, TimeUnit.MILLISECONDS);
70+
}
71+
72+
}
73+
74+
private void shutdownCheck() {
75+
try {
76+
lock.lock();
77+
if (this.executor.getQueue().isEmpty()) {
78+
this.executor.shutdownNow();
79+
this.executor = null;
80+
}
81+
}
82+
finally {
83+
lock.unlock();
84+
}
85+
86+
}
87+
88+
private ThreadFactory threadFactory() {
89+
return new ThreadFactory() {
90+
ThreadFactory threadFactory = Executors.defaultThreadFactory();
91+
92+
@Override
93+
public Thread newThread(Runnable runnable) {
94+
Thread ret = threadFactory.newThread(runnable);
95+
ret.setName("fabric8-leader-election" + "-" + ret.getName());
96+
ret.setDaemon(true);
97+
return ret;
98+
}
99+
};
100+
}
101+
102+
}

‎spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/PodReady.java‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
1+
/*
2+
* Copyright 2013-2025 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.cloud.kubernetes.commons;
18+
19+
import org.springframework.core.log.LogAccessor;
20+
21+
import java.util.concurrent.CompletableFuture;
22+
import java.util.concurrent.TimeUnit;
23+
import java.util.function.BooleanSupplier;
24+
25+
/**
26+
* @author wind57
27+
*/
28+
public final class PodReady {
29+
30+
private static final LogAccessor LOG = new LogAccessor(PodReady.class);
31+
32+
private final CachedSingleThreadScheduler podReadyScheduler = new CachedSingleThreadScheduler(
33+
TimeUnit.SECONDS.toMillis(10)
34+
);
35+
36+
public CompletableFuture<Void> podReady(BooleanSupplier isPodReady, String holderIdentity, String podNamespace) {
37+
38+
CompletableFuture<Void> podReadyFuture = new CompletableFuture<>();
39+
40+
podReadyScheduler.scheduleWithFixedDelay(() -> {
41+
42+
try {
43+
LOG.info(() -> "waiting for pod : " + holderIdentity + " in namespace : " + podNamespace
44+
+ " to be ready");
45+
if (isPodReady.getAsBoolean()) {
46+
LOG.info(() -> "Pod : " + holderIdentity + " in namespace : " + podNamespace + " is ready");
47+
podReadyFuture.complete(null);
48+
}
49+
else {
50+
LOG.debug(() -> "Pod : " + holderIdentity + " in namespace : " + podNamespace
51+
+ " is not ready, " + "will retry in one second");
52+
}
53+
}
54+
catch (Exception e) {
55+
LOG.error(() -> "exception waiting for pod : " + e.getMessage());
56+
LOG.error(() -> "leader election for " + holderIdentity + " was not successful");
57+
podReadyFuture.completeExceptionally(e);
58+
}
59+
60+
}, 1, 1, TimeUnit.SECONDS);
61+
62+
return podReadyFuture;
63+
64+
}
65+
66+
}

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

Lines changed: 11 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
import jakarta.annotation.PostConstruct;
3434
import jakarta.annotation.PreDestroy;
3535

36+
import org.springframework.cloud.kubernetes.commons.PodReady;
3637
import org.springframework.cloud.kubernetes.commons.leader.election.LeaderElectionProperties;
3738
import org.springframework.core.log.LogAccessor;
3839

@@ -43,7 +44,7 @@ final class Fabric8LeaderElectionInitiator {
4344

4445
private static final LogAccessor LOG = new LogAccessor(Fabric8LeaderElectionInitiator.class);
4546

46-
private final CachedSingleThreadScheduler podReadyScheduler = new CachedSingleThreadScheduler();
47+
private final PodReady podReady = new PodReady();
4748

4849
private final String holderIdentity;
4950

@@ -86,34 +87,18 @@ void postConstruct() {
8687
LOG.info(() -> "starting leader initiator : " + holderIdentity);
8788
executorService.set(Executors
8889
.newSingleThreadExecutor(r -> new Thread(r, "Fabric8LeaderElectionInitiator-" + holderIdentity)));
89-
CompletableFuture<Void> podReadyFuture = new CompletableFuture<>();
90+
CompletableFuture<Void> podReadyFuture;
9091

9192
// wait until pod is ready
9293
if (leaderElectionProperties.waitForPodReady()) {
93-
LOG.info(() -> "need to wait until pod is ready : " + holderIdentity);
94-
podReadyTask.set(podReadyScheduler.scheduleWithFixedDelay(() -> {
95-
96-
try {
97-
LOG.info(() -> "waiting for pod : " + holderIdentity + " in namespace : " + podNamespace
98-
+ " to be ready");
99-
Pod pod = fabric8KubernetesClient.pods().inNamespace(podNamespace).withName(holderIdentity).get();
100-
boolean podReady = Readiness.isPodReady(pod);
101-
if (podReady) {
102-
LOG.info(() -> "Pod : " + holderIdentity + " in namespace : " + podNamespace + " is ready");
103-
podReadyFuture.complete(null);
104-
}
105-
else {
106-
LOG.debug(() -> "Pod : " + holderIdentity + " in namespace : " + podNamespace
107-
+ " is not ready, " + "will retry in one second");
108-
}
109-
}
110-
catch (Exception e) {
111-
LOG.error(() -> "exception waiting for pod : " + e.getMessage());
112-
LOG.error(() -> "leader election for " + holderIdentity + " was not successful");
113-
podReadyFuture.completeExceptionally(e);
114-
}
115-
116-
}, 1, 1, TimeUnit.SECONDS));
94+
LOG.info(() -> "will wait until pod " + holderIdentity + " is ready : ");
95+
podReadyFuture = podReady.podReady(() -> {
96+
Pod pod = fabric8KubernetesClient.pods().inNamespace(podNamespace).withName(holderIdentity).get();
97+
return Readiness.isPodReady(pod);
98+
}, holderIdentity, podNamespace);
99+
100+
} else {
101+
podReadyFuture = new CompletableFuture<>();
117102
}
118103

119104
// wait in a different thread until the pod is ready

0 commit comments

Comments
 (0)