1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
|
// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.vespa.zookeeper;
import com.yahoo.cloud.config.ZookeeperServerConfig;
import com.yahoo.concurrent.DaemonThreadFactory;
import com.yahoo.protect.Process;
import com.yahoo.yolean.Exceptions;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
import static com.yahoo.vespa.defaults.Defaults.getDefaults;
import static com.yahoo.vespa.zookeeper.Configurator.zookeeperServerHostnames;
/**
* Writes zookeeper config and starts zookeeper server.
*
* @author Harald Musum
*/
public class ZooKeeperRunner implements Runnable {
private static final Logger log = java.util.logging.Logger.getLogger(ZooKeeperRunner.class.getName());
private static final Duration STOP_TIMEOUT = Duration.ofSeconds(10);
private static final Duration START_TIMEOUT = Duration.ofMinutes(10);
private final ExecutorService executorService;
private final ZookeeperServerConfig zookeeperServerConfig;
private final VespaZooKeeperServer server;
private final ExponentialBackoff backoff = new ExponentialBackoff(Duration.ofSeconds(5), Duration.ofSeconds(15));
private final Sleeper sleeper = new Sleeper();
public ZooKeeperRunner(ZookeeperServerConfig zookeeperServerConfig, VespaZooKeeperServer server) {
this.zookeeperServerConfig = zookeeperServerConfig;
this.server = server;
new Configurator(zookeeperServerConfig).writeConfigToDisk();
executorService = Executors.newSingleThreadExecutor(new DaemonThreadFactory("zookeeper-server-"));
executorService.submit(this);
}
void shutdown() {
log.log(Level.INFO, "Triggering shutdown");
server.shutdown();
executorService.shutdownNow();
try {
if (!executorService.awaitTermination(STOP_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS)) {
log.log(Level.WARNING, "Failed to shut down within " + STOP_TIMEOUT);
}
} catch (InterruptedException e) {
log.log(Level.WARNING, "Interrupted waiting for executor to complete", e);
}
}
@Override
public void run() {
Path path = Paths.get(getDefaults().underVespaHome(zookeeperServerConfig.zooKeeperConfigFile()));
// Retry start of server. An already running server might take some time to shut down, starting a new
// one will fail in that case, so retry
Instant now = Instant.now();
Instant end = now.plus(START_TIMEOUT);
for (int attempt = 1; now.isBefore(end) && !executorService.isShutdown(); attempt++) {
try {
startServer(path, attempt); // Will block in a real implementation of VespaZooKeeperServer
return;
} catch (RuntimeException e) {
String messagePart = "Starting " + serverDescription() + " failed on attempt " + attempt;
if (server.reconfigurable()) {
Duration delay = backoff.delay(attempt);
log.log(Level.WARNING, messagePart + ". Retrying in " + delay + ", time left " +
Duration.between(now, end), e);
sleeper.sleep(delay);
} else {
Process.logAndDie(messagePart + ". Forcing shutdown", e);
}
} finally {
now = Instant.now();
}
}
// Failed, log config
log.log(Level.INFO, "Current content of zookeeper config file at '" + path + "':\n" +
Exceptions.uncheck(() -> Files.readString(path)));
}
private String serverDescription() {
return (server.reconfigurable() ? "" : "non-") + "reconfigurable ZooKeeper server";
}
private void startServer(Path path, int attempt) {
if (attempt > 1)
log.log(Level.INFO, "Starting ZooKeeper server with " + path.toFile().getAbsolutePath() +
". Trying to establish ZooKeeper quorum (members: " +
zookeeperServerHostnames(zookeeperServerConfig) + ", attempt " + attempt + ")");
// Note: Hack to make this work in ZooKeeper 3.6, where metrics provider class is
// loaded by using Thread.currentThread().getContextClassLoader() which does not work
// well in the container
ClassLoader tccl = Thread.currentThread().getContextClassLoader();
Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());
try {
server.start(path);
} catch (Throwable e) {
throw new RuntimeException("Starting " + serverDescription() + " failed", e);
} finally {
Thread.currentThread().setContextClassLoader(tccl);
}
}
}
|
