Re: [PR] CASSSIDECAR-340: Add default process-based lifecycle provider to support the lifecycle API [cassandra-sidecar]

[via GitHub]

frankgh commented on code in PR #277:
URL: https://github.com/apache/cassandra-sidecar/pull/277#discussion_r2657783937


##########
server/src/main/java/org/apache/cassandra/sidecar/lifecycle/LifecycleManager.java:
##########
@@ -166,7 +166,7 @@ private Future<Void> submitStopTask(String instanceId)
             }
             catch (Exception e)
             {
-                LOG.error("Failed to stop instance {}: {}", instanceId, 
e.getMessage());
+                LOG.error("Failed to stop instance {}: {}", instanceId, 
e.getMessage(), e);

Review Comment:
   NIT:
   ```suggestion
                   LOG.error("Failed to stop instance {}", instanceId, e);
   ```



##########
server/src/main/java/org/apache/cassandra/sidecar/cluster/instance/InstanceMetadataImpl.java:
##########
@@ -136,6 +145,12 @@ public String stagingDir()
         return stagingDir;
     }
 
+    @Override
+    public String storageDir()
+    {
+        return  storageDir;

Review Comment:
   NIT (formatting)
   ```suggestion
           return storageDir;
   ```



##########
server/src/main/java/org/apache/cassandra/sidecar/cluster/instance/InstanceMetadataImpl.java:
##########
@@ -188,6 +203,16 @@ public InstanceMetrics metrics()
         return metrics;
     }
 
+    /**
+     * @return The lifecycle options for this Cassandra instance
+     */
+    @Override
+    @JsonProperty("lifecycle_options")

Review Comment:
   This class should not have the json property annotation. This class gets 
built from the configuration classes in 
https://github.com/apache/cassandra-sidecar/pull/277/files#diff-4de249cbcebfa53deecf4ac89d82a9419b32aea0a418fa56b5ed3b56b104a688R253
   ```suggestion
       @NonNull
   ```



