This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new e16a0c9ac94 Run multi-server quickstarts in separate JVMs (#20202)
e16a0c9ac94 is described below
commit e16a0c9ac943716222e43705ab99c63e3bea1035
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Tue Nov 10 20:49:29 2026 -0800
Run multi-server quickstarts in separate JVMs (#20202)
---
.github/workflows/scripts/.pinot_quickstart.sh | 2 +-
docker/images/pinot/README.md | 32 +++
.../java/org/apache/pinot/tools/Quickstart.java | 25 +-
.../tools/admin/command/QuickstartRunner.java | 79 +++++-
.../admin/command/QuickstartServerProcess.java | 290 +++++++++++++++++++
.../tools/admin/command/QuickstartRunnerTest.java | 113 ++++++++
.../admin/command/QuickstartServerHeapTest.java | 130 +++++++++
.../admin/command/QuickstartServerProcessTest.java | 306 +++++++++++++++++++++
8 files changed, 959 insertions(+), 18 deletions(-)
diff --git a/.github/workflows/scripts/.pinot_quickstart.sh
b/.github/workflows/scripts/.pinot_quickstart.sh
index dad7e5fb258..bb86f776d43 100755
--- a/.github/workflows/scripts/.pinot_quickstart.sh
+++ b/.github/workflows/scripts/.pinot_quickstart.sh
@@ -264,7 +264,7 @@ wait_for_query 8000 \
'.resultTable.rows[0][0] == 97889' \
'multi-stage count(*) on baseballStats' || PASS=0
-# Ordered so the assertion does not depend on how segments are spread across
the 3 servers. Every row must have
+# Ordered so the assertion does not depend on how segments are spread across
the servers. Every row must have
# resolved a team name, otherwise the join produced rows without matching the
dimension table.
wait_for_query 8000 \
'SET useMultistageEngine=true; SELECT a.playerName, a.teamID, b.teamName
FROM baseballStats_OFFLINE AS a JOIN dimBaseballTeams_OFFLINE AS b ON a.teamID
= b.teamID ORDER BY a.playerName, a.teamID LIMIT 10' \
diff --git a/docker/images/pinot/README.md b/docker/images/pinot/README.md
index 1208627767b..7f4260bd296 100644
--- a/docker/images/pinot/README.md
+++ b/docker/images/pinot/README.md
@@ -133,6 +133,38 @@ docker run -p 8098:8098 pinot:release-0.1.0 StartServer
-zkAddress 172.17.0.2:21
## QuickStart
+### Heap sizes for forked quickstarts
+
+Quickstarts that fork their servers run a parent JVM for the other services
and a separate JVM for each server.
+`JAVA_OPTS` sets the parent JVM's options. Its heap options are not copied to
the server JVMs. Set these system properties
+in the parent's `JAVA_OPTS` to configure each server's heap:
+
+| System property | Default | Child JVM option with defaults |
+| --- | --- | --- |
+| `pinot.quickstart.server.xmx` | `1g` | `-Xmx1g` |
+| `pinot.quickstart.server.xms` | Resolved `pinot.quickstart.server.xmx` value
| `-Xms1g` |
+
+Values use Java heap-size syntax, such as `256m` or `2g`. The initial heap
must not exceed the maximum heap.
+By default, each server uses equal initial and maximum heaps. Setting only
`pinot.quickstart.server.xmx=512m` gives
+each child `-Xms512m -Xmx512m`. Set `pinot.quickstart.server.xms` explicitly
to use a different initial heap.
+These properties affect forked quickstart servers only; they do not configure
separately launched `StartServer` commands.
+
+The Docker image defaults the parent JVM to `-Xms4G -Xmx4G`. With the default
child heap limits, the two-server BATCH
+quickstart therefore has a combined maximum Java heap of **6 GiB**: 4 GiB for
the parent plus 1 GiB for each server.
+A four-server quickstart has a combined maximum heap of **8 GiB**. Allocate
additional container memory for direct
+buffers, memory-mapped segment pages, metaspace, thread stacks, and other
native memory; these heap totals are not
+container memory limits or estimates of current memory usage.
+
+For example, configure equal 2 GiB initial and maximum heaps for both the
parent and each server:
+
+```SHELL
+docker run --rm --name pinot-quickstart \
+ -p 127.0.0.1:9000:9000 \
+ -e JAVA_OPTS="-Xms2g -Xmx2g -Dpinot.admin.system.exit=true
-Dpinot.quickstart.server.xmx=2g" \
+ pinot:latest QuickStart -type batch
+```
+
+For a local distribution, pass the same `JAVA_OPTS` environment variable to
`bin/pinot-admin.sh QuickStart -type batch`.
### Use docker compose to bring up Pinot stack
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/Quickstart.java
b/pinot-tools/src/main/java/org/apache/pinot/tools/Quickstart.java
index 6e8e1224f6d..594169f89fc 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/Quickstart.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/Quickstart.java
@@ -32,6 +32,8 @@ import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.tools.admin.PinotAdministrator;
import org.apache.pinot.tools.admin.command.QuickstartRunner;
import org.apache.pinot.tools.utils.SampleQueries;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/// The basic Batch/Offline Quickstart.
@@ -40,6 +42,7 @@ import org.apache.pinot.tools.utils.SampleQueries;
/// a dedicated quickstart each: the multi-stage engine, JSON indexes, complex
type handling, timestamp indexes,
/// lookup joins and vector similarity. The types those quickstarts used to
own are kept as deprecated aliases.
public class Quickstart extends QuickStartBase {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(Quickstart.class);
protected static final Map<String, String> OPTIONS_TO_USE_MSE =
Map.of("queryOptions",
CommonConstants.Broker.Request.QueryOptionKey.USE_MULTISTAGE_ENGINE +
"=true");
@@ -112,10 +115,9 @@ public class Quickstart extends QuickStartBase {
new QuickstartRunner(quickstartTableRequests, 1, 1,
getNumQuickstartRunnerServers(), 1, quickstartRunnerDir,
true, getAuthProvider(),
getConfigOverrides(), _zkExternalAddress, true,
getClusterConfigOverrides(),
- this::getIndividualBrokerConfigOverridesFunction,
this::getIndividualServerConfigOverridesFunction);
+ this::getIndividualBrokerConfigOverridesFunction,
this::getIndividualServerConfigOverridesFunction)
+ .setForkServers(getNumQuickstartRunnerServers() > 1);
- printStatus(Color.CYAN, "***** Starting Zookeeper, controller, broker,
server and minion *****");
- runner.startAll();
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
try {
printStatus(Color.GREEN, "***** Shutting down quick start *****");
@@ -125,6 +127,18 @@ public class Quickstart extends QuickStartBase {
e.printStackTrace();
}
}));
+ printStatus(Color.CYAN, "***** Starting Zookeeper, controller, broker,
server and minion *****");
+ try {
+ runner.startAll();
+ } catch (Exception e) {
+ LOGGER.error("Failed to start quickstart", e);
+ try {
+ runner.stop();
+ } catch (Exception cleanupException) {
+ e.addSuppressed(cleanupException);
+ }
+ throw e;
+ }
if (!CollectionUtils.isEmpty(quickstartTableRequests)) {
printStatus(Color.CYAN, "***** Bootstrap tables *****");
@@ -144,13 +158,12 @@ public class Quickstart extends QuickStartBase {
QuickstartRunner.DEFAULT_CONTROLLER_PORT));
}
- /// Three servers so that the multi-stage sample queries below actually
exercise a cross-server exchange, which is
- /// what the merged MULTI_STAGE and JOIN quickstarts used to set up.
+ /// Two servers in separate JVMs exercise cross-server exchanges without
sharing server-local static state.
///
/// This is the default for every quickstart extending this class, not just
`-type BATCH`. Subclasses that do not
/// need distributed execution can override it; [TPCHQuickStart] for
instance runs on a single server.
protected int getNumQuickstartRunnerServers() {
- return 3;
+ return 2;
}
@Override
diff --git
a/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/QuickstartRunner.java
b/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/QuickstartRunner.java
index c23f9b51810..b72d5ec1a1d 100644
---
a/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/QuickstartRunner.java
+++
b/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/QuickstartRunner.java
@@ -28,6 +28,7 @@ import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.IntFunction;
import org.apache.commons.io.FileUtils;
import org.apache.helix.HelixManager;
@@ -86,7 +87,10 @@ public class QuickstartRunner {
private final List<Integer> _controllerPorts = new ArrayList<>();
private final List<Integer> _brokerPorts = new ArrayList<>();
- private boolean _isStopped = false;
+ private final List<QuickstartServerProcess> _serverProcesses = new
CopyOnWriteArrayList<>();
+ private boolean _forkServers;
+ private boolean _stopRequested;
+ private boolean _isStopped;
public QuickstartRunner(List<QuickstartTableRequest> tableRequests, int
numControllers, int numBrokers,
int numServers, int numMinions, File tempDir, Map<String, Object>
configOverrides)
@@ -202,6 +206,13 @@ public class QuickstartRunner {
}
}
+ /// Starts servers in separate JVMs, isolating server-local static state
such as dimension-table managers.
+ /// Configure this before calling [#startAll].
+ public QuickstartRunner setForkServers(boolean forkServers) {
+ _forkServers = forkServers;
+ return this;
+ }
+
private void startServers()
throws Exception {
for (int i = 0; i < _numServers; i++) {
@@ -222,7 +233,18 @@ public class QuickstartRunner {
} else {
serverStarter.setConfigOverrides(_configOverrides);
}
- if (!serverStarter.execute()) {
+ if (_forkServers) {
+ QuickstartServerProcess process;
+ synchronized (_serverProcesses) {
+ if (_stopRequested) {
+ throw new IllegalStateException("Quickstart stopped while starting
server " + i);
+ }
+ process =
QuickstartServerProcess.launch(serverStarter.getServerConf(),
+ new File(_tempDir, "server-process-" + i));
+ _serverProcesses.add(process);
+ }
+ process.awaitReady();
+ } else if (!serverStarter.execute()) {
throw new RuntimeException("Failed to start Server");
}
}
@@ -248,26 +270,42 @@ public class QuickstartRunner {
public void startAll()
throws Exception {
- registerDefaultPinotFS();
- if (_zkExternalAddress == null) {
- startZookeeper();
+ try {
+ registerDefaultPinotFS();
+ if (_zkExternalAddress == null) {
+ startZookeeper();
+ }
+ startControllers();
+ startBrokers();
+ startServers();
+ startMinions();
+ } catch (Exception e) {
+ try {
+ stopServerProcesses();
+ } catch (Exception cleanupException) {
+ e.addSuppressed(cleanupException);
+ }
+ throw e;
}
- startControllers();
- startBrokers();
- startServers();
- startMinions();
}
- public void stop()
+ public synchronized void stop()
throws Exception {
if (_isStopped) {
return;
}
+ synchronized (_serverProcesses) {
+ _stopRequested = true;
+ }
+ stopServerProcesses();
// TODO: Stop Minion
StopProcessCommand stopper = new StopProcessCommand(false);
if (_zkExternalAddress == null) {
- stopper.stopController().stopBroker().stopServer().stopZookeeper();
+ stopper.stopController().stopBroker().stopZookeeper();
+ if (!_forkServers) {
+ stopper.stopServer();
+ }
}
stopper.execute();
if (_deleteExistingData) {
@@ -276,6 +314,25 @@ public class QuickstartRunner {
_isStopped = true;
}
+ private void stopServerProcesses()
+ throws Exception {
+ Exception failure = null;
+ for (QuickstartServerProcess process : _serverProcesses) {
+ try {
+ process.stop();
+ } catch (Exception e) {
+ if (failure == null) {
+ failure = e;
+ } else {
+ failure.addSuppressed(e);
+ }
+ }
+ }
+ if (failure != null) {
+ throw failure;
+ }
+ }
+
public void createServerTenantWith(int numOffline, int numRealtime, String
tenantName)
throws Exception {
new AddTenantCommand().setControllerUrl("http://localhost:"; +
_controllerPorts.get(0)).setName(tenantName)
diff --git
a/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/QuickstartServerProcess.java
b/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/QuickstartServerProcess.java
new file mode 100644
index 00000000000..f3f21b82b76
--- /dev/null
+++
b/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/QuickstartServerProcess.java
@@ -0,0 +1,290 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.tools.admin.command;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.RandomAccessFile;
+import java.lang.management.ManagementFactory;
+import java.net.HttpURLConnection;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.configuration2.convert.LegacyListDelimiterHandler;
+import org.apache.commons.configuration2.io.FileHandler;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.tools.admin.PinotAdministrator;
+
+
+/// Owns one quickstart server JVM, its configuration, and its startup log.
Startup waits for the server's readiness
+/// endpoint. The child exits when its parent's stdin pipe closes, including
when the parent is killed abruptly.
+/// [#stop] is synchronized and can be called repeatedly or concurrently with
waiting for readiness.
+final class QuickstartServerProcess {
+ static final String STARTED_FILE_NAME = "server.started";
+ static final String INITIAL_HEAP_PROPERTY = "pinot.quickstart.server.xms";
+ static final String MAX_HEAP_PROPERTY = "pinot.quickstart.server.xmx";
+ private static final String DEFAULT_MAX_HEAP = "1g";
+ private static final Duration STARTUP_TIMEOUT = Duration.ofMinutes(2);
+ private static final long SHUTDOWN_TIMEOUT_SECONDS = 30;
+ private static final long FORCED_SHUTDOWN_TIMEOUT_SECONDS = 10;
+ private static final long PARENT_LOSS_SHUTDOWN_TIMEOUT_SECONDS = 5;
+
+ private final Process _process;
+ private final File _logFile;
+ private final File _startedFile;
+ private final int _port;
+
+ private QuickstartServerProcess(Process process, File logFile, File
startedFile, int port) {
+ _process = process;
+ _logFile = logFile;
+ _startedFile = startedFile;
+ _port = port;
+ }
+
+ static QuickstartServerProcess launch(Map<String, Object> effectiveConfig,
File processDir)
+ throws Exception {
+ Files.createDirectories(processDir.toPath());
+ File configFile = new File(processDir, "server.properties");
+ writeConfig(effectiveConfig, configFile);
+ int port =
Integer.parseInt(effectiveConfig.get(CommonConstants.Server.CONFIG_OF_ADMIN_API_PORT).toString());
+ return launch(buildCommand(configFile, processDir), processDir, port);
+ }
+
+ static void writeConfig(Map<String, Object> effectiveConfig, File configFile)
+ throws Exception {
+ PropertiesConfiguration configuration = new PropertiesConfiguration();
+ configuration.setListDelimiterHandler(new LegacyListDelimiterHandler(','));
+ configuration.getLayout().setForceSingleLine(true);
+ effectiveConfig.forEach(configuration::setProperty);
+ new FileHandler(configuration).save(configFile);
+ }
+
+ static List<String> buildCommand(File configFile, File processDir) {
+ List<String> command = new ArrayList<>();
+ File javaExecutable = new File(System.getProperty("java.home"),
"bin/java");
+ command.add(javaExecutable.getAbsolutePath());
+ String maxHeap = System.getProperty(MAX_HEAP_PROPERTY, DEFAULT_MAX_HEAP);
+ command.add("-Xms" + System.getProperty(INITIAL_HEAP_PROPERTY, maxHeap));
+ command.add("-Xmx" + maxHeap);
+ Iterator<String> parentArguments =
ManagementFactory.getRuntimeMXBean().getInputArguments().iterator();
+ while (parentArguments.hasNext()) {
+ String argument = parentArguments.next();
+ for (String option : List.of("--add-opens", "--add-exports",
"--add-modules", "--enable-native-access")) {
+ if (argument.startsWith(option + "=")) {
+ command.add(argument);
+ break;
+ }
+ if (argument.equals(option) && parentArguments.hasNext()) {
+ command.add(argument);
+ command.add(parentArguments.next());
+ break;
+ }
+ }
+ }
+ for (String name : System.getProperties().stringPropertyNames()) {
+ if (name.startsWith("plugins.") || name.startsWith("log4j.") ||
name.startsWith("log4j2.")
+ || name.startsWith("io.netty.") ||
name.startsWith("io.grpc.netty.")) {
+ command.add("-D" + name + "=" + System.getProperty(name));
+ }
+ }
+ command.add("-Djava.io.tmpdir=" + new File(processDir,
"tmp").getAbsolutePath());
+ command.add("-Dpinot.admin.system.exit=true");
+ command.add("-cp");
+ command.add(System.getProperty("java.class.path"));
+ command.add(QuickstartServerProcess.class.getName());
+ command.add(configFile.getAbsolutePath());
+ command.add(new File(processDir, STARTED_FILE_NAME).getAbsolutePath());
+ return command;
+ }
+
+ static QuickstartServerProcess launch(List<String> command, File processDir,
int port)
+ throws IOException {
+ Files.createDirectories(new File(processDir, "tmp").toPath());
+ File logDir = new File(processDir, "logs");
+ Files.createDirectories(logDir.toPath());
+ File logFile = new File(processDir, "server.log");
+ File startedFile = new File(processDir, STARTED_FILE_NAME);
+ Files.deleteIfExists(startedFile.toPath());
+ ProcessBuilder processBuilder = new
ProcessBuilder(command).redirectErrorStream(true).redirectOutput(logFile);
+ processBuilder.environment().put("LOG_ROOT", logDir.getAbsolutePath());
+ processBuilder.environment().put("PINOT_COMPONENT", "server");
+ Process process = processBuilder.start();
+ return new QuickstartServerProcess(process, logFile, startedFile, port);
+ }
+
+ void awaitReady()
+ throws Exception {
+ awaitReady(STARTUP_TIMEOUT);
+ }
+
+ void awaitReady(Duration timeout)
+ throws Exception {
+ try {
+ waitForReadiness(timeout);
+ } catch (Exception e) {
+ try {
+ stop();
+ } catch (Exception stopException) {
+ e.addSuppressed(stopException);
+ }
+ addLogTail(e);
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ throw e;
+ }
+ }
+
+ private void waitForReadiness(Duration timeout)
+ throws IOException, InterruptedException {
+ long deadline = System.nanoTime() + timeout.toNanos();
+ URI healthUri = URI.create("http://127.0.0.1:"; + _port +
"/health/readiness");
+ while (System.nanoTime() < deadline) {
+ if (!_process.isAlive()) {
+ throw new IOException("Quickstart server exited with code " +
_process.exitValue() + "; see logs in "
+ + _logFile.getParentFile());
+ }
+ // An existing service on the same port must not make a child that has
not finished starting look ready.
+ if (!_startedFile.isFile()) {
+ Thread.sleep(200);
+ continue;
+ }
+ HttpURLConnection connection = (HttpURLConnection)
healthUri.toURL().openConnection();
+ connection.setConnectTimeout(1000);
+ connection.setReadTimeout(1000);
+ try {
+ if (connection.getResponseCode() == HttpURLConnection.HTTP_OK) {
+ try (InputStream response = connection.getInputStream()) {
+ if ("OK".equals(new String(response.readNBytes(3),
StandardCharsets.UTF_8).trim()) && _process.isAlive()) {
+ return;
+ }
+ }
+ }
+ } catch (IOException e) {
+ // The admin listener may not be bound yet; retry until readiness or
the startup deadline.
+ } finally {
+ connection.disconnect();
+ }
+ Thread.sleep(200);
+ }
+ throw new IOException("Timed out waiting for quickstart server readiness
at " + healthUri + "; see logs in "
+ + _logFile.getParentFile());
+ }
+
+ private void addLogTail(Exception failure) {
+ // Keep useful diagnostics in the exception even if the quickstart's
shutdown hook removes its working directory.
+ addLogTail(failure, _logFile);
+ File runtimeLog = new File(_logFile.getParentFile(),
"logs/pinot-server.log");
+ if (runtimeLog.isFile()) {
+ addLogTail(failure, runtimeLog);
+ }
+ }
+
+ private static void addLogTail(Exception failure, File logFile) {
+ try (RandomAccessFile log = new RandomAccessFile(logFile, "r")) {
+ int length = (int) Math.min(log.length(), 8192);
+ if (length > 0) {
+ byte[] bytes = new byte[length];
+ log.seek(log.length() - length);
+ log.readFully(bytes);
+ failure.addSuppressed(new IOException("Quickstart server log tail from
" + logFile + ":\n"
+ + new String(bytes, StandardCharsets.UTF_8)));
+ }
+ } catch (IOException e) {
+ failure.addSuppressed(e);
+ }
+ }
+
+ Process getProcess() {
+ return _process;
+ }
+
+ synchronized void stop()
+ throws IOException, InterruptedException {
+ if (!_process.isAlive()) {
+ return;
+ }
+ // Process.destroy() also closes stdin, which would trigger the shorter
parent-loss deadline during normal stop.
+ _process.toHandle().destroy();
+ try {
+ if (!_process.waitFor(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+ _process.destroyForcibly();
+ if (!_process.waitFor(FORCED_SHUTDOWN_TIMEOUT_SECONDS,
TimeUnit.SECONDS)) {
+ throw new IOException("Quickstart server did not exit after forced
shutdown; see " + _logFile);
+ }
+ }
+ } catch (InterruptedException e) {
+ _process.destroyForcibly();
+ try {
+ _process.waitFor(FORCED_SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ } catch (InterruptedException interruptedAgain) {
+ e.addSuppressed(interruptedAgain);
+ } finally {
+ Thread.currentThread().interrupt();
+ }
+ throw e;
+ }
+ }
+
+ static void watchParent() {
+ Thread.ofPlatform().daemon().name("quickstart-parent-watch").start(() -> {
+ try {
+ while (System.in.read() != -1) {
+ // The parent keeps this pipe open for the lifetime of the server.
+ }
+ exitAfterParentLoss(0);
+ } catch (IOException e) {
+ exitAfterParentLoss(1);
+ }
+ });
+ }
+
+ private static void exitAfterParentLoss(int exitCode) {
+ // Embedded ZooKeeper may have died with the parent. Server shutdown hooks
can then block on Helix retries,
+ // including if shutdown is already in progress, so enforce the deadline
independently of System.exit().
+ Thread.ofPlatform().daemon().name("quickstart-parent-loss-halt").start(()
-> {
+ try {
+ TimeUnit.SECONDS.sleep(PARENT_LOSS_SHUTDOWN_TIMEOUT_SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ Runtime.getRuntime().halt(exitCode);
+ });
+ System.exit(exitCode);
+ }
+
+ public static void main(String[] args) {
+ watchParent();
+ PinotAdministrator.main(new String[]{"StartServer", "-configFile",
args[0]});
+ try {
+ Files.createFile(new File(args[1]).toPath());
+ } catch (IOException e) {
+ e.printStackTrace();
+ System.exit(1);
+ }
+ }
+}
diff --git
a/pinot-tools/src/test/java/org/apache/pinot/tools/admin/command/QuickstartRunnerTest.java
b/pinot-tools/src/test/java/org/apache/pinot/tools/admin/command/QuickstartRunnerTest.java
new file mode 100644
index 00000000000..ceacaa447c6
--- /dev/null
+++
b/pinot-tools/src/test/java/org/apache/pinot/tools/admin/command/QuickstartRunnerTest.java
@@ -0,0 +1,113 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.tools.admin.command;
+
+import java.lang.reflect.Field;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.io.FileUtils;
+import org.mockito.MockedConstruction;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.RETURNS_SELF;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.withSettings;
+import static org.testng.Assert.assertEquals;
+
+
+/// Verifies quickstart process ownership during shutdown. Tests use
thread-scoped construction mocks and private
+/// temporary directories so they never scan real PID files, kill processes,
or change global system properties.
+public class QuickstartRunnerTest {
+ @DataProvider
+ public Object[][] forkModes() {
+ return new Object[][]{{false}, {true}};
+ }
+
+ @Test(dataProvider = "forkModes")
+ public void testExternalZookeeperSkipsGlobalPidScanning(boolean forkServers)
+ throws Exception {
+ Path directory = Files.createTempDirectory("quickstart-runner-test");
+ try (MockedConstruction<StopProcessCommand> stoppers =
+ mockConstruction(StopProcessCommand.class,
withSettings().defaultAnswer(RETURNS_SELF))) {
+ QuickstartRunner runner = newRunner(directory,
"external-zookeeper:2181", forkServers);
+ QuickstartServerProcess ownedServer =
mock(QuickstartServerProcess.class);
+ if (forkServers) {
+ addOwnedServer(runner, ownedServer);
+ }
+ runner.stop();
+ runner.stop();
+
+ assertEquals(stoppers.constructed().size(), 1);
+ StopProcessCommand stopper = stoppers.constructed().getFirst();
+ verify(stopper).execute();
+ // Selecting any role enables a global PID scan and can stop processes
belonging to another local cluster.
+ verifyNoMoreInteractions(stopper);
+ if (forkServers) {
+ verify(ownedServer).stop();
+ }
+ verifyNoMoreInteractions(ownedServer);
+ } finally {
+ FileUtils.deleteDirectory(directory.toFile());
+ }
+ }
+
+ @Test(dataProvider = "forkModes")
+ public void testEmbeddedZookeeperKeepsLegacyCleanup(boolean forkServers)
+ throws Exception {
+ Path directory = Files.createTempDirectory("quickstart-runner-test");
+ try (MockedConstruction<StopProcessCommand> stoppers =
+ mockConstruction(StopProcessCommand.class,
withSettings().defaultAnswer(RETURNS_SELF))) {
+ newRunner(directory, null, forkServers).stop();
+
+ assertEquals(stoppers.constructed().size(), 1);
+ StopProcessCommand stopper = stoppers.constructed().getFirst();
+ verify(stopper).stopController();
+ verify(stopper).stopBroker();
+ verify(stopper).stopZookeeper();
+ if (!forkServers) {
+ verify(stopper).stopServer();
+ }
+ verify(stopper).execute();
+ verifyNoMoreInteractions(stopper);
+ } finally {
+ FileUtils.deleteDirectory(directory.toFile());
+ }
+ }
+
+ private static QuickstartRunner newRunner(Path directory, String
externalZookeeper, boolean forkServers)
+ throws Exception {
+ return new QuickstartRunner(List.of(), 0, 0, 0, 0, directory.toFile(),
true, null, Map.of(), externalZookeeper,
+ false, Map.of()).setForkServers(forkServers);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static void addOwnedServer(QuickstartRunner runner,
QuickstartServerProcess server)
+ throws Exception {
+ // Register an owned process without starting a real cluster; this field
is declared with the same element type.
+ Field field = QuickstartRunner.class.getDeclaredField("_serverProcesses");
+ field.setAccessible(true);
+ ((List<QuickstartServerProcess>) field.get(runner)).add(server);
+ }
+}
diff --git
a/pinot-tools/src/test/java/org/apache/pinot/tools/admin/command/QuickstartServerHeapTest.java
b/pinot-tools/src/test/java/org/apache/pinot/tools/admin/command/QuickstartServerHeapTest.java
new file mode 100644
index 00000000000..e2b11a31384
--- /dev/null
+++
b/pinot-tools/src/test/java/org/apache/pinot/tools/admin/command/QuickstartServerHeapTest.java
@@ -0,0 +1,130 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.tools.admin.command;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.lang.management.ManagementFactory;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+
+/// Verifies the heap limits used by actual child JVMs. Tests execute
sequentially because constructing their commands
+/// temporarily changes the dedicated quickstart heap system properties,
restoring them before launching a child.
+public class QuickstartServerHeapTest {
+ @DataProvider
+ public Object[][] heapSizes() {
+ return new Object[][]{
+ {null, null, "1g", "1g", 1024L * 1024 * 1024},
+ {null, "512m", "512m", "512m", 512L * 1024 * 1024},
+ {"64m", "192m", "64m", "192m", 192L * 1024 * 1024}
+ };
+ }
+
+ @Test(dataProvider = "heapSizes", timeOut = 30_000)
+ public void testChildHeap(String initialHeap, String maxHeap, String
expectedInitialHeap, String expectedMaxHeap,
+ long expectedMaxBytes)
+ throws Exception {
+ Path processDir = Files.createTempDirectory("quickstart-server-heap-test");
+ Process process = null;
+ try {
+ Files.createDirectory(processDir.resolve("tmp"));
+ Path reportFile = processDir.resolve("heap.properties");
+ List<String> command = buildProbeCommand(processDir, reportFile,
initialHeap, maxHeap);
+ Path logFile = processDir.resolve("probe.log");
+ process = new
ProcessBuilder(command).redirectErrorStream(true).redirectOutput(logFile.toFile()).start();
+ assertTrue(process.waitFor(15, TimeUnit.SECONDS), "Heap probe did not
exit");
+ assertEquals(process.exitValue(), 0, Files.readString(logFile));
+ Properties report = new Properties();
+ try (InputStream input = Files.newInputStream(reportFile)) {
+ report.load(input);
+ }
+ List<String> arguments =
List.of(report.getProperty("arguments").split("\n"));
+ assertEquals(arguments.stream().filter(argument ->
argument.startsWith("-Xms")).toList(),
+ List.of("-Xms" + expectedInitialHeap));
+ assertEquals(arguments.stream().filter(argument ->
argument.startsWith("-Xmx")).toList(),
+ List.of("-Xmx" + expectedMaxHeap));
+ long maxBytes = Long.parseLong(report.getProperty("maxMemory"));
+ assertTrue(maxBytes <= expectedMaxBytes, "Child exceeded its configured
heap limit: " + maxBytes);
+ // Some collectors exclude a survivor region from Runtime.maxMemory().
+ assertTrue(maxBytes >= expectedMaxBytes * 3 / 4, "Child did not apply
the requested heap limit: " + maxBytes);
+ } finally {
+ if (process != null && process.isAlive()) {
+ process.destroyForcibly();
+ process.waitFor(10, TimeUnit.SECONDS);
+ }
+ FileUtils.deleteDirectory(processDir.toFile());
+ }
+ }
+
+ private static List<String> buildProbeCommand(Path processDir, Path
reportFile, String initialHeap, String maxHeap) {
+ String previousInitialHeap =
System.getProperty(QuickstartServerProcess.INITIAL_HEAP_PROPERTY);
+ String previousMaxHeap =
System.getProperty(QuickstartServerProcess.MAX_HEAP_PROPERTY);
+ List<String> command;
+ try {
+ setProperty(QuickstartServerProcess.INITIAL_HEAP_PROPERTY, initialHeap);
+ setProperty(QuickstartServerProcess.MAX_HEAP_PROPERTY, maxHeap);
+ command = new
ArrayList<>(QuickstartServerProcess.buildCommand(processDir.resolve("server.properties").toFile(),
+ processDir.toFile()));
+ } finally {
+ setProperty(QuickstartServerProcess.INITIAL_HEAP_PROPERTY,
previousInitialHeap);
+ setProperty(QuickstartServerProcess.MAX_HEAP_PROPERTY, previousMaxHeap);
+ }
+ command.set(command.indexOf("-cp") + 1,
+ System.getProperty("surefire.test.class.path",
System.getProperty("java.class.path")));
+ int mainIndex = command.indexOf(QuickstartServerProcess.class.getName());
+ command.subList(mainIndex, command.size()).clear();
+ command.addAll(List.of(HeapProbe.class.getName(), reportFile.toString()));
+ return command;
+ }
+
+ private static void setProperty(String name, String value) {
+ if (value == null) {
+ System.clearProperty(name);
+ } else {
+ System.setProperty(name, value);
+ }
+ }
+
+ /// Reports this JVM's heap settings and exits. The probe only runs on its
main thread.
+ public static final class HeapProbe {
+ private HeapProbe() {
+ }
+
+ public static void main(String[] args)
+ throws Exception {
+ Properties report = new Properties();
+ report.setProperty("arguments", String.join("\n",
ManagementFactory.getRuntimeMXBean().getInputArguments()));
+ report.setProperty("maxMemory",
Long.toString(Runtime.getRuntime().maxMemory()));
+ try (OutputStream output = Files.newOutputStream(Path.of(args[0]))) {
+ report.store(output, null);
+ }
+ }
+ }
+}
diff --git
a/pinot-tools/src/test/java/org/apache/pinot/tools/admin/command/QuickstartServerProcessTest.java
b/pinot-tools/src/test/java/org/apache/pinot/tools/admin/command/QuickstartServerProcessTest.java
new file mode 100644
index 00000000000..203d8e2b0c4
--- /dev/null
+++
b/pinot-tools/src/test/java/org/apache/pinot/tools/admin/command/QuickstartServerProcessTest.java
@@ -0,0 +1,306 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.tools.admin.command;
+
+import com.sun.net.httpserver.HttpServer;
+import java.io.File;
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.tools.utils.PinotConfigUtils;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+
+
+/// Exercises child JVM readiness and cleanup without starting a Pinot
cluster. Test methods run sequentially because
+/// they share the per-method temporary directory; the HTTP child handles
requests on its own executor.
+public class QuickstartServerProcessTest {
+ private Path _processDir;
+ private Path _pidFile;
+
+ @BeforeMethod
+ public void setUp()
+ throws Exception {
+ _processDir = Files.createTempDirectory("quickstart-server-process-test");
+ _pidFile = _processDir.resolve("child.pid");
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void tearDown()
+ throws Exception {
+ // Also clean up a child if an assertion fails before the helper can stop
it.
+ if (Files.exists(_pidFile)) {
+ ProcessHandle process = ProcessHandle.of(readPid()).orElse(null);
+ if (process != null && process.isAlive()) {
+ process.destroyForcibly();
+ process.onExit().get(10, TimeUnit.SECONDS);
+ }
+ }
+ FileUtils.deleteDirectory(_processDir.toFile());
+ }
+
+ @Test(timeOut = 30_000)
+ public void testReadyChildAndRepeatedStop()
+ throws Exception {
+ int port = availablePort();
+ QuickstartServerProcess server = startServer("ready", port,
Duration.ofSeconds(15));
+ try {
+ assertNotEquals(readPid(), ProcessHandle.current().pid());
+ assertEquals(server.getProcess().pid(), readPid());
+ assertTrue(server.getProcess().isAlive());
+ assertTrue(Files.exists(_processDir.resolve("health-checked")));
+ List<String> arguments =
Files.readAllLines(_processDir.resolve("jvm-arguments"));
+ assertEquals(arguments.stream().filter(argument ->
argument.startsWith("-Xmx")).toList(), List.of("-Xmx1g"));
+ assertFalse(arguments.stream().anyMatch(argument ->
argument.startsWith("-agentlib:")
+ || argument.startsWith("-javaagent:")));
+ } finally {
+ server.stop();
+ }
+ server.stop();
+ assertFalse(server.getProcess().isAlive());
+ }
+
+ @Test(timeOut = 30_000)
+ public void testEarlyExitIsReported()
+ throws Exception {
+ int port = availablePort();
+ IOException failure = expectThrows(IOException.class,
+ () -> startServer("exit", port, Duration.ofSeconds(15)));
+ assertTrue(failure.getMessage().contains("23"), failure.getMessage());
+ assertTrue(failure.getMessage().contains(_processDir.toString()),
failure.getMessage());
+ assertFalse(isChildAlive());
+ }
+
+ @Test(timeOut = 30_000)
+ public void testReadinessTimeoutStopsChild()
+ throws Exception {
+ int port = availablePort();
+ expectThrows(IOException.class,
+ () -> startServer("unready", port, Duration.ofSeconds(5)));
+ assertTrue(Files.exists(_processDir.resolve("health-checked")), "The child
must run but remain unready");
+ assertFalse(isChildAlive(), "A failed startup must not leave its child
running");
+ }
+
+ @Test(timeOut = 30_000)
+ public void testExistingHealthyListenerDoesNotMakeChildReady()
+ throws Exception {
+ HttpServer existingServer = HttpServer.create(new
InetSocketAddress("127.0.0.1", 0), 0);
+ existingServer.createContext("/health/readiness", exchange -> {
+ byte[] body = "OK".getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, body.length);
+ exchange.getResponseBody().write(body);
+ exchange.close();
+ });
+ existingServer.start();
+ try {
+ int port = existingServer.getAddress().getPort();
+ // A marker from a previous run must not be accepted either.
+
Files.writeString(_processDir.resolve(QuickstartServerProcess.STARTED_FILE_NAME),
"stale");
+ IOException failure = expectThrows(IOException.class,
+ () -> startServer("no-marker", port, Duration.ofSeconds(5)));
+ assertTrue(failure.getMessage().contains("Timed out"),
failure.getMessage());
+ assertFalse(isChildAlive(), "A child that did not start its own server
must be stopped");
+ } finally {
+ existingServer.stop(0);
+ }
+ }
+
+ @Test(timeOut = 30_000)
+ public void testParentPipeClosureStopsChild()
+ throws Exception {
+ int port = availablePort();
+ QuickstartServerProcess server = startServer("watch-parent", port,
Duration.ofSeconds(15));
+ try {
+ server.getProcess().getOutputStream().close();
+ assertTrue(server.getProcess().waitFor(10, TimeUnit.SECONDS), "The child
must exit when its parent disappears");
+ assertEquals(server.getProcess().exitValue(), 0);
+ } finally {
+ server.stop();
+ }
+ }
+
+ @DataProvider
+ public Object[][] shutdownStates() {
+ return new Object[][]{{false}, {true}};
+ }
+
+ @Test(timeOut = 30_000)
+ public void testNormalStopPreservesGracefulShutdownDeadline()
+ throws Exception {
+ int port = availablePort();
+ QuickstartServerProcess server = startServer("watch-parent-slow-shutdown",
port, Duration.ofSeconds(15));
+ try {
+ server.stop();
+ assertTrue(Files.exists(_processDir.resolve("shutdown-completed")),
+ "Normal stop must allow hooks longer than the parent-loss deadline
to finish");
+ } finally {
+ server.getProcess().destroyForcibly();
+ assertTrue(server.getProcess().waitFor(10, TimeUnit.SECONDS));
+ }
+ }
+
+ @Test(dataProvider = "shutdownStates", timeOut = 40_000)
+ public void testParentPipeClosureStopsChildWithBlockingShutdownHook(boolean
shutdownAlreadyStarted)
+ throws Exception {
+ int port = availablePort();
+ QuickstartServerProcess server =
startServer("watch-parent-blocking-shutdown", port, Duration.ofSeconds(15));
+ try {
+ if (shutdownAlreadyStarted) {
+ server.getProcess().toHandle().destroy();
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ while (!Files.exists(_processDir.resolve("shutdown-started")) &&
System.nanoTime() < deadline) {
+ Thread.sleep(20);
+ }
+ assertTrue(Files.exists(_processDir.resolve("shutdown-started")),
"Shutdown hook must have started before EOF");
+ assertTrue(server.getProcess().isAlive(), "The shutdown hook must
still be blocked");
+ }
+ server.getProcess().getOutputStream().close();
+ assertTrue(server.getProcess().waitFor(10, TimeUnit.SECONDS),
+ "A blocking shutdown hook must not leave the child alive after its
parent disappears");
+ assertTrue(Files.exists(_processDir.resolve("shutdown-started")),
"Graceful shutdown must be attempted first");
+ assertEquals(server.getProcess().exitValue(), 0);
+ } finally {
+ // Force cleanup on assertion failure: the deliberately blocked hook
cannot finish a graceful stop.
+ server.getProcess().destroyForcibly();
+ assertTrue(server.getProcess().waitFor(10, TimeUnit.SECONDS));
+ }
+ }
+
+ @Test
+ public void testConfigRoundTrip()
+ throws Exception {
+ File configFile = _processDir.resolve("server.properties").toFile();
+ String directory = _processDir.resolve("data with
spaces=and-unicode-\u00e9").toString();
+ Map<String, Object> properties = Map.of("pinot.server.instance.dataDir",
directory,
+ "pinot.server.adminapi.port", 7500,
"pinot.server.instance.enableThreadCpuTimeMeasurement", true,
+ "test.list", List.of("first", "second"), "test.literal",
"comma,value=with:punctuation");
+ QuickstartServerProcess.writeConfig(properties, configFile);
+ PinotConfiguration loaded = new
PinotConfiguration(PinotConfigUtils.readConfigFromFile(configFile.toString()));
+ assertEquals(loaded.getProperty("pinot.server.instance.dataDir"),
directory);
+ assertEquals(loaded.getProperty("pinot.server.adminapi.port", 0), 7500);
+
assertTrue(loaded.getProperty("pinot.server.instance.enableThreadCpuTimeMeasurement",
false));
+ assertEquals(loaded.getCommaSeparatedList("test.list", List.of()),
List.of("first", "second"));
+ assertEquals(loaded.getProperty("test.literal"),
properties.get("test.literal"));
+ }
+
+ private QuickstartServerProcess startServer(String mode, int port, Duration
timeout)
+ throws Exception {
+ QuickstartServerProcess server =
+ QuickstartServerProcess.launch(childCommand(mode, port),
_processDir.toFile(), port);
+ server.awaitReady(timeout);
+ return server;
+ }
+
+ private List<String> childCommand(String mode, int port) {
+ List<String> command = new
ArrayList<>(QuickstartServerProcess.buildCommand(
+ _processDir.resolve("server.properties").toFile(),
_processDir.toFile()));
+ command.set(command.indexOf("-cp") + 1,
+ System.getProperty("surefire.test.class.path",
System.getProperty("java.class.path")));
+ int mainIndex = command.indexOf(QuickstartServerProcess.class.getName());
+ command.subList(mainIndex, command.size()).clear();
+ command.addAll(List.of(TestServer.class.getName(), mode,
Integer.toString(port), _pidFile.toString()));
+ return command;
+ }
+
+ private static int availablePort()
+ throws Exception {
+ try (ServerSocket socket = new ServerSocket(0)) {
+ return socket.getLocalPort();
+ }
+ }
+
+ private long readPid()
+ throws Exception {
+ return Long.parseLong(Files.readString(_pidFile));
+ }
+
+ private boolean isChildAlive()
+ throws Exception {
+ return
ProcessHandle.of(readPid()).map(ProcessHandle::isAlive).orElse(false);
+ }
+
+ /// Minimal child process used by these tests. Its mode and paths are
immutable after startup; HTTP requests only
+ /// write an idempotent readiness marker.
+ public static final class TestServer {
+ private TestServer() {
+ }
+
+ public static void main(String[] args)
+ throws Exception {
+ String mode = args[0];
+ Path pidFile = Path.of(args[2]);
+ if (mode.equals("watch-parent-blocking-shutdown") ||
mode.equals("watch-parent-slow-shutdown")) {
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ try {
+ Files.writeString(pidFile.resolveSibling("shutdown-started"),
"started");
+ if (mode.equals("watch-parent-blocking-shutdown")) {
+ new CountDownLatch(1).await();
+ } else {
+ TimeUnit.SECONDS.sleep(7);
+ Files.writeString(pidFile.resolveSibling("shutdown-completed"),
"completed");
+ }
+ } catch (IOException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }));
+ }
+ if (mode.startsWith("watch-parent")) {
+ QuickstartServerProcess.watchParent();
+ }
+ Files.writeString(pidFile, Long.toString(ProcessHandle.current().pid()));
+ Files.write(pidFile.resolveSibling("jvm-arguments"),
ManagementFactory.getRuntimeMXBean().getInputArguments());
+ if (mode.equals("exit")) {
+ System.exit(23);
+ }
+ if (mode.equals("no-marker")) {
+ new CountDownLatch(1).await();
+ }
+ HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1",
Integer.parseInt(args[1])), 0);
+ server.createContext("/health/readiness", exchange -> {
+ Files.writeString(pidFile.resolveSibling("health-checked"), "checked");
+ byte[] body = "OK".getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(mode.equals("unready") ? 503 : 200,
body.length);
+ exchange.getResponseBody().write(body);
+ exchange.close();
+ });
+ server.start();
+
Files.writeString(pidFile.resolveSibling(QuickstartServerProcess.STARTED_FILE_NAME),
"started");
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