##########
server/src/main/java/org/apache/cassandra/sidecar/lifecycle/LifecycleManager.java:
##########
@@ -148,7 +148,7 @@ private Future<Void> submitStartTask(String instanceId)
             }
             catch (Exception e)
             {
-                LOG.error("Failed to start instance {}: {}", instanceId, 
e.getMessage());
+                LOG.error("Failed to start instance {}: {}", instanceId, 
e.getMessage(), e);

Review Comment:
   NIT:
   ```suggestion
                   LOG.error("Failed to start instance {}", instanceId, e);
   ```



##########
server/src/main/java/org/apache/cassandra/sidecar/lifecycle/ProcessRuntimeConfiguration.java:
##########
@@ -0,0 +1,237 @@
+/*
+ * 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.cassandra.sidecar.lifecycle;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Represents the configuration for a Cassandra process instance.
+ */
+public class ProcessRuntimeConfiguration
+{
+    static class Builder
+    {
+        private String host;
+        private String cassandraHome;
+        private String cassandraConfDir;
+        private String cassandraLogDir;
+        private String storageDir;
+        private Map<String, String> jvmOpts;
+        private Map<String, String> envVars;
+
+        public Builder withHost(String host)
+        {
+            this.host = host;
+            return this;
+        }
+
+        public Builder withCassandraHome(String cassandraHome)
+        {
+            this.cassandraHome = cassandraHome;
+            return this;
+        }
+
+        public Builder withCassandraConfDir(String cassandraConfDir)
+        {
+            this.cassandraConfDir = cassandraConfDir;
+            return this;
+        }
+
+        public Builder withCassandraLogDir(String cassandraLogDir)
+        {
+            this.cassandraLogDir = cassandraLogDir;
+            return this;
+        }
+
+        public Builder withStorageDir(String storageDir)
+        {
+            this.storageDir = storageDir;
+            return this;
+        }
+
+        public Builder withJvmOptions(Map<String, String> jvmOptions)
+        {
+            this.jvmOpts = jvmOptions;
+            return this;
+        }
+
+        public Builder withEnvVars(Map<String, String> envVars)
+        {
+            this.envVars = envVars;
+            return this;
+        }
+
+        public ProcessRuntimeConfiguration build()
+        {
+            ProcessRuntimeConfiguration casCfg = new 
ProcessRuntimeConfiguration(host, cassandraHome, cassandraConfDir, 
cassandraLogDir, storageDir);
+            casCfg.extraEnvVars.putAll(envVars != null ? envVars : Map.of());
+            casCfg.extraJvmOpts.putAll(jvmOpts != null ? jvmOpts : Map.of());
+            return casCfg;
+        }
+    }
+
+    @NotNull
+    final String instanceName;
+    @NotNull
+    final Path cassandraHome;
+    @NotNull
+    final Path cassandraConfDir;
+    @Nullable
+    final String cassandraLogDir;
+    @Nullable
+    final String storageDir;
+    final Map<String, String> extraJvmOpts = new HashMap<>();
+    final Map<String, String> extraEnvVars = new HashMap<>();
+
+    private ProcessRuntimeConfiguration(@NotNull String host, String 
cassandraHome, String cassandraConfDir,
+                                        @Nullable String cassandraLogDir, 
@Nullable String storageDir)
+    {
+        this.instanceName = host;
+        this.cassandraHome = Path.of(cassandraHome);
+        this.cassandraConfDir = Path.of(cassandraConfDir);
+        this.cassandraLogDir = cassandraLogDir;
+        this.storageDir = storageDir;
+    }
+
+    public String instanceName()
+    {
+        return instanceName;
+    }
+
+    public Path cassandraHome()
+    {
+        return cassandraHome;
+    }
+
+    public String cassandraConf()
+    {
+        return cassandraConfDir.toString();
+    }
+
+    public Path cassandraBin()
+    {
+        return cassandraHome.resolve(Path.of("bin", "cassandra"));
+    }
+
+    public Path stopServerBin()
+    {
+        return cassandraHome.resolve(Path.of("bin", "stop-server"));
+    }
+
+    private Path cassandraYaml()
+    {
+        return cassandraConfDir.resolve(Path.of("cassandra.yaml"));
+    }
+
+    public void validateStart() throws IllegalArgumentException
+    {
+        // Check existence
+        if (!Files.isDirectory(cassandraHome))
+        {
+            throw new IllegalArgumentException("Cassandra home does not exist 
or is not a directory: " + cassandraHome);
+        }
+        if (!Files.isDirectory(cassandraConfDir))
+        {
+            throw new IllegalArgumentException("Cassandra configuration 
directory does not exist or is not a directory: " + cassandraConfDir);
+        }
+        if (!Files.isRegularFile(cassandraYaml()))
+        {
+            throw new IllegalArgumentException("Cassandra YAML configuration 
file does not exist: " + cassandraYaml());
+        }
+        if (!Files.isRegularFile(cassandraBin()))
+        {
+            throw new IllegalArgumentException("Cassandra binary does not 
exist or is not a regular file: " + cassandraBin());
+        }
+        // Check permissions
+        if (!Files.isExecutable(cassandraBin()))
+        {
+            throw new IllegalArgumentException("Cassandra binary is not 
executable: " + cassandraBin());
+        }
+        if (!Files.isReadable(cassandraConfDir))
+        {
+            throw new IllegalArgumentException("Cassandra configuration 
directory is not readable: " + cassandraConfDir);
+        }
+    }
+
+    public ProcessBuilder buildStartCommand(String pidFileLocation, String 
stdoutFileLocation, String stderrFileLocation)
+    {
+        validateStart();
+
+        List<String> startCassandraCmd = new ArrayList<>();
+        startCassandraCmd.add(cassandraBin().toString());
+        startCassandraCmd.add("-p");
+        startCassandraCmd.add(pidFileLocation);
+        for (Map.Entry<String, String> jvmOpt : extraJvmOpts.entrySet())
+        {
+            startCassandraCmd.add("-D" + jvmOpt.getKey() + "=" + 
jvmOpt.getValue());
+        }
+
+        // Override storage dir if present in sidecar configuration
+        if (storageDir != null)
+        {
+            startCassandraCmd.add("-D");
+            startCassandraCmd.add("cassandra.storagedir=" + storageDir);
+        }
+
+        ProcessBuilder processBuilder = new ProcessBuilder();
+        processBuilder.command(startCassandraCmd);
+
+        // Set environment variables
+        Map<String, String> env = processBuilder.environment();
+        env.put("CASSANDRA_HOME", cassandraHome().toString());
+        env.put("CASSANDRA_CONF", cassandraConf());
+        env.putAll(extraEnvVars);
+
+        // Only override CASSANDRA_LOG_DIR if it is set in the configuration
+        if (cassandraLogDir != null)
+        {
+            env.put("CASSANDRA_LOG_DIR", cassandraLogDir);
+        }
+
+        // Redirect output to logs
+        processBuilder.redirectOutput(ProcessBuilder.Redirect.to(new 
File(stdoutFileLocation)));
+        processBuilder.redirectError(ProcessBuilder.Redirect.to(new 
File(stderrFileLocation)));
+
+        // Set working directory
+        processBuilder.directory(cassandraHome().toFile());
+        return processBuilder;
+    }
+
+    public String toString()

Review Comment:
   NIT (style) : add override annotation
   ```suggestion
       @Override
       public String toString()
   ```



##########
server/src/main/java/org/apache/cassandra/sidecar/cluster/instance/InstanceMetadataImpl.java:
##########
@@ -23,10 +23,12 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.stream.Collectors;
 
 import com.codahale.metrics.MetricRegistry;
+import com.fasterxml.jackson.annotation.JsonProperty;

Review Comment:
   ```suggestion
   ```



##########
server/src/main/java/org/apache/cassandra/sidecar/lifecycle/ProcessLifecycleProvider.java:
##########
@@ -18,38 +18,354 @@
 
 package org.apache.cassandra.sidecar.lifecycle;
 
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.exceptions.ConfigurationException;
+import org.jetbrains.annotations.VisibleForTesting;
 
 /**
- * A {@link LifecycleProvider} that manages Cassandra instances as OS 
processes.
- * <p>
- * This implementation is a placeholder and is not yet implemented.
+ * Manage the lifecycle of Cassandra instances running on local processes
  */
 public class ProcessLifecycleProvider implements LifecycleProvider
 {
+    static final String OPT_CASSANDRA_HOME = "cassandra_home";
+    static final String OPT_CASSANDRA_CONF_DIR = "cassandra_conf_dir";
+    static final String OPT_CASSANDRA_LOG_DIR = "cassandra_log_dir";
+    static final String OPT_STATE_DIR = "state_dir";
+
+    protected static final Logger LOG = 
LoggerFactory.getLogger(ProcessLifecycleProvider.class);
+    public static final long CASSANDRA_PROCESS_TIMEOUT_MS = 
Long.getLong("cassandra.sidecar.lifecycle.process.timeout.ms", 120_000L);

Review Comment:
   Any reason why this is configured as a system property instead of having it 
configured in the sidecar.yaml file?



##########
server/src/main/java/org/apache/cassandra/sidecar/lifecycle/ProcessLifecycleProvider.java:
##########
@@ -18,38 +18,354 @@
 
 package org.apache.cassandra.sidecar.lifecycle;
 
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.exceptions.ConfigurationException;
+import org.jetbrains.annotations.VisibleForTesting;
 
 /**
- * A {@link LifecycleProvider} that manages Cassandra instances as OS 
processes.
- * <p>
- * This implementation is a placeholder and is not yet implemented.
+ * Manage the lifecycle of Cassandra instances running on local processes
  */
 public class ProcessLifecycleProvider implements LifecycleProvider
 {
+    static final String OPT_CASSANDRA_HOME = "cassandra_home";
+    static final String OPT_CASSANDRA_CONF_DIR = "cassandra_conf_dir";
+    static final String OPT_CASSANDRA_LOG_DIR = "cassandra_log_dir";
+    static final String OPT_STATE_DIR = "state_dir";
+
+    protected static final Logger LOG = 
LoggerFactory.getLogger(ProcessLifecycleProvider.class);

Review Comment:
   Let's keep the same naming across the project for consistency 
   ```suggestion
       protected static final Logger LOGGER = 
LoggerFactory.getLogger(ProcessLifecycleProvider.class);
   ```



##########
server/src/main/java/org/apache/cassandra/sidecar/lifecycle/ProcessLifecycleProvider.java:
##########
@@ -18,38 +18,354 @@
 
 package org.apache.cassandra.sidecar.lifecycle;
 
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.exceptions.ConfigurationException;
+import org.jetbrains.annotations.VisibleForTesting;
 
 /**
- * A {@link LifecycleProvider} that manages Cassandra instances as OS 
processes.
- * <p>
- * This implementation is a placeholder and is not yet implemented.
+ * Manage the lifecycle of Cassandra instances running on local processes
  */
 public class ProcessLifecycleProvider implements LifecycleProvider
 {
+    static final String OPT_CASSANDRA_HOME = "cassandra_home";
+    static final String OPT_CASSANDRA_CONF_DIR = "cassandra_conf_dir";
+    static final String OPT_CASSANDRA_LOG_DIR = "cassandra_log_dir";
+    static final String OPT_STATE_DIR = "state_dir";
+
+    protected static final Logger LOG = 
LoggerFactory.getLogger(ProcessLifecycleProvider.class);
+    public static final long CASSANDRA_PROCESS_TIMEOUT_MS = 
Long.getLong("cassandra.sidecar.lifecycle.process.timeout.ms", 120_000L);
+
+    private final String lifecycleDir;
+    private final String defaultCassandraHome;
+    private final Map<String, String> defaultJvmProperties = new HashMap<>();
+    private final Map<String, String> defaultEnvVars = new HashMap<>();
+
     public ProcessLifecycleProvider(Map<String, String> params)
     {
-        // Params unused for now
+        // Extract any JVM properties or environment variables from the params
+        for (Map.Entry<String, String> entry : params.entrySet())
+        {
+            if (entry.getKey().startsWith("cassandra."))
+            {
+                defaultJvmProperties.put(entry.getKey(), entry.getValue());
+            }
+            else if (entry.getKey().startsWith("env."))
+            {
+                defaultEnvVars.put(entry.getKey().replaceAll("env.", ""), 
entry.getValue());
+            }
+        }
+        this.lifecycleDir = params.get(OPT_STATE_DIR);
+        this.defaultCassandraHome = params.get(OPT_CASSANDRA_HOME);
+        validateConfiguration();
+    }
+
+    private void validateConfiguration()
+    {
+        if (lifecycleDir == null || lifecycleDir.isEmpty())
+        {
+            throw new ConfigurationException("Configuration property '" + 
OPT_STATE_DIR + "' must be set for ProcessLifecycleProvider");
+        }
+        if (defaultCassandraHome == null || defaultCassandraHome.isEmpty())
+        {
+            throw new ConfigurationException("Configuration property '" + 
OPT_CASSANDRA_HOME + "' must be set for ProcessLifecycleProvider");
+        }
+
+        Path stateDir = Path.of(lifecycleDir);
+        if (!Files.isDirectory(stateDir))
+        {
+            throw new ConfigurationException("State directory '" + 
lifecycleDir + "' does not exist or is not a directory");
+        }
+        if (!Files.isWritable(stateDir))
+        {
+            throw new ConfigurationException("State directory '" + 
lifecycleDir + "' is not writable");
+        }
+
+        Path cassandraHomePath = Path.of(defaultCassandraHome);
+        if (!Files.isDirectory(cassandraHomePath))
+        {
+            throw new ConfigurationException("Cassandra home '" + 
defaultCassandraHome + "' does not exist or is not a directory");
+        }
+        if (!Files.isReadable(cassandraHomePath))
+        {
+            throw new ConfigurationException("Cassandra home '" + 
defaultCassandraHome + "' is not readable");
+        }
     }
 
     @Override
     public void start(InstanceMetadata instance)
     {
-        throw new UnsupportedOperationException("Not implemented yet");
+        if (isCassandraProcessRunning(instance))
+        {
+            LOG.info("Cassandra instance {} is already running.", 
instance.host());
+            return;
+        }
+        startCassandra(instance);
     }
 
     @Override
     public void stop(InstanceMetadata instance)
     {
-        throw new UnsupportedOperationException("Not implemented yet");
-
+        if (!isCassandraProcessRunning(instance))
+        {
+            LOG.info("Cassandra instance {} is already stopped.", 
instance.host());
+            return;
+        }
+        stopCassandra(instance);
     }
 
     @Override
     public boolean isRunning(InstanceMetadata instance)
     {
-        throw new UnsupportedOperationException("Not implemented yet");
+        return isCassandraProcessRunning(instance);
+    }
+
+    private void startCassandra(InstanceMetadata instance)
+    {
+        ProcessRuntimeConfiguration runtimeConfig = 
getRuntimeConfiguration(instance);
+        try
+        {
+            String stdoutLocation = 
getStdoutLocation(runtimeConfig.instanceName());
+            String stderrLocation = 
getStderrLocation(runtimeConfig.instanceName());
+            String pidFileLocation = 
getPidFileLocation(runtimeConfig.instanceName());
+            ProcessBuilder processBuilder = 
runtimeConfig.buildStartCommand(pidFileLocation,
+                                                                            
stdoutLocation,
+                                                                            
stderrLocation);
+            LOG.info("Starting Cassandra instance {} with command: {}", 
runtimeConfig.instanceName(), processBuilder.command());
+
+            Process process = processBuilder.start();
+            process.waitFor(CASSANDRA_PROCESS_TIMEOUT_MS, 
TimeUnit.MILLISECONDS); // blocking call, make async?
+
+            if (isCassandraProcessRunning(instance))
+            {
+                LOG.info("Started Cassandra instance {} with PID {}", 
runtimeConfig.instanceName(), readPidFromFile(Path.of(pidFileLocation)));
+            }
+            else
+            {
+                throw new RuntimeException("Failed to start Cassandra instance 
" + runtimeConfig.instanceName() +
+                                           ". Check stdout at " + 
stdoutLocation + " and stderr at " + stderrLocation);
+            }
+        }
+        catch (Throwable t)
+        {
+            throw new RuntimeException("Failed to start Cassandra instance " + 
runtimeConfig.instanceName() + " due to " + t.getMessage(), t);
+        }
+    }
+
+    private void stopCassandra(InstanceMetadata instance)
+    {
+        ProcessRuntimeConfiguration casCfg = getRuntimeConfiguration(instance);
+        try
+        {
+            String pidFileLocation = getPidFileLocation(casCfg.instanceName());
+            Long pid = readPidFromFile(Path.of(pidFileLocation));
+            Optional<ProcessHandle> processHandle = ProcessHandle.of(pid);
+            if (processHandle.isPresent())
+            {
+                LOG.info("Stopping process of Cassandra instance {} with PID 
{}.", casCfg.instanceName(), pid);
+                CompletableFuture<ProcessHandle> terminationFuture = 
processHandle.get().onExit();
+                processHandle.get().destroy();
+                try
+                {
+                    terminationFuture.get(CASSANDRA_PROCESS_TIMEOUT_MS, 
TimeUnit.MILLISECONDS);
+                }
+                catch (TimeoutException e)
+                {
+                    LOG.warn("Process {} did not terminate within timeout, 
forcing destroy.", pid);
+                    boolean destroyed = processHandle.get().destroyForcibly();
+                    if (!destroyed)
+                    {
+                        throw new RuntimeException("Failed to forcibly destroy 
process " + pid +
+                                                   " for Cassandra instance " 
+ casCfg.instanceName(), e);
+                    }
+                    LOG.info("Process {} was forcibly destroyed.", pid);
+                }
+                Files.deleteIfExists(Path.of(pidFileLocation));
+            }
+            else
+            {
+                LOG.warn("No process running for Cassandra instance {} with 
PID {}.", casCfg.instanceName(), pid);
+            }
+        }
+        catch (Throwable t)
+        {
+            throw new RuntimeException("Failed to stop process for Cassandra 
instance " + casCfg.instanceName() + " due to " + t.getMessage(), t);
+        }
+    }
+
+    @VisibleForTesting
+    protected ProcessRuntimeConfiguration 
getRuntimeConfiguration(InstanceMetadata instance)
+    {
+        String cassandraHome = 
Optional.ofNullable(instance.lifecycleOptions().get(OPT_CASSANDRA_HOME))
+                                       .orElse(defaultCassandraHome);
+        String cassandraConfDir = 
instance.lifecycleOptions().get(OPT_CASSANDRA_CONF_DIR);
+        String cassandraLogDir = 
instance.lifecycleOptions().get(OPT_CASSANDRA_LOG_DIR);
+        return new ProcessRuntimeConfiguration.Builder()
+                                        .withHost(instance.host())
+                                        .withCassandraHome(cassandraHome)
+                                        .withCassandraConfDir(cassandraConfDir)
+                                        .withCassandraLogDir(cassandraLogDir)
+                                        .withStorageDir(instance.storageDir())
+                                        .withJvmOptions(defaultJvmProperties)
+                                        .withEnvVars(defaultEnvVars)
+                                        .build();
+    }
+
+    /**
+     * Checks whether a Cassandra instance is currently running as a local 
process
+     * and automatically cleans up stale PID files.
+     *
+     * Performs four validation steps:
+     * 1. Verifies the PID file exists and is readable. Returns false if not 
found.
+     * 2. Reads the PID and checks if the process is alive. Returns false and 
deletes the
+     *    PID file if the process no longer exists or is not alive.
+     * 3. Verifies the process is a Cassandra instance by checking for
+     *    org.apache.cassandra.service.CassandraDaemon in the command line. 
Returns true if
+     *    the command line contains the Cassandra daemon class or cannot be 
determined.
+     * 4. If the process is running but is not a Cassandra process, returns 
false and deletes
+     *    the stale PID file.
+     *
+     * @param instance the instance metadata containing host information
+     * @return true if the instance is running as a Cassandra process, false 
otherwise
+     */
+    private boolean isCassandraProcessRunning(InstanceMetadata instance)

Review Comment:
   can we debug at warn level here instead of debug? Any reason why we debug 
was used?



##########
server/src/main/java/org/apache/cassandra/sidecar/lifecycle/ProcessLifecycleProvider.java:
##########
@@ -18,38 +18,354 @@
 
 package org.apache.cassandra.sidecar.lifecycle;
 
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.exceptions.ConfigurationException;
+import org.jetbrains.annotations.VisibleForTesting;
 
 /**
- * A {@link LifecycleProvider} that manages Cassandra instances as OS 
processes.
- * <p>
- * This implementation is a placeholder and is not yet implemented.
+ * Manage the lifecycle of Cassandra instances running on local processes
  */
 public class ProcessLifecycleProvider implements LifecycleProvider
 {
+    static final String OPT_CASSANDRA_HOME = "cassandra_home";
+    static final String OPT_CASSANDRA_CONF_DIR = "cassandra_conf_dir";
+    static final String OPT_CASSANDRA_LOG_DIR = "cassandra_log_dir";
+    static final String OPT_STATE_DIR = "state_dir";
+
+    protected static final Logger LOG = 
LoggerFactory.getLogger(ProcessLifecycleProvider.class);
+    public static final long CASSANDRA_PROCESS_TIMEOUT_MS = 
Long.getLong("cassandra.sidecar.lifecycle.process.timeout.ms", 120_000L);
+
+    private final String lifecycleDir;
+    private final String defaultCassandraHome;
+    private final Map<String, String> defaultJvmProperties = new HashMap<>();
+    private final Map<String, String> defaultEnvVars = new HashMap<>();
+
     public ProcessLifecycleProvider(Map<String, String> params)
     {
-        // Params unused for now
+        // Extract any JVM properties or environment variables from the params
+        for (Map.Entry<String, String> entry : params.entrySet())
+        {
+            if (entry.getKey().startsWith("cassandra."))
+            {
+                defaultJvmProperties.put(entry.getKey(), entry.getValue());
+            }
+            else if (entry.getKey().startsWith("env."))
+            {
+                defaultEnvVars.put(entry.getKey().replaceAll("env.", ""), 
entry.getValue());

Review Comment:
   can we use replace first instead here?
   ```suggestion
                   defaultEnvVars.put(entry.getKey().replaceFirst("^env\\.", 
""), entry.getValue());
   ```



##########
conf/sidecar.yaml:
##########
@@ -455,3 +458,15 @@ live_migration:
   migration_map:                          # Map of source and destination 
Cassandra instances
 #    localhost1: localhost4               # This entry says that localhost1 
will be migrated to localhost4
   max_concurrent_downloads: 20            # Maximum number of concurrent 
downloads allowed
+
+# Configuration to allow sidecar start and stop Cassandra instances via the 
lifecycle API (disabled by default)
+lifecycle:
+  enabled: false
+  provider:
+    class_name: org.apache.cassandra.sidecar.lifecycle.ProcessLifecycleProvider
+    parameters:
+      state_dir: /var/lib/cassandra-sidecar/lifecycle # The directory where 
the process state is stored
+      cassandra_home: /opt/cassandra # The default Cassandra installation 
directory

Review Comment:
   similar to how we configure environment variables can we also use a prefix 
for java system properties. This would allow for configuring system properties 
that apply to the Cassandra process but are not Cassandra-core specific, for 
example a JMX property could be configured like this: 
`sys.com.sun.management.jmxremote.authenticate: true`.



##########
server/src/main/java/org/apache/cassandra/sidecar/lifecycle/ProcessRuntimeConfiguration.java:
##########
@@ -0,0 +1,237 @@
+/*
+ * 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.cassandra.sidecar.lifecycle;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Represents the configuration for a Cassandra process instance.
+ */
+public class ProcessRuntimeConfiguration
+{
+    static class Builder

Review Comment:
   I think we should stick to the existing Builder  pattern that we have in the 
repository implementing `DataObjectBuilder`. Also, the `with` pattern in the 
builder is also diverging from the existing codebase.



##########
examples/lifecycle/setup.sh:
##########
@@ -0,0 +1,54 @@
+#!/bin/bash
+#
+# 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.
+#
+set -eu
+
+SCRIPT_DIR=$(realpath "$(dirname "$0")")
+CASSANDRA_VERSION=4.1.10
+CASSANDRA_DIR="apache-cassandra-${CASSANDRA_VERSION}"
+TARBALL_NAME="${CASSANDRA_DIR}-bin.tar.gz"
+TARBALL_URL="https://dlcdn.apache.org/cassandra/${CASSANDRA_VERSION}/${TARBALL_NAME}";
+NODE_DIR="${SCRIPT_DIR}/nodes/localhost"
+
+SIDECAR_YAML="${SCRIPT_DIR}/conf/sidecar.yaml"
+SIDECAR_YAML_TEMPLATE="${SCRIPT_DIR}/conf/sidecar.yaml.template"
+CASSANDRA_HOME="${NODE_DIR}/opt/${CASSANDRA_DIR}"
+CASSANDRA_LOG_DIR="${NODE_DIR}/var/log/cassandra"
+CASSANDRA_CONF="${NODE_DIR}/etc/cassandra"
+CASSANDRA_STORAGE_DIR="${NODE_DIR}/var/lib/cassandra"
+SIDECAR_LIFECYCLE_DIR="${NODE_DIR}/var/lib/cassandra-sidecar/lifecycle"
+TMP_DIR="${NODE_DIR}/tmp"
+
+echo "Creating directories"
+mkdir -p ${CASSANDRA_HOME} ${CASSANDRA_LOG_DIR} ${CASSANDRA_CONF} 
${CASSANDRA_STORAGE_DIR} ${SIDECAR_LIFECYCLE_DIR} ${TMP_DIR}
+
+if [ -f ${CASSANDRA_HOME}/bin/cassandra ]; then
+  echo "Cassandra already installed at ${CASSANDRA_HOME}, skipping install"
+else
+  echo "Installing Cassandra at ${CASSANDRA_HOME}"
+  echo "Downloading ${TARBALL_URL}"
+  wget -P ${TMP_DIR} ${TARBALL_URL}
+
+  echo "Extracting tarball"
+  tar -xvzf ${TMP_DIR}/${TARBALL_NAME} -C $(dirname $CASSANDRA_HOME)
+fi
+
+echo "Creating configuration"

Review Comment:
   NIT:
   ```suggestion
   echo "Creating Sidecar configuration"
   ```



##########
examples/lifecycle/setup.sh:
##########
@@ -0,0 +1,54 @@
+#!/bin/bash
+#
+# 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.
+#
+set -eu
+
+SCRIPT_DIR=$(realpath "$(dirname "$0")")
+CASSANDRA_VERSION=4.1.10
+CASSANDRA_DIR="apache-cassandra-${CASSANDRA_VERSION}"
+TARBALL_NAME="${CASSANDRA_DIR}-bin.tar.gz"
+TARBALL_URL="https://dlcdn.apache.org/cassandra/${CASSANDRA_VERSION}/${TARBALL_NAME}";
+NODE_DIR="${SCRIPT_DIR}/nodes/localhost"
+
+SIDECAR_YAML="${SCRIPT_DIR}/conf/sidecar.yaml"
+SIDECAR_YAML_TEMPLATE="${SCRIPT_DIR}/conf/sidecar.yaml.template"
+CASSANDRA_HOME="${NODE_DIR}/opt/${CASSANDRA_DIR}"
+CASSANDRA_LOG_DIR="${NODE_DIR}/var/log/cassandra"
+CASSANDRA_CONF="${NODE_DIR}/etc/cassandra"
+CASSANDRA_STORAGE_DIR="${NODE_DIR}/var/lib/cassandra"
+SIDECAR_LIFECYCLE_DIR="${NODE_DIR}/var/lib/cassandra-sidecar/lifecycle"
+TMP_DIR="${NODE_DIR}/tmp"
+
+echo "Creating directories"
+mkdir -p ${CASSANDRA_HOME} ${CASSANDRA_LOG_DIR} ${CASSANDRA_CONF} 
${CASSANDRA_STORAGE_DIR} ${SIDECAR_LIFECYCLE_DIR} ${TMP_DIR}
+
+if [ -f ${CASSANDRA_HOME}/bin/cassandra ]; then
+  echo "Cassandra already installed at ${CASSANDRA_HOME}, skipping install"
+else
+  echo "Installing Cassandra at ${CASSANDRA_HOME}"
+  echo "Downloading ${TARBALL_URL}"
+  wget -P ${TMP_DIR} ${TARBALL_URL}
+
+  echo "Extracting tarball"
+  tar -xvzf ${TMP_DIR}/${TARBALL_NAME} -C $(dirname $CASSANDRA_HOME)
+fi
+
+echo "Creating configuration"
+cp -r ${CASSANDRA_HOME}/conf/* ${CASSANDRA_CONF}
+sed "s#\$cassandraHome#${CASSANDRA_HOME}#g" ${SIDECAR_YAML_TEMPLATE} > 
${SIDECAR_YAML}
+sed -i '' "s#\$baseDir#${NODE_DIR}#g" ${SIDECAR_YAML}

Review Comment:
   NIT add message when setup is complete:
   ```suggestion
   sed -i '' "s#\$baseDir#${NODE_DIR}#g" ${SIDECAR_YAML}
   
   echo "Setup complete!"
   ```



##########
examples/lifecycle/setup.sh:
##########
@@ -0,0 +1,54 @@
+#!/bin/bash
+#
+# 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.
+#
+set -eu
+
+SCRIPT_DIR=$(realpath "$(dirname "$0")")
+CASSANDRA_VERSION=4.1.10
+CASSANDRA_DIR="apache-cassandra-${CASSANDRA_VERSION}"
+TARBALL_NAME="${CASSANDRA_DIR}-bin.tar.gz"
+TARBALL_URL="https://dlcdn.apache.org/cassandra/${CASSANDRA_VERSION}/${TARBALL_NAME}";
+NODE_DIR="${SCRIPT_DIR}/nodes/localhost"
+
+SIDECAR_YAML="${SCRIPT_DIR}/conf/sidecar.yaml"
+SIDECAR_YAML_TEMPLATE="${SCRIPT_DIR}/conf/sidecar.yaml.template"
+CASSANDRA_HOME="${NODE_DIR}/opt/${CASSANDRA_DIR}"
+CASSANDRA_LOG_DIR="${NODE_DIR}/var/log/cassandra"
+CASSANDRA_CONF="${NODE_DIR}/etc/cassandra"
+CASSANDRA_STORAGE_DIR="${NODE_DIR}/var/lib/cassandra"
+SIDECAR_LIFECYCLE_DIR="${NODE_DIR}/var/lib/cassandra-sidecar/lifecycle"
+TMP_DIR="${NODE_DIR}/tmp"
+
+echo "Creating directories"
+mkdir -p ${CASSANDRA_HOME} ${CASSANDRA_LOG_DIR} ${CASSANDRA_CONF} 
${CASSANDRA_STORAGE_DIR} ${SIDECAR_LIFECYCLE_DIR} ${TMP_DIR}
+
+if [ -f ${CASSANDRA_HOME}/bin/cassandra ]; then
+  echo "Cassandra already installed at ${CASSANDRA_HOME}, skipping install"
+else
+  echo "Installing Cassandra at ${CASSANDRA_HOME}"
+  echo "Downloading ${TARBALL_URL}"
+  wget -P ${TMP_DIR} ${TARBALL_URL}
+
+  echo "Extracting tarball"

Review Comment:
   NIT
   ```suggestion
     echo "Extracting Cassandra tarball"
   ```



##########
server/src/main/java/org/apache/cassandra/sidecar/lifecycle/ProcessRuntimeConfiguration.java:
##########
@@ -0,0 +1,237 @@
+/*
+ * 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.cassandra.sidecar.lifecycle;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Represents the configuration for a Cassandra process instance.
+ */
+public class ProcessRuntimeConfiguration
+{
+    static class Builder
+    {
+        private String host;
+        private String cassandraHome;
+        private String cassandraConfDir;
+        private String cassandraLogDir;
+        private String storageDir;
+        private Map<String, String> jvmOpts;
+        private Map<String, String> envVars;
+
+        public Builder withHost(String host)
+        {
+            this.host = host;
+            return this;
+        }
+
+        public Builder withCassandraHome(String cassandraHome)
+        {
+            this.cassandraHome = cassandraHome;
+            return this;
+        }
+
+        public Builder withCassandraConfDir(String cassandraConfDir)
+        {
+            this.cassandraConfDir = cassandraConfDir;
+            return this;
+        }
+
+        public Builder withCassandraLogDir(String cassandraLogDir)
+        {
+            this.cassandraLogDir = cassandraLogDir;
+            return this;
+        }
+
+        public Builder withStorageDir(String storageDir)
+        {
+            this.storageDir = storageDir;
+            return this;
+        }
+
+        public Builder withJvmOptions(Map<String, String> jvmOptions)
+        {
+            this.jvmOpts = jvmOptions;
+            return this;
+        }
+
+        public Builder withEnvVars(Map<String, String> envVars)
+        {
+            this.envVars = envVars;
+            return this;
+        }
+
+        public ProcessRuntimeConfiguration build()
+        {
+            ProcessRuntimeConfiguration casCfg = new 
ProcessRuntimeConfiguration(host, cassandraHome, cassandraConfDir, 
cassandraLogDir, storageDir);
+            casCfg.extraEnvVars.putAll(envVars != null ? envVars : Map.of());
+            casCfg.extraJvmOpts.putAll(jvmOpts != null ? jvmOpts : Map.of());
+            return casCfg;
+        }
+    }
+
+    @NotNull
+    final String instanceName;
+    @NotNull
+    final Path cassandraHome;
+    @NotNull
+    final Path cassandraConfDir;
+    @Nullable
+    final String cassandraLogDir;
+    @Nullable
+    final String storageDir;
+    final Map<String, String> extraJvmOpts = new HashMap<>();
+    final Map<String, String> extraEnvVars = new HashMap<>();
+
+    private ProcessRuntimeConfiguration(@NotNull String host, String 
cassandraHome, String cassandraConfDir,
+                                        @Nullable String cassandraLogDir, 
@Nullable String storageDir)
+    {
+        this.instanceName = host;
+        this.cassandraHome = Path.of(cassandraHome);
+        this.cassandraConfDir = Path.of(cassandraConfDir);
+        this.cassandraLogDir = cassandraLogDir;
+        this.storageDir = storageDir;
+    }
+
+    public String instanceName()
+    {
+        return instanceName;
+    }
+
+    public Path cassandraHome()
+    {
+        return cassandraHome;
+    }
+
+    public String cassandraConf()
+    {
+        return cassandraConfDir.toString();
+    }
+
+    public Path cassandraBin()
+    {
+        return cassandraHome.resolve(Path.of("bin", "cassandra"));
+    }
+
+    public Path stopServerBin()
+    {
+        return cassandraHome.resolve(Path.of("bin", "stop-server"));
+    }
+
+    private Path cassandraYaml()
+    {
+        return cassandraConfDir.resolve(Path.of("cassandra.yaml"));
+    }
+
+    public void validateStart() throws IllegalArgumentException

Review Comment:
   should we throw a ConfigurationException here instead?



##########
server/src/main/java/org/apache/cassandra/sidecar/lifecycle/ProcessLifecycleProvider.java:
##########
@@ -18,38 +18,232 @@
 
 package org.apache.cassandra.sidecar.lifecycle;
 
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.exceptions.ConfigurationException;
+import org.jetbrains.annotations.VisibleForTesting;
 
 /**
- * A {@link LifecycleProvider} that manages Cassandra instances as OS 
processes.
- * <p>
- * This implementation is a placeholder and is not yet implemented.
+ * Manage the lifecycle of Cassandra instances running on local processes
  */
 public class ProcessLifecycleProvider implements LifecycleProvider
 {
+    static final String OPT_CASSANDRA_HOME = "cassandra_home";
+    static final String OPT_CASSANDRA_CONF_DIR = "cassandra_conf_dir";
+    static final String OPT_CASSANDRA_LOG_DIR = "cassandra_log_dir";
+    static final String OPT_STATE_DIR = "state_dir";
+
+    protected static final Logger LOG = 
LoggerFactory.getLogger(ProcessLifecycleProvider.class);
+    public static final long CASSANDRA_PROCESS_TIMEOUT_MS = 
Long.getLong("cassandra.sidecar.lifecycle.process.timeout.ms", 120_000L);
+
+    private final String lifecycleDir;
+    private final String defaultCassandraHome;
+    private final Map<String, String> defaultJvmProperties = new HashMap<>();
+    private final Map<String, String> defaultEnvVars = new HashMap<>();
+
     public ProcessLifecycleProvider(Map<String, String> params)
     {
-        // Params unused for now
+        // Extract any JVM properties or environment variables from the params
+        for (Map.Entry<String, String> entry : params.entrySet())
+        {
+            if (entry.getKey().startsWith("cassandra."))
+            {
+                defaultJvmProperties.put(entry.getKey(), entry.getValue());
+            }
+            else if (entry.getKey().startsWith("env."))
+            {
+                defaultEnvVars.put(entry.getKey().replaceAll("env.", ""), 
entry.getValue());
+            }
+        }
+        this.lifecycleDir = params.get(OPT_STATE_DIR);
+        this.defaultCassandraHome = params.get(OPT_CASSANDRA_HOME);
+        if (lifecycleDir == null || lifecycleDir.isEmpty())
+        {
+            throw new ConfigurationException("Configuration property '" + 
OPT_STATE_DIR + "' must be set for ProcessLifecycleProvider");
+        }
+        if (defaultCassandraHome == null || defaultCassandraHome.isEmpty())
+        {
+            throw new ConfigurationException("Configuration property '" + 
OPT_CASSANDRA_HOME + "' must be set for ProcessLifecycleProvider");
+        }
     }
 
     @Override
     public void start(InstanceMetadata instance)
     {
-        throw new UnsupportedOperationException("Not implemented yet");
+        if (isCassandraProcessRunning(instance))
+        {
+            LOG.info("Cassandra instance {} is already running.", 
instance.host());
+            return;
+        }
+        startCassandra(instance);
     }
 
     @Override
     public void stop(InstanceMetadata instance)
     {
-        throw new UnsupportedOperationException("Not implemented yet");
-
+        if (!isCassandraProcessRunning(instance))
+        {
+            LOG.info("Cassandra instance {} is already stopped.", 
instance.host());
+            return;
+        }
+        stopCassandra(instance);
     }
 
     @Override
     public boolean isRunning(InstanceMetadata instance)
     {
-        throw new UnsupportedOperationException("Not implemented yet");
+        return isCassandraProcessRunning(instance);
+    }
+
+    private void startCassandra(InstanceMetadata instance)
+    {
+        ProcessRuntimeConfiguration runtimeConfig = 
getRuntimeConfiguration(instance);
+        try
+        {
+            String stdoutLocation = 
getStdoutLocation(runtimeConfig.instanceName());
+            String stderrLocation = 
getStderrLocation(runtimeConfig.instanceName());
+            String pidFileLocation = 
getPidFileLocation(runtimeConfig.instanceName());
+            ProcessBuilder processBuilder = 
runtimeConfig.buildStartCommand(pidFileLocation,

Review Comment:
   I agree with Paulo here, I think it's fine to keep a single stdout / stderr 
file for the startup of the process. I'd expect minimal log entries in these 
files. 



##########
server/src/main/java/org/apache/cassandra/sidecar/cluster/instance/InstanceMetadataImpl.java:
##########
@@ -89,6 +94,10 @@ protected InstanceMetadataImpl(Builder builder)
         hintsDir = builder.resolveHintsDir();
         savedCachesDir = builder.resolveSavedCachesDir();
         localSystemDataFileDir = 
FileUtils.maybeResolveHomeDirectory(builder.localSystemDataFileDir);
+        lifecycleOptions = builder.lifecycleOptions != null
+                                 ? 
Collections.unmodifiableMap(builder.lifecycleOptions)
+                                 : Collections.emptyMap();

Review Comment:
   NIT (formatting)
   ```suggestion
           lifecycleOptions = builder.lifecycleOptions != null
                              ? 
Collections.unmodifiableMap(builder.lifecycleOptions)
                              : Collections.emptyMap();
   ```



##########
server/src/main/java/org/apache/cassandra/sidecar/lifecycle/ProcessRuntimeConfiguration.java:
##########
@@ -0,0 +1,237 @@
+/*
+ * 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.cassandra.sidecar.lifecycle;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Represents the configuration for a Cassandra process instance.
+ */
+public class ProcessRuntimeConfiguration
+{
+    static class Builder
+    {
+        private String host;
+        private String cassandraHome;
+        private String cassandraConfDir;
+        private String cassandraLogDir;
+        private String storageDir;
+        private Map<String, String> jvmOpts;
+        private Map<String, String> envVars;
+
+        public Builder withHost(String host)
+        {
+            this.host = host;
+            return this;
+        }
+
+        public Builder withCassandraHome(String cassandraHome)
+        {
+            this.cassandraHome = cassandraHome;
+            return this;
+        }
+
+        public Builder withCassandraConfDir(String cassandraConfDir)
+        {
+            this.cassandraConfDir = cassandraConfDir;
+            return this;
+        }
+
+        public Builder withCassandraLogDir(String cassandraLogDir)
+        {
+            this.cassandraLogDir = cassandraLogDir;
+            return this;
+        }
+
+        public Builder withStorageDir(String storageDir)
+        {
+            this.storageDir = storageDir;
+            return this;
+        }
+
+        public Builder withJvmOptions(Map<String, String> jvmOptions)
+        {
+            this.jvmOpts = jvmOptions;
+            return this;
+        }
+
+        public Builder withEnvVars(Map<String, String> envVars)
+        {
+            this.envVars = envVars;
+            return this;
+        }
+
+        public ProcessRuntimeConfiguration build()
+        {
+            ProcessRuntimeConfiguration casCfg = new 
ProcessRuntimeConfiguration(host, cassandraHome, cassandraConfDir, 
cassandraLogDir, storageDir);
+            casCfg.extraEnvVars.putAll(envVars != null ? envVars : Map.of());
+            casCfg.extraJvmOpts.putAll(jvmOpts != null ? jvmOpts : Map.of());
+            return casCfg;
+        }
+    }
+
+    @NotNull
+    final String instanceName;
+    @NotNull
+    final Path cassandraHome;
+    @NotNull
+    final Path cassandraConfDir;
+    @Nullable
+    final String cassandraLogDir;
+    @Nullable
+    final String storageDir;
+    final Map<String, String> extraJvmOpts = new HashMap<>();
+    final Map<String, String> extraEnvVars = new HashMap<>();
+
+    private ProcessRuntimeConfiguration(@NotNull String host, String 
cassandraHome, String cassandraConfDir,
+                                        @Nullable String cassandraLogDir, 
@Nullable String storageDir)
+    {
+        this.instanceName = host;
+        this.cassandraHome = Path.of(cassandraHome);
+        this.cassandraConfDir = Path.of(cassandraConfDir);
+        this.cassandraLogDir = cassandraLogDir;
+        this.storageDir = storageDir;
+    }
+
+    public String instanceName()
+    {
+        return instanceName;
+    }
+
+    public Path cassandraHome()
+    {
+        return cassandraHome;
+    }
+
+    public String cassandraConf()
+    {
+        return cassandraConfDir.toString();
+    }
+
+    public Path cassandraBin()
+    {
+        return cassandraHome.resolve(Path.of("bin", "cassandra"));
+    }
+
+    public Path stopServerBin()

Review Comment:
   this is unused, do we need it? Looking at the implementation of the script, 
it looks like it only prints information and it doesn't actually do anything. 
So maybe remove this since it won't be used



##########
.circleci/config.yml:
##########
@@ -150,8 +154,10 @@ jobs:
       - checkout
       - attach_workspace:
           at: dtest-jars
+      - attach_workspace:
+          at: cassandra-tarballs
       - run: ./scripts/install-shaded-dtest-jar-local.sh
-      - run: ./gradlew --no-daemon -PdtestVersion=4.0.16 
-Dcassandra.sidecar.versions_to_test="4.0" integrationTestHeavyWeight 
--stacktrace
+      - run: ./gradlew --no-daemon -PdtestVersion=4.0.16 -PtarballVersion=4.0 
-Dcassandra.sidecar.versions_to_test="4.0" jar integrationTestHeavyWeight 
--stacktrace

Review Comment:
   just out of curiosity why do we need to execute `jar` before running the 
integration tests?



##########
server/src/main/java/org/apache/cassandra/sidecar/lifecycle/ProcessLifecycleProvider.java:
##########
@@ -18,38 +18,354 @@
 
 package org.apache.cassandra.sidecar.lifecycle;
 
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.exceptions.ConfigurationException;
+import org.jetbrains.annotations.VisibleForTesting;
 
 /**
- * A {@link LifecycleProvider} that manages Cassandra instances as OS 
processes.
- * <p>
- * This implementation is a placeholder and is not yet implemented.
+ * Manage the lifecycle of Cassandra instances running on local processes
  */
 public class ProcessLifecycleProvider implements LifecycleProvider
 {
+    static final String OPT_CASSANDRA_HOME = "cassandra_home";
+    static final String OPT_CASSANDRA_CONF_DIR = "cassandra_conf_dir";
+    static final String OPT_CASSANDRA_LOG_DIR = "cassandra_log_dir";
+    static final String OPT_STATE_DIR = "state_dir";
+
+    protected static final Logger LOG = 
LoggerFactory.getLogger(ProcessLifecycleProvider.class);
+    public static final long CASSANDRA_PROCESS_TIMEOUT_MS = 
Long.getLong("cassandra.sidecar.lifecycle.process.timeout.ms", 120_000L);
+
+    private final String lifecycleDir;
+    private final String defaultCassandraHome;
+    private final Map<String, String> defaultJvmProperties = new HashMap<>();
+    private final Map<String, String> defaultEnvVars = new HashMap<>();
+
     public ProcessLifecycleProvider(Map<String, String> params)
     {
-        // Params unused for now
+        // Extract any JVM properties or environment variables from the params

Review Comment:
   can we annotate `params` in the constructor parameter with `@NotNull`?



##########
examples/lifecycle/setup.sh:
##########
@@ -0,0 +1,54 @@
+#!/bin/bash
+#
+# 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.
+#
+set -eu
+
+SCRIPT_DIR=$(realpath "$(dirname "$0")")
+CASSANDRA_VERSION=4.1.10
+CASSANDRA_DIR="apache-cassandra-${CASSANDRA_VERSION}"
+TARBALL_NAME="${CASSANDRA_DIR}-bin.tar.gz"
+TARBALL_URL="https://dlcdn.apache.org/cassandra/${CASSANDRA_VERSION}/${TARBALL_NAME}";
+NODE_DIR="${SCRIPT_DIR}/nodes/localhost"
+
+SIDECAR_YAML="${SCRIPT_DIR}/conf/sidecar.yaml"
+SIDECAR_YAML_TEMPLATE="${SCRIPT_DIR}/conf/sidecar.yaml.template"
+CASSANDRA_HOME="${NODE_DIR}/opt/${CASSANDRA_DIR}"
+CASSANDRA_LOG_DIR="${NODE_DIR}/var/log/cassandra"
+CASSANDRA_CONF="${NODE_DIR}/etc/cassandra"
+CASSANDRA_STORAGE_DIR="${NODE_DIR}/var/lib/cassandra"
+SIDECAR_LIFECYCLE_DIR="${NODE_DIR}/var/lib/cassandra-sidecar/lifecycle"
+TMP_DIR="${NODE_DIR}/tmp"
+
+echo "Creating directories"
+mkdir -p ${CASSANDRA_HOME} ${CASSANDRA_LOG_DIR} ${CASSANDRA_CONF} 
${CASSANDRA_STORAGE_DIR} ${SIDECAR_LIFECYCLE_DIR} ${TMP_DIR}
+
+if [ -f ${CASSANDRA_HOME}/bin/cassandra ]; then
+  echo "Cassandra already installed at ${CASSANDRA_HOME}, skipping install"
+else
+  echo "Installing Cassandra at ${CASSANDRA_HOME}"
+  echo "Downloading ${TARBALL_URL}"
+  wget -P ${TMP_DIR} ${TARBALL_URL}

Review Comment:
   can we use curl here instead? I think curl is more widely available than wget
   ```suggestion
     curl -L -o ${TMP_DIR}/$(basename ${TARBALL_URL}) ${TARBALL_URL}
   ```



##########
server/src/main/java/org/apache/cassandra/sidecar/lifecycle/ProcessLifecycleProvider.java:
##########
@@ -18,38 +18,232 @@
 
 package org.apache.cassandra.sidecar.lifecycle;
 
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.exceptions.ConfigurationException;
+import org.jetbrains.annotations.VisibleForTesting;
 
 /**
- * A {@link LifecycleProvider} that manages Cassandra instances as OS 
processes.
- * <p>
- * This implementation is a placeholder and is not yet implemented.
+ * Manage the lifecycle of Cassandra instances running on local processes
  */
 public class ProcessLifecycleProvider implements LifecycleProvider
 {
+    static final String OPT_CASSANDRA_HOME = "cassandra_home";
+    static final String OPT_CASSANDRA_CONF_DIR = "cassandra_conf_dir";
+    static final String OPT_CASSANDRA_LOG_DIR = "cassandra_log_dir";
+    static final String OPT_STATE_DIR = "state_dir";
+
+    protected static final Logger LOG = 
LoggerFactory.getLogger(ProcessLifecycleProvider.class);
+    public static final long CASSANDRA_PROCESS_TIMEOUT_MS = 
Long.getLong("cassandra.sidecar.lifecycle.process.timeout.ms", 120_000L);
+
+    private final String lifecycleDir;
+    private final String defaultCassandraHome;
+    private final Map<String, String> defaultJvmProperties = new HashMap<>();
+    private final Map<String, String> defaultEnvVars = new HashMap<>();
+
     public ProcessLifecycleProvider(Map<String, String> params)
     {
-        // Params unused for now
+        // Extract any JVM properties or environment variables from the params
+        for (Map.Entry<String, String> entry : params.entrySet())
+        {
+            if (entry.getKey().startsWith("cassandra."))
+            {
+                defaultJvmProperties.put(entry.getKey(), entry.getValue());
+            }
+            else if (entry.getKey().startsWith("env."))
+            {
+                defaultEnvVars.put(entry.getKey().replaceAll("env.", ""), 
entry.getValue());
+            }
+        }
+        this.lifecycleDir = params.get(OPT_STATE_DIR);
+        this.defaultCassandraHome = params.get(OPT_CASSANDRA_HOME);
+        if (lifecycleDir == null || lifecycleDir.isEmpty())
+        {
+            throw new ConfigurationException("Configuration property '" + 
OPT_STATE_DIR + "' must be set for ProcessLifecycleProvider");
+        }
+        if (defaultCassandraHome == null || defaultCassandraHome.isEmpty())
+        {
+            throw new ConfigurationException("Configuration property '" + 
OPT_CASSANDRA_HOME + "' must be set for ProcessLifecycleProvider");
+        }
     }
 
     @Override
     public void start(InstanceMetadata instance)
     {
-        throw new UnsupportedOperationException("Not implemented yet");
+        if (isCassandraProcessRunning(instance))
+        {
+            LOG.info("Cassandra instance {} is already running.", 
instance.host());
+            return;
+        }
+        startCassandra(instance);
     }
 
     @Override
     public void stop(InstanceMetadata instance)
     {
-        throw new UnsupportedOperationException("Not implemented yet");
-
+        if (!isCassandraProcessRunning(instance))
+        {
+            LOG.info("Cassandra instance {} is already stopped.", 
instance.host());
+            return;
+        }
+        stopCassandra(instance);
     }
 
     @Override
     public boolean isRunning(InstanceMetadata instance)
     {
-        throw new UnsupportedOperationException("Not implemented yet");
+        return isCassandraProcessRunning(instance);
+    }
+
+    private void startCassandra(InstanceMetadata instance)
+    {
+        ProcessRuntimeConfiguration runtimeConfig = 
getRuntimeConfiguration(instance);
+        try
+        {
+            String stdoutLocation = 
getStdoutLocation(runtimeConfig.instanceName());
+            String stderrLocation = 
getStderrLocation(runtimeConfig.instanceName());
+            String pidFileLocation = 
getPidFileLocation(runtimeConfig.instanceName());
+            ProcessBuilder processBuilder = 
runtimeConfig.buildStartCommand(pidFileLocation,
+                                                                            
stdoutLocation,
+                                                                            
stderrLocation);
+            LOG.info("Starting Cassandra instance {} with command: {}", 
runtimeConfig.instanceName(), processBuilder.command());
+
+            Process process = processBuilder.start();
+            process.waitFor(CASSANDRA_PROCESS_TIMEOUT_MS, 
TimeUnit.MILLISECONDS); // blocking call, make async?

Review Comment:
   Looking at where this code is running , it seems like we always are calling 
this in a blocking thread. So I think it's fine to keep the code as is, under 
the assumption that the entry point for the call of this code will be run in a 
blocking thread.



##########
server/src/main/java/org/apache/cassandra/sidecar/lifecycle/ProcessLifecycleProvider.java:
##########
@@ -18,38 +18,354 @@
 
 package org.apache.cassandra.sidecar.lifecycle;
 
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.exceptions.ConfigurationException;
+import org.jetbrains.annotations.VisibleForTesting;
 
 /**
- * A {@link LifecycleProvider} that manages Cassandra instances as OS 
processes.
- * <p>
- * This implementation is a placeholder and is not yet implemented.
+ * Manage the lifecycle of Cassandra instances running on local processes
  */
 public class ProcessLifecycleProvider implements LifecycleProvider
 {
+    static final String OPT_CASSANDRA_HOME = "cassandra_home";
+    static final String OPT_CASSANDRA_CONF_DIR = "cassandra_conf_dir";
+    static final String OPT_CASSANDRA_LOG_DIR = "cassandra_log_dir";
+    static final String OPT_STATE_DIR = "state_dir";
+
+    protected static final Logger LOG = 
LoggerFactory.getLogger(ProcessLifecycleProvider.class);
+    public static final long CASSANDRA_PROCESS_TIMEOUT_MS = 
Long.getLong("cassandra.sidecar.lifecycle.process.timeout.ms", 120_000L);
+
+    private final String lifecycleDir;
+    private final String defaultCassandraHome;
+    private final Map<String, String> defaultJvmProperties = new HashMap<>();
+    private final Map<String, String> defaultEnvVars = new HashMap<>();
+
     public ProcessLifecycleProvider(Map<String, String> params)
     {
-        // Params unused for now
+        // Extract any JVM properties or environment variables from the params
+        for (Map.Entry<String, String> entry : params.entrySet())
+        {
+            if (entry.getKey().startsWith("cassandra."))
+            {
+                defaultJvmProperties.put(entry.getKey(), entry.getValue());
+            }
+            else if (entry.getKey().startsWith("env."))
+            {
+                defaultEnvVars.put(entry.getKey().replaceAll("env.", ""), 
entry.getValue());
+            }
+        }
+        this.lifecycleDir = params.get(OPT_STATE_DIR);
+        this.defaultCassandraHome = params.get(OPT_CASSANDRA_HOME);
+        validateConfiguration();
+    }
+
+    private void validateConfiguration()
+    {
+        if (lifecycleDir == null || lifecycleDir.isEmpty())
+        {
+            throw new ConfigurationException("Configuration property '" + 
OPT_STATE_DIR + "' must be set for ProcessLifecycleProvider");
+        }
+        if (defaultCassandraHome == null || defaultCassandraHome.isEmpty())
+        {
+            throw new ConfigurationException("Configuration property '" + 
OPT_CASSANDRA_HOME + "' must be set for ProcessLifecycleProvider");
+        }
+
+        Path stateDir = Path.of(lifecycleDir);
+        if (!Files.isDirectory(stateDir))
+        {
+            throw new ConfigurationException("State directory '" + 
lifecycleDir + "' does not exist or is not a directory");
+        }
+        if (!Files.isWritable(stateDir))
+        {
+            throw new ConfigurationException("State directory '" + 
lifecycleDir + "' is not writable");
+        }
+
+        Path cassandraHomePath = Path.of(defaultCassandraHome);
+        if (!Files.isDirectory(cassandraHomePath))
+        {
+            throw new ConfigurationException("Cassandra home '" + 
defaultCassandraHome + "' does not exist or is not a directory");
+        }
+        if (!Files.isReadable(cassandraHomePath))
+        {
+            throw new ConfigurationException("Cassandra home '" + 
defaultCassandraHome + "' is not readable");
+        }
     }
 
     @Override
     public void start(InstanceMetadata instance)
     {
-        throw new UnsupportedOperationException("Not implemented yet");
+        if (isCassandraProcessRunning(instance))
+        {
+            LOG.info("Cassandra instance {} is already running.", 
instance.host());
+            return;
+        }
+        startCassandra(instance);
     }
 
     @Override
     public void stop(InstanceMetadata instance)
     {
-        throw new UnsupportedOperationException("Not implemented yet");
-
+        if (!isCassandraProcessRunning(instance))
+        {
+            LOG.info("Cassandra instance {} is already stopped.", 
instance.host());
+            return;
+        }
+        stopCassandra(instance);
     }
 
     @Override
     public boolean isRunning(InstanceMetadata instance)
     {
-        throw new UnsupportedOperationException("Not implemented yet");
+        return isCassandraProcessRunning(instance);
+    }
+
+    private void startCassandra(InstanceMetadata instance)
+    {
+        ProcessRuntimeConfiguration runtimeConfig = 
getRuntimeConfiguration(instance);
+        try
+        {
+            String stdoutLocation = 
getStdoutLocation(runtimeConfig.instanceName());
+            String stderrLocation = 
getStderrLocation(runtimeConfig.instanceName());
+            String pidFileLocation = 
getPidFileLocation(runtimeConfig.instanceName());
+            ProcessBuilder processBuilder = 
runtimeConfig.buildStartCommand(pidFileLocation,
+                                                                            
stdoutLocation,
+                                                                            
stderrLocation);
+            LOG.info("Starting Cassandra instance {} with command: {}", 
runtimeConfig.instanceName(), processBuilder.command());
+
+            Process process = processBuilder.start();
+            process.waitFor(CASSANDRA_PROCESS_TIMEOUT_MS, 
TimeUnit.MILLISECONDS); // blocking call, make async?
+
+            if (isCassandraProcessRunning(instance))
+            {
+                LOG.info("Started Cassandra instance {} with PID {}", 
runtimeConfig.instanceName(), readPidFromFile(Path.of(pidFileLocation)));
+            }
+            else
+            {
+                throw new RuntimeException("Failed to start Cassandra instance 
" + runtimeConfig.instanceName() +
+                                           ". Check stdout at " + 
stdoutLocation + " and stderr at " + stderrLocation);
+            }
+        }
+        catch (Throwable t)
+        {
+            throw new RuntimeException("Failed to start Cassandra instance " + 
runtimeConfig.instanceName() + " due to " + t.getMessage(), t);
+        }
+    }
+
+    private void stopCassandra(InstanceMetadata instance)
+    {
+        ProcessRuntimeConfiguration casCfg = getRuntimeConfiguration(instance);
+        try
+        {
+            String pidFileLocation = getPidFileLocation(casCfg.instanceName());
+            Long pid = readPidFromFile(Path.of(pidFileLocation));
+            Optional<ProcessHandle> processHandle = ProcessHandle.of(pid);
+            if (processHandle.isPresent())
+            {
+                LOG.info("Stopping process of Cassandra instance {} with PID 
{}.", casCfg.instanceName(), pid);
+                CompletableFuture<ProcessHandle> terminationFuture = 
processHandle.get().onExit();
+                processHandle.get().destroy();
+                try
+                {
+                    terminationFuture.get(CASSANDRA_PROCESS_TIMEOUT_MS, 
TimeUnit.MILLISECONDS);
+                }
+                catch (TimeoutException e)
+                {
+                    LOG.warn("Process {} did not terminate within timeout, 
forcing destroy.", pid);
+                    boolean destroyed = processHandle.get().destroyForcibly();
+                    if (!destroyed)
+                    {
+                        throw new RuntimeException("Failed to forcibly destroy 
process " + pid +
+                                                   " for Cassandra instance " 
+ casCfg.instanceName(), e);
+                    }
+                    LOG.info("Process {} was forcibly destroyed.", pid);
+                }
+                Files.deleteIfExists(Path.of(pidFileLocation));
+            }
+            else
+            {
+                LOG.warn("No process running for Cassandra instance {} with 
PID {}.", casCfg.instanceName(), pid);
+            }
+        }
+        catch (Throwable t)
+        {
+            throw new RuntimeException("Failed to stop process for Cassandra 
instance " + casCfg.instanceName() + " due to " + t.getMessage(), t);
+        }
+    }
+
+    @VisibleForTesting
+    protected ProcessRuntimeConfiguration 
getRuntimeConfiguration(InstanceMetadata instance)
+    {
+        String cassandraHome = 
Optional.ofNullable(instance.lifecycleOptions().get(OPT_CASSANDRA_HOME))
+                                       .orElse(defaultCassandraHome);
+        String cassandraConfDir = 
instance.lifecycleOptions().get(OPT_CASSANDRA_CONF_DIR);
+        String cassandraLogDir = 
instance.lifecycleOptions().get(OPT_CASSANDRA_LOG_DIR);
+        return new ProcessRuntimeConfiguration.Builder()
+                                        .withHost(instance.host())
+                                        .withCassandraHome(cassandraHome)
+                                        .withCassandraConfDir(cassandraConfDir)
+                                        .withCassandraLogDir(cassandraLogDir)
+                                        .withStorageDir(instance.storageDir())
+                                        .withJvmOptions(defaultJvmProperties)
+                                        .withEnvVars(defaultEnvVars)
+                                        .build();
+    }
+
+    /**
+     * Checks whether a Cassandra instance is currently running as a local 
process
+     * and automatically cleans up stale PID files.
+     *
+     * Performs four validation steps:
+     * 1. Verifies the PID file exists and is readable. Returns false if not 
found.
+     * 2. Reads the PID and checks if the process is alive. Returns false and 
deletes the
+     *    PID file if the process no longer exists or is not alive.
+     * 3. Verifies the process is a Cassandra instance by checking for
+     *    org.apache.cassandra.service.CassandraDaemon in the command line. 
Returns true if
+     *    the command line contains the Cassandra daemon class or cannot be 
determined.
+     * 4. If the process is running but is not a Cassandra process, returns 
false and deletes
+     *    the stale PID file.
+     *
+     * @param instance the instance metadata containing host information
+     * @return true if the instance is running as a Cassandra process, false 
otherwise
+     */
+    private boolean isCassandraProcessRunning(InstanceMetadata instance)
+    {
+        Path pidFilePath = Path.of(getPidFileLocation(instance.host()));
+        try
+        {
+            // Case 1: PID file does not exist or is not readable
+            if (!Files.isRegularFile(pidFilePath) || 
!Files.isReadable(pidFilePath))
+            {
+                LOG.debug("PID file does not exist or is not readable for 
instance {} at path {}", instance.host(), pidFilePath);

Review Comment:
   We should rely on the instance's toString() implementation for the log 
entries here:
   ```suggestion
                   LOG.debug("PID file does not exist or is not readable for 
instance {} at path {}", instance, pidFilePath);
   ```



##########
server/src/main/java/org/apache/cassandra/sidecar/lifecycle/ProcessLifecycleProvider.java:
##########
@@ -18,38 +18,354 @@
 
 package org.apache.cassandra.sidecar.lifecycle;
 
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.exceptions.ConfigurationException;
+import org.jetbrains.annotations.VisibleForTesting;
 
 /**
- * A {@link LifecycleProvider} that manages Cassandra instances as OS 
processes.
- * <p>
- * This implementation is a placeholder and is not yet implemented.
+ * Manage the lifecycle of Cassandra instances running on local processes
  */
 public class ProcessLifecycleProvider implements LifecycleProvider
 {
+    static final String OPT_CASSANDRA_HOME = "cassandra_home";
+    static final String OPT_CASSANDRA_CONF_DIR = "cassandra_conf_dir";
+    static final String OPT_CASSANDRA_LOG_DIR = "cassandra_log_dir";
+    static final String OPT_STATE_DIR = "state_dir";
+
+    protected static final Logger LOG = 
LoggerFactory.getLogger(ProcessLifecycleProvider.class);
+    public static final long CASSANDRA_PROCESS_TIMEOUT_MS = 
Long.getLong("cassandra.sidecar.lifecycle.process.timeout.ms", 120_000L);
+
+    private final String lifecycleDir;
+    private final String defaultCassandraHome;
+    private final Map<String, String> defaultJvmProperties = new HashMap<>();
+    private final Map<String, String> defaultEnvVars = new HashMap<>();
+
     public ProcessLifecycleProvider(Map<String, String> params)
     {
-        // Params unused for now
+        // Extract any JVM properties or environment variables from the params
+        for (Map.Entry<String, String> entry : params.entrySet())
+        {
+            if (entry.getKey().startsWith("cassandra."))
+            {
+                defaultJvmProperties.put(entry.getKey(), entry.getValue());
+            }
+            else if (entry.getKey().startsWith("env."))
+            {
+                defaultEnvVars.put(entry.getKey().replaceAll("env.", ""), 
entry.getValue());
+            }
+        }
+        this.lifecycleDir = params.get(OPT_STATE_DIR);
+        this.defaultCassandraHome = params.get(OPT_CASSANDRA_HOME);
+        validateConfiguration();
+    }
+
+    private void validateConfiguration()
+    {
+        if (lifecycleDir == null || lifecycleDir.isEmpty())
+        {
+            throw new ConfigurationException("Configuration property '" + 
OPT_STATE_DIR + "' must be set for ProcessLifecycleProvider");
+        }
+        if (defaultCassandraHome == null || defaultCassandraHome.isEmpty())
+        {
+            throw new ConfigurationException("Configuration property '" + 
OPT_CASSANDRA_HOME + "' must be set for ProcessLifecycleProvider");
+        }
+
+        Path stateDir = Path.of(lifecycleDir);
+        if (!Files.isDirectory(stateDir))
+        {
+            throw new ConfigurationException("State directory '" + 
lifecycleDir + "' does not exist or is not a directory");
+        }
+        if (!Files.isWritable(stateDir))
+        {
+            throw new ConfigurationException("State directory '" + 
lifecycleDir + "' is not writable");
+        }
+
+        Path cassandraHomePath = Path.of(defaultCassandraHome);
+        if (!Files.isDirectory(cassandraHomePath))
+        {
+            throw new ConfigurationException("Cassandra home '" + 
defaultCassandraHome + "' does not exist or is not a directory");
+        }
+        if (!Files.isReadable(cassandraHomePath))
+        {
+            throw new ConfigurationException("Cassandra home '" + 
defaultCassandraHome + "' is not readable");
+        }
     }
 
     @Override
     public void start(InstanceMetadata instance)
     {
-        throw new UnsupportedOperationException("Not implemented yet");
+        if (isCassandraProcessRunning(instance))
+        {
+            LOG.info("Cassandra instance {} is already running.", 
instance.host());
+            return;
+        }
+        startCassandra(instance);
     }
 
     @Override
     public void stop(InstanceMetadata instance)
     {
-        throw new UnsupportedOperationException("Not implemented yet");
-
+        if (!isCassandraProcessRunning(instance))
+        {
+            LOG.info("Cassandra instance {} is already stopped.", 
instance.host());
+            return;
+        }
+        stopCassandra(instance);
     }
 
     @Override
     public boolean isRunning(InstanceMetadata instance)
     {
-        throw new UnsupportedOperationException("Not implemented yet");
+        return isCassandraProcessRunning(instance);
+    }
+
+    private void startCassandra(InstanceMetadata instance)
+    {
+        ProcessRuntimeConfiguration runtimeConfig = 
getRuntimeConfiguration(instance);
+        try
+        {
+            String stdoutLocation = 
getStdoutLocation(runtimeConfig.instanceName());
+            String stderrLocation = 
getStderrLocation(runtimeConfig.instanceName());
+            String pidFileLocation = 
getPidFileLocation(runtimeConfig.instanceName());
+            ProcessBuilder processBuilder = 
runtimeConfig.buildStartCommand(pidFileLocation,
+                                                                            
stdoutLocation,
+                                                                            
stderrLocation);
+            LOG.info("Starting Cassandra instance {} with command: {}", 
runtimeConfig.instanceName(), processBuilder.command());
+
+            Process process = processBuilder.start();
+            process.waitFor(CASSANDRA_PROCESS_TIMEOUT_MS, 
TimeUnit.MILLISECONDS); // blocking call, make async?
+
+            if (isCassandraProcessRunning(instance))
+            {
+                LOG.info("Started Cassandra instance {} with PID {}", 
runtimeConfig.instanceName(), readPidFromFile(Path.of(pidFileLocation)));
+            }
+            else
+            {
+                throw new RuntimeException("Failed to start Cassandra instance 
" + runtimeConfig.instanceName() +
+                                           ". Check stdout at " + 
stdoutLocation + " and stderr at " + stderrLocation);
+            }
+        }
+        catch (Throwable t)
+        {
+            throw new RuntimeException("Failed to start Cassandra instance " + 
runtimeConfig.instanceName() + " due to " + t.getMessage(), t);
+        }
+    }
+
+    private void stopCassandra(InstanceMetadata instance)
+    {
+        ProcessRuntimeConfiguration casCfg = getRuntimeConfiguration(instance);
+        try
+        {
+            String pidFileLocation = getPidFileLocation(casCfg.instanceName());
+            Long pid = readPidFromFile(Path.of(pidFileLocation));
+            Optional<ProcessHandle> processHandle = ProcessHandle.of(pid);
+            if (processHandle.isPresent())
+            {
+                LOG.info("Stopping process of Cassandra instance {} with PID 
{}.", casCfg.instanceName(), pid);
+                CompletableFuture<ProcessHandle> terminationFuture = 
processHandle.get().onExit();
+                processHandle.get().destroy();
+                try
+                {
+                    terminationFuture.get(CASSANDRA_PROCESS_TIMEOUT_MS, 
TimeUnit.MILLISECONDS);
+                }
+                catch (TimeoutException e)
+                {
+                    LOG.warn("Process {} did not terminate within timeout, 
forcing destroy.", pid);
+                    boolean destroyed = processHandle.get().destroyForcibly();
+                    if (!destroyed)
+                    {
+                        throw new RuntimeException("Failed to forcibly destroy 
process " + pid +
+                                                   " for Cassandra instance " 
+ casCfg.instanceName(), e);
+                    }
+                    LOG.info("Process {} was forcibly destroyed.", pid);
+                }
+                Files.deleteIfExists(Path.of(pidFileLocation));
+            }
+            else
+            {
+                LOG.warn("No process running for Cassandra instance {} with 
PID {}.", casCfg.instanceName(), pid);
+            }
+        }
+        catch (Throwable t)
+        {
+            throw new RuntimeException("Failed to stop process for Cassandra 
instance " + casCfg.instanceName() + " due to " + t.getMessage(), t);
+        }
+    }
+
+    @VisibleForTesting
+    protected ProcessRuntimeConfiguration 
getRuntimeConfiguration(InstanceMetadata instance)
+    {
+        String cassandraHome = 
Optional.ofNullable(instance.lifecycleOptions().get(OPT_CASSANDRA_HOME))
+                                       .orElse(defaultCassandraHome);

Review Comment:
   NIT, simplify:
   ```suggestion
           String cassandraHome = 
instance.lifecycleOptions().getOrDefault(OPT_CASSANDRA_HOME, 
defaultCassandraHome);
   ```



##########
server/src/main/java/org/apache/cassandra/sidecar/config/InstanceConfiguration.java:
##########
@@ -104,4 +105,9 @@ public interface InstanceConfiguration
      * @return the password for the JMX role for the JMX service for the 
Cassandra instance
      */
     String jmxRolePassword();
+
+    /**
+     * @return The lifecycle options for this Cassandra instance
+     */
+    Map<String, String> lifecycleOptions();

Review Comment:
   any reason for having this as a generic map, instead of using strongly typed 
convention?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: pr-unsubscr...@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscr...@cassandra.apache.org
For additional commands, e-mail: pr-h...@cassandra.apache.org