diff --git a/README.md b/README.md
index 79b7827f..f4ee0865 100644
--- a/README.md
+++ b/README.md
@@ -148,7 +148,20 @@ For non-Java applications or tools like `cqlsh`, you can run the Spanner Cassand
This will create an executable jar file `spanner-cassandra-launcher.jar` inside the folder `google-cloud-spanner-cassandra/target`.
-* Run the jar using the command:
+* Run the jar using one of the following methods:
+
+ **1. Using a YAML Configuration File (Recommended for Production)**
+
+ For production setups, it is recommended to use a YAML file to configure the adapter. This method supports multiple listeners and global settings. See the [Configuration Options](docs/config-options.md) for a complete list of all supported options. An example `config.yaml` file can be found [here](docs/config-options.md#example-configyaml).
+
+ Then, run the launcher with the `-DconfigFilePath` system property:
+ ```bash
+ java -DconfigFilePath=/path/to/config.yaml -jar path/to/your/spanner-cassandra-launcher.jar
+ ```
+
+ **2. Using System Properties (for a single listener)**
+
+ For simpler setups or quick testing, you can provide the configuration via system properties. This method only supports a single adapter listener.
```bash
java -DdatabaseUri=projects/my-project/instances/my-instance/databases/my-database \
@@ -159,9 +172,16 @@ For non-Java applications or tools like `cqlsh`, you can run the Spanner Cassand
-jar path/to/your/spanner-cassandra-launcher.jar
```
- * Replace the value of `-DdatabaseUri` with your Spanner database URI.
- * You can omit `-Dhost` to use the default `0.0.0.0`, omit `-Dport` to use the default `9042`, and omit `-DnumGrpcChannels` to use the default `4`.
- * `-DhealthCheckPort` is optional. If specified, a health check endpoint will be started on same IP address as that of the client on the specified port at url `/debug/health`. The health check endpoint will return HTTP status `200: OK` if the client is up and running, and `503: Service Unavailable` otherwise. The health check endpoint is NOT enabled by default.
+ **Configuration Notes:**
+
+ * **Database URI**: You must specify the Spanner database URI. This is done via the `databaseUri` property in YAML or the `-DdatabaseUri` system property.
+ * **Host**: The default host is `0.0.0.0`. This can be overridden with the `host` property in YAML or `-Dhost`.
+ * **Port**: The default port is `9042`. This can be overridden with the `port` property in YAML or `-Dport`.
+ * **gRPC Channels**: The default number of gRPC channels is `4`. This can be overridden with `numGrpcChannels` in YAML or `-DnumGrpcChannels`.
+ * **Health Check**: You can optionally enable a health check endpoint.
+ * In YAML, set `healthCheckEndpoint` to a `host:port` value (e.g., "127.0.0.1:8080").
+ * With system properties, use `-DhealthCheckPort` and specify a port. The host will default to the adapter's host.
+ * When enabled, the endpoint is available at `/debug/health` and returns HTTP `200 OK` if the client is running, or `503 Service Unavailable` otherwise. The health check is disabled by default.
## View and manage client-side metrics
diff --git a/config.yaml b/config.yaml
new file mode 100644
index 00000000..c0f7e4ca
--- /dev/null
+++ b/config.yaml
@@ -0,0 +1,22 @@
+# Global client configuration
+globalClientConfigs:
+ enableBuiltInMetrics: true
+ healthCheckEndpoint: "127.0.0.1:8080"
+
+# List of all listeners
+listeners:
+ - # Configuration for listener_1
+ name: "listener_1"
+ host: "127.0.0.1"
+ port: 9042
+ spanner:
+ databaseUri: "projects/span-cloud-testing/instances/pecheverri-cassandra/databases/default"
+ numGrpcChannels: 4
+ maxCommitDelayMillis: 5
+ - # Configuration for listener_2
+ name: "listener_2"
+ host: "127.0.0.2"
+ port: 9043
+ spanner:
+ databaseUri: "projects/span-cloud-testing/instances/pecheverri-cassandra/databases/default"
+ numGrpcChannels: 8
diff --git a/docs/config-options.md b/docs/config-options.md
new file mode 100644
index 00000000..18138543
--- /dev/null
+++ b/docs/config-options.md
@@ -0,0 +1,59 @@
+# Config Options
+
+This file documents all the configuration options supported by the Spanner Cassandra Adapter.
+
+```yaml
+# [Optional] Global client configurations that apply to all listeners.
+globalClientConfigs:
+ # [Optional] Enables built-in metrics. Defaults to false. It is highly recommended to enable metrics in production environments for improved debuggability.
+ enableBuiltInMetrics: true
+ # [Optional] The endpoint for the health check server. If not specified, the health check server will not be started.
+ # To check the status, send a GET request to the '/debug/health' path on this endpoint.
+ # - A '200 OK' status indicates that the service is healthy.
+ # - A '503 Service Unavailable' status indicates that one or more listeners failed to start.
+ healthCheckEndpoint: "127.0.0.1:8080"
+
+# A list of all listeners to start.
+listeners:
+ # The name of the listener. It is recommended to use a meaningful name, such as the cluster name.
+ - name: "listener_1"
+ # [Optional] The host to bind the listener to. Defaults to "0.0.0.0".
+ host: "127.0.0.1"
+ # [Optional] The port to bind the listener to. Defaults to 9042.
+ port: 9042
+ # Spanner configuration for this listener.
+ spanner:
+ # The URI of the Spanner database.
+ databaseUri: "projects/my-project/instances/my-instance/databases/my-database"
+ # [Optional] The number of gRPC channels to use. Defaults to 4.
+ numGrpcChannels: 4
+ # [Optional] The maximum commit delay in milliseconds. Defaults to 0ms.
+ # This is the amount of latency this request is willing to incur in order
+ # to improve throughput. If this field is not set, Spanner assumes requests
+ # are relatively latency sensitive and automatically determines an appropriate
+ # delay time.
+ maxCommitDelayMillis: 5
+```
+
+# Example config.yaml
+
+```yaml
+globalClientConfigs:
+ enableBuiltInMetrics: true
+ healthCheckEndpoint: "127.0.0.1:8080"
+
+listeners:
+ - name: "listener_1"
+ host: "127.0.0.1"
+ port: 9042
+ spanner:
+ databaseUri: "projects/my-project/instances/my-instance/databases/my-database"
+ numGrpcChannels: 4
+ maxCommitDelayMillis: 5
+ - name: "listener_2"
+ host: "127.0.0.2"
+ port: 9043
+ spanner:
+ databaseUri: "projects/my-project/instances/my-instance/databases/my-database-2"
+ numGrpcChannels: 8
+```
diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Launcher.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Launcher.java
index 9fd88192..c4848682 100644
--- a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Launcher.java
+++ b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Launcher.java
@@ -13,6 +13,7 @@
See the License for the specific language governing permissions and
limitations under the License.
*/
+
package com.google.cloud.spanner.adapter;
import com.google.cloud.spanner.adapter.metrics.BuiltInMetricsProvider;
@@ -22,18 +23,41 @@
import java.io.IOException;
import java.net.InetAddress;
import java.time.Duration;
-import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Main entry point for running a Spanner Cassandra Adapter as a stand-alone application.
*
- * This class reads configuration parameters from system properties, initializes the underlying
- * {@link Adapter}, registers a shutdown hook for graceful termination, and starts the adapter
- * service.
+ *
The adapter can be configured using a YAML file, specified by the {@code
+ * -DconfigFilePath=/path/to/config.yaml} system property. This is the recommended approach for
+ * production and complex setups, as it supports multiple listeners and global settings.
+ *
+ *
For simpler setups or quick testing, configuration can be provided via system properties. This
+ * method only supports a single adapter listener.
+ *
+ *
YAML Configuration Structure:
+ *
+ *
+ * globalClientConfigs:
+ * enableBuiltInMetrics: true
+ * healthCheckEndpoint: "127.0.0.1:8080"
+ * listeners:
+ * - name: "listener_1"
+ * host: "127.0.0.1"
+ * port: 9042
+ * spanner:
+ * databaseUri: "projects/my-project/instances/my-instance/databases/my-database"
+ * - name: "listener_2"
+ * ...
+ *
*
- * Configuration is provided via the following system properties:
+ *
System Property Configuration (for a single listener):
*
*
* - {@code databaseUri}: (Required) The URI of the target Spanner database.
@@ -48,7 +72,15 @@
* unspecifed, health check server will NOT be started.
*
*
- * Example usage:
+ * Example Usage:
+ *
+ *
Using a YAML configuration file:
+ *
+ *
+ * java -DconfigFilePath=/path/to/config.yaml -jar path/to/your/spanner-cassandra-launcher.jar
+ *
+ *
+ * Using system properties for a single adapter:
*
*
* java -DdatabaseUri=projects/my-project/instances/my-instance/databases/my-database \
@@ -57,7 +89,7 @@
* -DnumGrpcChannels=4 \
* -DmaxCommitDelayMillis=5 \
* -DhealthCheckPort=8080 \
- * -jar com.google.cloud.spanner.adapter.SpannerCassandraLauncher
+ * -jar path/to/your/spanner-cassandra-launcher.jar
*
*
* @see Adapter
@@ -66,120 +98,168 @@ public class Launcher {
private static final Logger LOG = LoggerFactory.getLogger(Launcher.class);
private static final BuiltInMetricsProvider builtInMetricsProvider =
BuiltInMetricsProvider.INSTANCE;
- private static final String DEFAULT_SPANNER_ENDPOINT = "spanner.googleapis.com:443";
- private static final String DATABASE_URI_PROP_KEY = "databaseUri";
- private static final String HOST_PROP_KEY = "host";
- private static final String PORT_PROP_KEY = "port";
- private static final String NUM_GRPC_CHANNELS_PROP_KEY = "numGrpcChannels";
- private static final String DEFAULT_HOST = "0.0.0.0";
- private static final String DEFAULT_PORT = "9042";
- private static final String DEFAULT_NUM_GRPC_CHANNELS = "4";
- private static final String MAX_COMMIT_DELAY_PROP_KEY = "maxCommitDelayMillis";
- private static final String ENABLE_BUILTIN_METRICS_PROP_KEY = "enableBuiltInMetrics";
- private static final String HEALTH_CHECK_PORT_PROP_KEY = "healthCheckPort";
-
- private final Adapter adapter;
- private final HealthCheckServer healthCheckServer;
-
- Launcher(Adapter adapter, @Nullable HealthCheckServer healthCheckServer) {
- this.adapter = adapter;
- this.healthCheckServer = healthCheckServer;
+ private final AdapterFactory adapterFactory;
+ private final List adapters = new ArrayList<>();
+ private HealthCheckServer healthCheckServer;
+
+ /**
+ * Factory for creating Adapter and HealthCheckServer instances. This class allows for mocking
+ * these dependencies in tests.
+ */
+ public static class AdapterFactory {
+ public Adapter createAdapter(AdapterOptions options) {
+ return new Adapter(options);
+ }
+
+ public HealthCheckServer createHealthCheckServer(InetAddress hostAddress, int port)
+ throws IOException {
+ return new HealthCheckServer(hostAddress, port);
+ }
}
- void launch() {
+ public Launcher() {
+ this(new AdapterFactory());
+ }
+
+ public Launcher(AdapterFactory adapterFactory) {
+ this.adapterFactory = adapterFactory;
+ }
+
+ public static void main(String[] args) throws Exception {
+ Launcher launcher = new Launcher();
+
+ Map propertiesMap =
+ System.getProperties().stringPropertyNames().stream()
+ .collect(Collectors.toMap(Function.identity(), System.getProperties()::getProperty));
+ final LauncherConfig config = LauncherConfigParser.parse(propertiesMap);
+ launcher.run(config);
+
+ // Keep the main thread alive until all adapters are shut down.
+ try {
+ Thread.currentThread().join();
+ } catch (InterruptedException e) {
+ LOG.info("Main thread interrupted, shutting down.");
+ launcher.shutdown();
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ /**
+ * Starts all configured listeners and the health check server, and registers a shutdown hook for
+ * graceful termination.
+ *
+ * @param config The configuration for the launcher.
+ * @throws IOException if there is an error starting the network servers.
+ * @throws IllegalStateException if one or more adapters fail to start.
+ */
+ public void run(LauncherConfig config) throws Exception {
+ if (config.getHealthCheckConfig() != null) {
+ startHealthCheckServer(config.getHealthCheckConfig());
+ } else {
+ LOG.info("Health check server is disabled.");
+ }
+
+ final List failedListeners = new ArrayList<>();
+ for (ListenerConfig listenerConfig : config.getListeners()) {
+ try {
+ startAdapter(listenerConfig);
+ } catch (Exception e) {
+ String error = String.format("listener on port %d", listenerConfig.getPort());
+ LOG.error("Failed to start adapter for {}: {}", error, e.getMessage());
+ failedListeners.add(error);
+ }
+ }
+
+ final boolean allAdaptersStarted = failedListeners.isEmpty();
if (healthCheckServer != null) {
- healthCheckServer.start();
+ healthCheckServer.setReady(allAdaptersStarted);
+ }
+
+ if (!allAdaptersStarted) {
+ shutdown();
+ throw new IllegalStateException("One or more adapters failed to start: " + failedListeners);
}
- adapter.start();
+ // Register the single shutdown hook after all adapters are configured and started.
Runtime.getRuntime()
.addShutdownHook(
new Thread(
() -> {
- if (healthCheckServer != null) {
- healthCheckServer.stop();
- }
- try {
- adapter.stop();
- } catch (IOException e) {
- LOG.warn("Error while stopping Adapter: " + e.getMessage());
- }
+ LOG.info("Shutdown hook triggered. Stopping all adapters.");
+ shutdown();
}));
+ }
+ /**
+ * Stops all running adapters and the health check server. This method is automatically called by
+ * a shutdown hook when the JVM terminates, but can also be called programmatically for a graceful
+ * shutdown.
+ */
+ public void shutdown() {
if (healthCheckServer != null) {
- healthCheckServer.setReady(true);
+ healthCheckServer.stop();
}
+ adapters.forEach(
+ adapter -> {
+ try {
+ adapter.stop();
+ } catch (IOException e) {
+ LOG.warn("Error while stopping Adapter: " + e.getMessage());
+ }
+ });
}
- public static void main(String[] args) throws Exception {
- final String databaseUri = System.getProperty(DATABASE_URI_PROP_KEY);
- final InetAddress inetAddress =
- InetAddress.getByName(System.getProperty(HOST_PROP_KEY, DEFAULT_HOST));
- final int port = Integer.parseInt(System.getProperty(PORT_PROP_KEY, DEFAULT_PORT));
- final int numGrpcChannels =
- Integer.parseInt(System.getProperty(NUM_GRPC_CHANNELS_PROP_KEY, DEFAULT_NUM_GRPC_CHANNELS));
- final String maxCommitDelayProperty = System.getProperty(MAX_COMMIT_DELAY_PROP_KEY);
- final boolean enableBuiltInMetrics =
- Boolean.parseBoolean(System.getProperty(ENABLE_BUILTIN_METRICS_PROP_KEY, "false"));
- final String healthCheckPortStr = System.getProperty(HEALTH_CHECK_PORT_PROP_KEY);
- HealthCheckServer healthCheckServer = null;
-
- if (databaseUri == null) {
- throw new IllegalArgumentException(
- "Spanner database URI not set. Please set it using -DdatabaseUri option.");
- }
+ private void startHealthCheckServer(HealthCheckConfig config) throws IOException {
+ healthCheckServer =
+ adapterFactory.createHealthCheckServer(config.getHostAddress(), config.getPort());
+ healthCheckServer.start();
+ }
- if (healthCheckPortStr != null) {
- final int healthCheckPort = Integer.parseInt(healthCheckPortStr);
- if (healthCheckPort < 0 || healthCheckPort > 65535) {
- throw new IllegalArgumentException(
- "Invalid health check port '" + healthCheckPort + "'. Must be between 0 and 65535");
- }
- healthCheckServer = new HealthCheckServer(inetAddress, healthCheckPort);
- } else {
- LOG.debug("Health check server is disabled.");
+ private AdapterOptions buildAdapterOptions(
+ ListenerConfig config, BuiltInMetricsRecorder metricsRecorder) {
+ final AdapterOptions.Builder opBuilder =
+ new AdapterOptions.Builder()
+ .spannerEndpoint(config.getSpannerEndpoint())
+ .tcpPort(config.getPort())
+ .databaseUri(config.getDatabaseUri())
+ .inetAddress(config.getHostAddress())
+ .numGrpcChannels(config.getNumGrpcChannels())
+ .metricsRecorder(metricsRecorder);
+ if (config.getMaxCommitDelayMillis() != null) {
+ opBuilder.maxCommitDelay(Duration.ofMillis(config.getMaxCommitDelayMillis()));
}
+ return opBuilder.build();
+ }
- DatabaseName databaseName = DatabaseName.parse(databaseUri);
- OpenTelemetry openTelemetry =
+ private BuiltInMetricsRecorder createMetricsRecorder(
+ boolean enableBuiltInMetrics, DatabaseName databaseName) {
+ final OpenTelemetry openTelemetry =
enableBuiltInMetrics
? builtInMetricsProvider.getOrCreateOpenTelemetry(
databaseName.getProject(), databaseName.getInstance())
: OpenTelemetry.noop();
- BuiltInMetricsRecorder metricsRecorder =
- new BuiltInMetricsRecorder(
- openTelemetry,
- builtInMetricsProvider.createDefaultAttributes(databaseName.getDatabase()));
-
- AdapterOptions.Builder opBuilder =
- new AdapterOptions.Builder()
- .spannerEndpoint(DEFAULT_SPANNER_ENDPOINT)
- .tcpPort(port)
- .databaseUri(databaseUri)
- .inetAddress(inetAddress)
- .numGrpcChannels(numGrpcChannels)
- .metricsRecorder(metricsRecorder);
- if (maxCommitDelayProperty != null) {
- opBuilder.maxCommitDelay(Duration.ofMillis(Integer.parseInt(maxCommitDelayProperty)));
- }
+ return new BuiltInMetricsRecorder(
+ openTelemetry, builtInMetricsProvider.createDefaultAttributes(databaseName.getDatabase()));
+ }
- Adapter adapter = new Adapter(opBuilder.build());
+ private void startAdapter(ListenerConfig config) throws IOException {
LOG.info(
"Starting Adapter for Spanner database {} on {}:{} with {} gRPC channels, max commit"
+ " delay of {} and built-in metrics enabled: {}",
- databaseUri,
- inetAddress,
- port,
- numGrpcChannels,
- maxCommitDelayProperty,
- enableBuiltInMetrics);
- Launcher launcher = new Launcher(adapter, healthCheckServer);
- launcher.launch();
+ config.getDatabaseUri(),
+ config.getHostAddress(),
+ config.getPort(),
+ config.getNumGrpcChannels(),
+ config.getMaxCommitDelayMillis(),
+ config.isEnableBuiltInMetrics());
- try {
- Thread.currentThread().join();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
+ final DatabaseName databaseName = DatabaseName.parse(config.getDatabaseUri());
+ final BuiltInMetricsRecorder metricsRecorder =
+ createMetricsRecorder(config.isEnableBuiltInMetrics(), databaseName);
+ final AdapterOptions options = buildAdapterOptions(config, metricsRecorder);
+
+ final Adapter adapter = adapterFactory.createAdapter(options);
+ adapters.add(adapter);
+ adapter.start();
}
}
diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/LauncherConfig.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/LauncherConfig.java
new file mode 100644
index 00000000..bdb2f983
--- /dev/null
+++ b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/LauncherConfig.java
@@ -0,0 +1,351 @@
+/*
+Copyright 2025 Google LLC
+
+Licensed 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 com.google.cloud.spanner.adapter;
+
+import com.google.cloud.spanner.adapter.configs.ConfigConstants;
+import com.google.cloud.spanner.adapter.configs.ListenerConfigs;
+import com.google.cloud.spanner.adapter.configs.UserConfigs;
+import com.google.common.base.Strings;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
+
+/** Encapsulates the full configuration for the Launcher. */
+public final class LauncherConfig {
+ private final List listeners;
+ @Nullable private final HealthCheckConfig healthCheckConfig;
+
+ private LauncherConfig(
+ List listeners, @Nullable HealthCheckConfig healthCheckConfig) {
+ this.listeners = listeners;
+ this.healthCheckConfig = healthCheckConfig;
+ }
+
+ public List getListeners() {
+ return listeners;
+ }
+
+ @Nullable
+ public HealthCheckConfig getHealthCheckConfig() {
+ return healthCheckConfig;
+ }
+
+ static LauncherConfig fromUserConfigs(UserConfigs userConfigs) throws UnknownHostException {
+ if (userConfigs == null) {
+ throw new IllegalArgumentException("UserConfigs cannot be null.");
+ }
+ if (userConfigs.getListeners() == null || userConfigs.getListeners().isEmpty()) {
+ throw new IllegalArgumentException("No listeners defined in the configuration.");
+ }
+
+ final String globalSpannerEndpoint;
+ final boolean globalEnableBuiltInMetrics;
+ HealthCheckConfig healthCheckConfig = null;
+
+ if (userConfigs.getGlobalClientConfigs() != null) {
+ globalSpannerEndpoint =
+ userConfigs.getGlobalClientConfigs().getSpannerEndpoint() != null
+ ? userConfigs.getGlobalClientConfigs().getSpannerEndpoint()
+ : ConfigConstants.DEFAULT_SPANNER_ENDPOINT;
+ globalEnableBuiltInMetrics =
+ userConfigs.getGlobalClientConfigs().getEnableBuiltInMetrics() != null
+ && userConfigs.getGlobalClientConfigs().getEnableBuiltInMetrics();
+ if (userConfigs.getGlobalClientConfigs().getHealthCheckEndpoint() != null) {
+ healthCheckConfig =
+ HealthCheckConfig.fromEndpointString(
+ userConfigs.getGlobalClientConfigs().getHealthCheckEndpoint());
+ }
+ } else {
+ globalSpannerEndpoint = ConfigConstants.DEFAULT_SPANNER_ENDPOINT;
+ globalEnableBuiltInMetrics = false;
+ }
+
+ List listenerConfigs = new ArrayList<>();
+ for (ListenerConfigs listener : userConfigs.getListeners()) {
+ validateListenerConfig(listener);
+ listenerConfigs.add(
+ ListenerConfig.fromListenerConfigs(
+ listener, globalSpannerEndpoint, globalEnableBuiltInMetrics));
+ }
+
+ return new LauncherConfig(listenerConfigs, healthCheckConfig);
+ }
+
+ static LauncherConfig fromProperties(Map properties) throws UnknownHostException {
+ String databaseUri = properties.get(ConfigConstants.DATABASE_URI_PROP_KEY);
+ if (databaseUri == null) {
+ throw new IllegalArgumentException(
+ "Spanner database URI not set. Please set it using the '"
+ + ConfigConstants.DATABASE_URI_PROP_KEY
+ + "' property.");
+ }
+
+ ListenerConfig listenerConfig = ListenerConfig.fromProperties(properties);
+ HealthCheckConfig healthCheckConfig = HealthCheckConfig.fromProperties(properties);
+
+ return new LauncherConfig(Collections.singletonList(listenerConfig), healthCheckConfig);
+ }
+
+ private static void validateListenerConfig(ListenerConfigs listener) {
+ if (listener.getSpanner() == null
+ || Strings.isNullOrEmpty(listener.getSpanner().getDatabaseUri())) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Listener '%s' on port %s must have a non-empty 'spanner.databaseUri' defined.",
+ listener.getName(), listener.getPort()));
+ }
+ }
+}
+
+/** Encapsulates the configuration for a single Adapter listener. */
+final class ListenerConfig {
+ private final String databaseUri;
+ private final InetAddress hostAddress;
+ private final int port;
+ private final String spannerEndpoint;
+ private final int numGrpcChannels;
+ @Nullable private final Integer maxCommitDelayMillis;
+ private final boolean enableBuiltInMetrics;
+
+ private ListenerConfig(Builder builder) {
+ this.databaseUri = builder.databaseUri;
+ this.hostAddress = builder.hostAddress;
+ this.port = builder.port;
+ this.spannerEndpoint = builder.spannerEndpoint;
+ this.numGrpcChannels = builder.numGrpcChannels;
+ this.maxCommitDelayMillis = builder.maxCommitDelayMillis;
+ this.enableBuiltInMetrics = builder.enableBuiltInMetrics;
+ }
+
+ public String getDatabaseUri() {
+ return databaseUri;
+ }
+
+ public InetAddress getHostAddress() {
+ return hostAddress;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ public String getSpannerEndpoint() {
+ return spannerEndpoint;
+ }
+
+ public int getNumGrpcChannels() {
+ return numGrpcChannels;
+ }
+
+ @Nullable
+ public Integer getMaxCommitDelayMillis() {
+ return maxCommitDelayMillis;
+ }
+
+ public boolean isEnableBuiltInMetrics() {
+ return enableBuiltInMetrics;
+ }
+
+ static ListenerConfig fromListenerConfigs(
+ ListenerConfigs listener, String globalSpannerEndpoint, boolean globalEnableBuiltInMetrics)
+ throws UnknownHostException {
+ String host = listener.getHost() != null ? listener.getHost() : ConfigConstants.DEFAULT_HOST;
+ int port = listener.getPort() != null ? listener.getPort() : ConfigConstants.DEFAULT_PORT;
+ int numGrpcChannels =
+ listener.getSpanner().getNumGrpcChannels() != null
+ ? listener.getSpanner().getNumGrpcChannels()
+ : ConfigConstants.DEFAULT_NUM_GRPC_CHANNELS;
+ Integer maxCommitDelayMillis = listener.getSpanner().getMaxCommitDelayMillis();
+
+ return newBuilder()
+ .databaseUri(listener.getSpanner().getDatabaseUri())
+ .hostAddress(InetAddress.getByName(host))
+ .port(port)
+ .spannerEndpoint(globalSpannerEndpoint)
+ .numGrpcChannels(numGrpcChannels)
+ .maxCommitDelayMillis(maxCommitDelayMillis)
+ .enableBuiltInMetrics(globalEnableBuiltInMetrics)
+ .build();
+ }
+
+ static ListenerConfig fromProperties(Map properties) throws UnknownHostException {
+ String host =
+ properties.getOrDefault(ConfigConstants.HOST_PROP_KEY, ConfigConstants.DEFAULT_HOST);
+ int port =
+ Integer.parseInt(
+ properties.getOrDefault(
+ ConfigConstants.PORT_PROP_KEY, String.valueOf(ConfigConstants.DEFAULT_PORT)));
+ int numGrpcChannels =
+ Integer.parseInt(
+ properties.getOrDefault(
+ ConfigConstants.NUM_GRPC_CHANNELS_PROP_KEY,
+ String.valueOf(ConfigConstants.DEFAULT_NUM_GRPC_CHANNELS)));
+ String maxCommitDelayProperty = properties.get(ConfigConstants.MAX_COMMIT_DELAY_PROP_KEY);
+ Integer maxCommitDelayMillis =
+ maxCommitDelayProperty != null ? Integer.parseInt(maxCommitDelayProperty) : null;
+ boolean enableBuiltInMetrics =
+ Boolean.parseBoolean(
+ properties.getOrDefault(ConfigConstants.ENABLE_BUILTIN_METRICS_PROP_KEY, "false"));
+
+ return newBuilder()
+ .databaseUri(properties.get(ConfigConstants.DATABASE_URI_PROP_KEY))
+ .hostAddress(InetAddress.getByName(host))
+ .port(port)
+ .spannerEndpoint(ConfigConstants.DEFAULT_SPANNER_ENDPOINT)
+ .numGrpcChannels(numGrpcChannels)
+ .maxCommitDelayMillis(maxCommitDelayMillis)
+ .enableBuiltInMetrics(enableBuiltInMetrics)
+ .build();
+ }
+
+ static Builder newBuilder() {
+ return new Builder();
+ }
+
+ static class Builder {
+ private String databaseUri;
+ private InetAddress hostAddress;
+ private int port;
+ private String spannerEndpoint;
+ private int numGrpcChannels;
+ @Nullable private Integer maxCommitDelayMillis;
+ private boolean enableBuiltInMetrics;
+
+ public Builder databaseUri(String databaseUri) {
+ this.databaseUri = databaseUri;
+ return this;
+ }
+
+ public Builder hostAddress(InetAddress hostAddress) {
+ this.hostAddress = hostAddress;
+ return this;
+ }
+
+ public Builder port(int port) {
+ this.port = port;
+ return this;
+ }
+
+ public Builder spannerEndpoint(String spannerEndpoint) {
+ this.spannerEndpoint = spannerEndpoint;
+ return this;
+ }
+
+ public Builder numGrpcChannels(int numGrpcChannels) {
+ this.numGrpcChannels = numGrpcChannels;
+ return this;
+ }
+
+ public Builder maxCommitDelayMillis(@Nullable Integer maxCommitDelayMillis) {
+ this.maxCommitDelayMillis = maxCommitDelayMillis;
+ return this;
+ }
+
+ public Builder enableBuiltInMetrics(boolean enableBuiltInMetrics) {
+ this.enableBuiltInMetrics = enableBuiltInMetrics;
+ return this;
+ }
+
+ public ListenerConfig build() {
+ return new ListenerConfig(this);
+ }
+ }
+}
+
+/** Encapsulates the configuration for the health check server. */
+final class HealthCheckConfig {
+ private final InetAddress hostAddress;
+ private final int port;
+
+ private HealthCheckConfig(Builder builder) {
+ this.hostAddress = builder.hostAddress;
+ this.port = builder.port;
+ }
+
+ public InetAddress getHostAddress() {
+ return hostAddress;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ static HealthCheckConfig fromEndpointString(String endpoint) throws UnknownHostException {
+ String[] parts = endpoint.split(":");
+ if (parts.length != 2) {
+ throw new IllegalArgumentException(
+ "Invalid health check endpoint format '" + endpoint + "'. Expected 'host:port'.");
+ }
+ String host = parts[0];
+ int port = parsePort(parts[1]);
+ return newBuilder().hostAddress(InetAddress.getByName(host)).port(port).build();
+ }
+
+ @Nullable
+ static HealthCheckConfig fromProperties(Map properties)
+ throws UnknownHostException {
+ String healthCheckPortStr = properties.get(ConfigConstants.HEALTH_CHECK_PORT_PROP_KEY);
+ if (healthCheckPortStr == null) {
+ return null;
+ }
+ String host =
+ properties.getOrDefault(ConfigConstants.HOST_PROP_KEY, ConfigConstants.DEFAULT_HOST);
+ int port = parsePort(healthCheckPortStr);
+ return newBuilder().hostAddress(InetAddress.getByName(host)).port(port).build();
+ }
+
+ private static int parsePort(String portStr) {
+ try {
+ int port = Integer.parseInt(portStr);
+ if (port < 0 || port > 65535) {
+ throw new IllegalArgumentException(
+ String.format("Invalid health check port '%s'. Must be between 0 and 65535", port));
+ }
+ return port;
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ String.format("Invalid health check port '%s'. Must be a number.", portStr), e);
+ }
+ }
+
+ static Builder newBuilder() {
+ return new Builder();
+ }
+
+ static class Builder {
+ private InetAddress hostAddress;
+ private int port;
+
+ public Builder hostAddress(InetAddress hostAddress) {
+ this.hostAddress = hostAddress;
+ return this;
+ }
+
+ public Builder port(int port) {
+ this.port = port;
+ return this;
+ }
+
+ public HealthCheckConfig build() {
+ return new HealthCheckConfig(this);
+ }
+ }
+}
diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/LauncherConfigParser.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/LauncherConfigParser.java
new file mode 100644
index 00000000..dc297538
--- /dev/null
+++ b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/LauncherConfigParser.java
@@ -0,0 +1,64 @@
+/*
+Copyright 2025 Google LLC
+
+Licensed 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 com.google.cloud.spanner.adapter;
+
+import com.google.cloud.spanner.adapter.configs.ConfigConstants;
+import com.google.cloud.spanner.adapter.configs.UserConfigs;
+import com.google.cloud.spanner.adapter.configs.YamlConfigLoader;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Parses the configuration for the {@link Launcher}. */
+public class LauncherConfigParser {
+ private static final Logger LOG = LoggerFactory.getLogger(LauncherConfigParser.class);
+
+ /**
+ * Parses the configuration from the given properties.
+ *
+ * @param properties The properties to parse.
+ * @return The parsed {@link LauncherConfig}.
+ * @throws IOException If the configuration file cannot be read.
+ */
+ public static LauncherConfig parse(Map properties) throws IOException {
+ final String configFilePath = properties.get(ConfigConstants.CONFIG_FILE_PROP_KEY);
+ if (configFilePath != null) {
+ LOG.info("Loading configuration from file: {}", configFilePath);
+ try (InputStream inputStream = new FileInputStream(configFilePath)) {
+ return parse(inputStream);
+ } catch (FileNotFoundException e) {
+ throw new IllegalArgumentException("Configuration file not found: " + configFilePath, e);
+ }
+ } else {
+ LOG.info("Loading configuration from system properties.");
+ return LauncherConfig.fromProperties(properties);
+ }
+ }
+
+ static LauncherConfig parse(InputStream inputStream) throws IOException {
+ try {
+ UserConfigs userConfigs = YamlConfigLoader.load(inputStream);
+ return LauncherConfig.fromUserConfigs(userConfigs);
+ } catch (Exception e) {
+ throw new IOException("Failed to parse configuration from input stream", e);
+ }
+ }
+}
diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/configs/ConfigConstants.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/configs/ConfigConstants.java
new file mode 100644
index 00000000..4caaa4dd
--- /dev/null
+++ b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/configs/ConfigConstants.java
@@ -0,0 +1,37 @@
+/*
+Copyright 2025 Google LLC
+
+Licensed 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 com.google.cloud.spanner.adapter.configs;
+
+/** Centralized constants for configuration keys and default values. */
+public final class ConfigConstants {
+
+ // Private constructor to prevent instantiation
+ private ConfigConstants() {}
+
+ public static final String DEFAULT_SPANNER_ENDPOINT = "spanner.googleapis.com:443";
+ public static final String DATABASE_URI_PROP_KEY = "databaseUri";
+ public static final String HOST_PROP_KEY = "host";
+ public static final String PORT_PROP_KEY = "port";
+ public static final String NUM_GRPC_CHANNELS_PROP_KEY = "numGrpcChannels";
+ public static final String DEFAULT_HOST = "0.0.0.0";
+ public static final int DEFAULT_PORT = 9042;
+ public static final int DEFAULT_NUM_GRPC_CHANNELS = 4;
+ public static final String MAX_COMMIT_DELAY_PROP_KEY = "maxCommitDelayMillis";
+ public static final String ENABLE_BUILTIN_METRICS_PROP_KEY = "enableBuiltInMetrics";
+ public static final String HEALTH_CHECK_PORT_PROP_KEY = "healthCheckPort";
+ public static final String CONFIG_FILE_PROP_KEY = "configFilePath";
+}
diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/configs/OperationConfigs.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/configs/OperationConfigs.java
deleted file mode 100644
index 55d91b67..00000000
--- a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/configs/OperationConfigs.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
-Copyright 2025 Google LLC
-
-Licensed 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 com.google.cloud.spanner.adapter.configs;
-
-import java.util.Map;
-
-/** Holds configurations for Spanner operations. */
-public class OperationConfigs {
- private final Integer maxCommitDelayMillis;
-
- public OperationConfigs(Integer maxCommitDelayMillis) {
- this.maxCommitDelayMillis = maxCommitDelayMillis;
- }
-
- public static OperationConfigs fromMap(Map yamlMap) {
- Integer maxCommitDelayMillis = (Integer) yamlMap.get("maxCommitDelayMillis");
- return new OperationConfigs(maxCommitDelayMillis);
- }
-
- public Integer getMaxCommitDelayMillis() {
- return maxCommitDelayMillis;
- }
-}
diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/configs/SessionConfigs.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/configs/SessionConfigs.java
deleted file mode 100644
index 4d53128d..00000000
--- a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/configs/SessionConfigs.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
-Copyright 2025 Google LLC
-
-Licensed 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 com.google.cloud.spanner.adapter.configs;
-
-import java.util.Map;
-
-/** Holds configurations for Spanner session management. */
-public class SessionConfigs {
- private final Integer numGrpcChannels;
-
- public SessionConfigs(Integer numGrpcChannels) {
- this.numGrpcChannels = numGrpcChannels;
- }
-
- public static SessionConfigs fromMap(Map yamlMap) {
- Integer numGrpcChannels = (Integer) yamlMap.get("numGrpcChannels");
- return new SessionConfigs(numGrpcChannels);
- }
-
- public Integer getNumGrpcChannels() {
- return numGrpcChannels;
- }
-}
diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/configs/SpannerConfigs.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/configs/SpannerConfigs.java
index ac403ed5..e0523c98 100644
--- a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/configs/SpannerConfigs.java
+++ b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/configs/SpannerConfigs.java
@@ -16,52 +16,44 @@
package com.google.cloud.spanner.adapter.configs;
+import static com.google.cloud.spanner.adapter.configs.ConfigConstants.DATABASE_URI_PROP_KEY;
+import static com.google.cloud.spanner.adapter.configs.ConfigConstants.MAX_COMMIT_DELAY_PROP_KEY;
+import static com.google.cloud.spanner.adapter.configs.ConfigConstants.NUM_GRPC_CHANNELS_PROP_KEY;
+
import java.util.Map;
/**
- * Represents the Spanner client configurations, including details about sessions and operations,
- * loaded from a YAML file.
+ * Represents the Spanner client configurations, including the database URI, the number of gRPC
+ * channels, and the maximum commit delay. This object is loaded from a YAML file.
*/
public class SpannerConfigs {
private final String databaseUri;
- private final SessionConfigs session;
- private final OperationConfigs operation;
+ private final Integer numGrpcChannels;
+ private final Integer maxCommitDelayMillis;
- public SpannerConfigs(String databaseUri, SessionConfigs session, OperationConfigs operation) {
+ public SpannerConfigs(String databaseUri, Integer numGrpcChannels, Integer maxCommitDelayMillis) {
this.databaseUri = databaseUri;
- this.session = session;
- this.operation = operation;
+ this.numGrpcChannels = numGrpcChannels;
+ this.maxCommitDelayMillis = maxCommitDelayMillis;
}
public static SpannerConfigs fromMap(Map yamlMap) {
- String databaseUri = (String) yamlMap.get("databaseUri");
-
- SessionConfigs session = null;
- if (yamlMap.containsKey("session")) {
- @SuppressWarnings("unchecked")
- Map sessionMap = (Map) yamlMap.get("session");
- session = SessionConfigs.fromMap(sessionMap);
- }
-
- OperationConfigs operation = null;
- if (yamlMap.containsKey("operation")) {
- @SuppressWarnings("unchecked")
- Map operationMap = (Map) yamlMap.get("operation");
- operation = OperationConfigs.fromMap(operationMap);
- }
+ String databaseUri = (String) yamlMap.get(DATABASE_URI_PROP_KEY);
+ Integer numGrpcChannels = (Integer) yamlMap.get(NUM_GRPC_CHANNELS_PROP_KEY);
+ Integer maxCommitDelayMillis = (Integer) yamlMap.get(MAX_COMMIT_DELAY_PROP_KEY);
- return new SpannerConfigs(databaseUri, session, operation);
+ return new SpannerConfigs(databaseUri, numGrpcChannels, maxCommitDelayMillis);
}
public String getDatabaseUri() {
return databaseUri;
}
- public SessionConfigs getSession() {
- return session;
+ public Integer getNumGrpcChannels() {
+ return numGrpcChannels;
}
- public OperationConfigs getOperation() {
- return operation;
+ public Integer getMaxCommitDelayMillis() {
+ return maxCommitDelayMillis;
}
}
diff --git a/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/LauncherConfigParserTest.java b/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/LauncherConfigParserTest.java
new file mode 100644
index 00000000..6db05ec4
--- /dev/null
+++ b/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/LauncherConfigParserTest.java
@@ -0,0 +1,194 @@
+/*
+Copyright 2025 Google LLC
+
+Licensed 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 com.google.cloud.spanner.adapter;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+import com.google.cloud.spanner.adapter.configs.GlobalClientConfigs;
+import com.google.cloud.spanner.adapter.configs.ListenerConfigs;
+import com.google.cloud.spanner.adapter.configs.SpannerConfigs;
+import com.google.cloud.spanner.adapter.configs.UserConfigs;
+import com.google.cloud.spanner.adapter.configs.YamlConfigLoader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.MockedStatic;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class LauncherConfigParserTest {
+
+ private static final String DEFAULT_DATABASE_URI = "projects/p/instances/i/databases/d";
+
+ @Test
+ public void testParse_withValidConfigFile() throws Exception {
+ UserConfigs userConfigs =
+ new UserConfigs(
+ new GlobalClientConfigs("spanner.googleapis.com:443", true, "127.0.0.1:8080"),
+ Arrays.asList(
+ new ListenerConfigs(
+ "listener_1",
+ "127.0.0.1",
+ 9042,
+ new SpannerConfigs("projects/p/instances/i/databases/d-1-config-test", 4, 5)),
+ new ListenerConfigs(
+ "listener_2",
+ "0.0.0.0",
+ 9043,
+ new SpannerConfigs(
+ "projects/p/instances/i/databases/d-2-config-test", 8, null))));
+
+ try (MockedStatic mockedLoader = mockStatic(YamlConfigLoader.class);
+ MockedStatic mockedInetAddress = mockStatic(InetAddress.class)) {
+ InetAddress mockAddress = mock(InetAddress.class);
+ mockedInetAddress
+ .when(() -> InetAddress.getByName(any(String.class)))
+ .thenReturn(mockAddress);
+ mockedLoader
+ .when(() -> YamlConfigLoader.load(any(InputStream.class)))
+ .thenReturn(userConfigs);
+
+ LauncherConfig config = LauncherConfigParser.parse(mock(InputStream.class));
+
+ assertThat(config.getListeners()).hasSize(2);
+ assertThat(config.getHealthCheckConfig()).isNotNull();
+ }
+ }
+
+ @Test
+ public void testParse_withConfigFileAndOtherParams_usesConfigFile() throws Exception {
+ String configFile = getClass().getClassLoader().getResource("valid-config.yaml").getFile();
+ Map properties = new HashMap<>();
+ properties.put("configFilePath", configFile);
+ // The following properties should be ignored, as the config file takes precedence.
+ properties.put("databaseUri", "projects/p/instances/i/databases/d-from-props");
+ properties.put("port", "9044");
+
+ LauncherConfig config = LauncherConfigParser.parse(properties);
+
+ assertThat(config.getListeners()).hasSize(2);
+ ListenerConfig listenerConfig1 = config.getListeners().get(0);
+ assertThat(listenerConfig1.getDatabaseUri())
+ .isEqualTo("projects/my-project/instances/my-instance/databases/my-database");
+ assertThat(listenerConfig1.getPort()).isEqualTo(9042);
+
+ ListenerConfig listenerConfig2 = config.getListeners().get(1);
+ assertThat(listenerConfig2.getDatabaseUri())
+ .isEqualTo("projects/my-project/instances/my-instance/databases/my-database-2");
+ assertThat(listenerConfig2.getPort()).isEqualTo(9043);
+ }
+
+ @Test
+ public void testParse_withSystemProperties() throws Exception {
+ Map properties = new HashMap<>();
+ properties.put("databaseUri", DEFAULT_DATABASE_URI);
+ properties.put("host", "127.0.0.1");
+ properties.put("port", "9042");
+ properties.put("numGrpcChannels", "8");
+ properties.put("maxCommitDelayMillis", "100");
+ properties.put("enableBuiltInMetrics", "true");
+ properties.put("healthCheckPort", "8080");
+
+ try (MockedStatic mockedInetAddress = mockStatic(InetAddress.class)) {
+ InetAddress mockAddress = mock(InetAddress.class);
+ when(mockAddress.getHostAddress()).thenReturn("127.0.0.1");
+ mockedInetAddress.when(() -> InetAddress.getByName("127.0.0.1")).thenReturn(mockAddress);
+
+ LauncherConfig config = LauncherConfigParser.parse(properties);
+ assertThat(config.getListeners()).hasSize(1);
+ ListenerConfig listenerConfig = config.getListeners().get(0);
+ assertThat(listenerConfig.getDatabaseUri()).isEqualTo(DEFAULT_DATABASE_URI);
+ assertThat(listenerConfig.getPort()).isEqualTo(9042);
+ assertThat(listenerConfig.getHostAddress().getHostAddress()).isEqualTo("127.0.0.1");
+ assertThat(listenerConfig.getNumGrpcChannels()).isEqualTo(8);
+ assertThat(listenerConfig.getMaxCommitDelayMillis()).isEqualTo(100);
+ assertThat(listenerConfig.isEnableBuiltInMetrics()).isTrue();
+ assertThat(config.getHealthCheckConfig()).isNotNull();
+ assertThat(config.getHealthCheckConfig().getPort()).isEqualTo(8080);
+ }
+ }
+
+ @Test
+ public void testParse_withMissingDatabaseUri_throwsIllegalArgumentException() {
+ Map properties = Collections.emptyMap();
+
+ IllegalArgumentException thrown =
+ assertThrows(IllegalArgumentException.class, () -> LauncherConfigParser.parse(properties));
+ assertThat(thrown.getMessage()).contains("Spanner database URI not set.");
+ }
+
+ @Test
+ public void testParse_withInvalidHealthCheckPort_throwsIllegalArgumentException() {
+ Map properties = new HashMap<>();
+ properties.put("databaseUri", DEFAULT_DATABASE_URI);
+ properties.put("healthCheckPort", "99999");
+
+ IllegalArgumentException thrown =
+ assertThrows(IllegalArgumentException.class, () -> LauncherConfigParser.parse(properties));
+ assertThat(thrown.getMessage()).contains("Invalid health check port '99999'");
+ }
+
+ @Test
+ public void testParse_withInvalidConfigFile_throwsIOException() {
+ try (MockedStatic mockedLoader = mockStatic(YamlConfigLoader.class)) {
+ mockedLoader
+ .when(() -> YamlConfigLoader.load(any(InputStream.class)))
+ .thenReturn(new UserConfigs(null, null));
+ IOException thrown =
+ assertThrows(
+ IOException.class, () -> LauncherConfigParser.parse(mock(InputStream.class)));
+ assertThat(thrown.getCause()).isInstanceOf(IllegalArgumentException.class);
+ assertThat(thrown.getCause().getMessage())
+ .contains("No listeners defined in the configuration.");
+ }
+ }
+
+ @Test
+ public void testParse_withUnknownHost_throwsIOException() throws IOException {
+ UserConfigs userConfigs =
+ new UserConfigs(
+ new GlobalClientConfigs("spanner.googleapis.com:443", true, "unknown-host:8080"),
+ Collections.singletonList(
+ new ListenerConfigs(
+ "listener_1",
+ "unknown-host",
+ 9042,
+ new SpannerConfigs(DEFAULT_DATABASE_URI, 4, 5))));
+ try (MockedStatic mockedLoader = mockStatic(YamlConfigLoader.class)) {
+ mockedLoader
+ .when(() -> YamlConfigLoader.load(any(InputStream.class)))
+ .thenReturn(userConfigs);
+
+ IOException thrown =
+ assertThrows(
+ IOException.class, () -> LauncherConfigParser.parse(mock(InputStream.class)));
+ assertThat(thrown.getCause()).isInstanceOf(UnknownHostException.class);
+ }
+ }
+}
diff --git a/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/LauncherTest.java b/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/LauncherTest.java
new file mode 100644
index 00000000..e5e6c91a
--- /dev/null
+++ b/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/LauncherTest.java
@@ -0,0 +1,194 @@
+/*
+Copyright 2025 Google LLC
+
+Licensed 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 com.google.cloud.spanner.adapter;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.cloud.spanner.adapter.configs.GlobalClientConfigs;
+import com.google.cloud.spanner.adapter.configs.ListenerConfigs;
+import com.google.cloud.spanner.adapter.configs.SpannerConfigs;
+import com.google.cloud.spanner.adapter.configs.UserConfigs;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.net.InetAddress;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class LauncherTest {
+
+ private static final String DEFAULT_DATABASE_URI = "projects/p/instances/i/databases/d";
+
+ @Mock private Launcher.AdapterFactory mockAdapterFactory;
+ @Mock private Adapter mockAdapter;
+ @Mock private HealthCheckServer mockHealthCheckServer;
+ @Captor private ArgumentCaptor shutdownHookCaptor;
+ @Captor private ArgumentCaptor adapterOptionsCaptor;
+
+ private Launcher launcher;
+ private PrintStream originalSystemOut;
+
+ @Before
+ public void setUp() throws IOException {
+ launcher = new Launcher(mockAdapterFactory);
+ when(mockAdapterFactory.createAdapter(any())).thenReturn(mockAdapter);
+ when(mockAdapterFactory.createHealthCheckServer(any(), anyInt()))
+ .thenReturn(mockHealthCheckServer);
+
+ // Redirect System.out to avoid console output during test
+ originalSystemOut = System.out;
+ System.setOut(new PrintStream(new ByteArrayOutputStream()));
+ }
+
+ @After
+ public void tearDown() {
+ System.setOut(originalSystemOut);
+ }
+
+ @Test
+ public void testRun_withMultipleListeners_startsMultipleAdapters() throws Exception {
+ UserConfigs userConfigs =
+ new UserConfigs(
+ new GlobalClientConfigs("spanner.googleapis.com:443", true, "127.0.0.1:8080"),
+ Arrays.asList(
+ new ListenerConfigs(
+ "listener_1",
+ "127.0.0.1",
+ 9042,
+ new SpannerConfigs("projects/p/instances/i/databases/d-1-config-test", 4, 5)),
+ new ListenerConfigs(
+ "listener_2",
+ "0.0.0.0",
+ 9043,
+ new SpannerConfigs(
+ "projects/p/instances/i/databases/d-2-config-test", 8, null))));
+ LauncherConfig config = LauncherConfig.fromUserConfigs(userConfigs);
+
+ launcher.run(config);
+
+ verify(mockAdapterFactory, times(2)).createAdapter(adapterOptionsCaptor.capture());
+ verify(mockAdapterFactory, times(1)).createHealthCheckServer(any(), eq(8080));
+ verify(mockAdapter, times(2)).start();
+ verify(mockHealthCheckServer).start();
+ verify(mockHealthCheckServer).setReady(true);
+
+ AdapterOptions options1 = adapterOptionsCaptor.getAllValues().get(0);
+ assertThat(options1.getDatabaseUri())
+ .isEqualTo("projects/p/instances/i/databases/d-1-config-test");
+ assertThat(options1.getTcpPort()).isEqualTo(9042);
+ assertThat(options1.getInetAddress()).isEqualTo(InetAddress.getByName("127.0.0.1"));
+
+ AdapterOptions options2 = adapterOptionsCaptor.getAllValues().get(1);
+ assertThat(options2.getDatabaseUri())
+ .isEqualTo("projects/p/instances/i/databases/d-2-config-test");
+ assertThat(options2.getTcpPort()).isEqualTo(9043);
+ assertThat(options2.getInetAddress()).isEqualTo(InetAddress.getByName("0.0.0.0"));
+ }
+
+ @Test
+ public void testRun_withSingleListener_startsAdapterWithOptions() throws Exception {
+ Map properties = new HashMap<>();
+ properties.put("databaseUri", DEFAULT_DATABASE_URI);
+ properties.put("host", "127.0.0.1");
+ properties.put("port", "9042");
+ properties.put("numGrpcChannels", "8");
+ properties.put("maxCommitDelayMillis", "100");
+ properties.put("enableBuiltInMetrics", "true");
+ properties.put("healthCheckPort", "8080");
+ LauncherConfig config = LauncherConfig.fromProperties(properties);
+
+ launcher.run(config);
+
+ verify(mockAdapterFactory, times(1)).createAdapter(adapterOptionsCaptor.capture());
+ verify(mockAdapterFactory, times(1)).createHealthCheckServer(any(), eq(8080));
+ verify(mockAdapter, times(1)).start();
+ verify(mockHealthCheckServer).start();
+ verify(mockHealthCheckServer).setReady(true);
+
+ AdapterOptions options = adapterOptionsCaptor.getValue();
+ assertThat(options.getDatabaseUri()).isEqualTo(DEFAULT_DATABASE_URI);
+ assertThat(options.getTcpPort()).isEqualTo(9042);
+ assertThat(options.getInetAddress()).isEqualTo(InetAddress.getByName("127.0.0.1"));
+ assertThat(options.getNumGrpcChannels()).isEqualTo(8);
+ assertThat(options.getMaxCommitDelay().get().toMillis()).isEqualTo(100);
+ }
+
+ @Test
+ public void testRun_withNoHealthCheckPort_noHealthCheckServerIsCreated() throws Exception {
+ Map properties = new HashMap<>();
+ properties.put("databaseUri", DEFAULT_DATABASE_URI);
+ LauncherConfig config = LauncherConfig.fromProperties(properties);
+
+ launcher.run(config);
+
+ verify(mockAdapterFactory, times(1)).createAdapter(any(AdapterOptions.class));
+ verify(mockAdapter, times(1)).start();
+ verify(mockAdapterFactory, never()).createHealthCheckServer(any(), anyInt());
+ verify(mockHealthCheckServer, never()).start();
+ verify(mockHealthCheckServer, never()).setReady(anyBoolean());
+ }
+
+ @Test
+ public void testRun_whenAdapterStartFails_healthCheckIsNotReady() throws Exception {
+ Map properties = new HashMap<>();
+ properties.put("databaseUri", DEFAULT_DATABASE_URI);
+ properties.put("healthCheckPort", "8080");
+ LauncherConfig config = LauncherConfig.fromProperties(properties);
+ doThrow(new RuntimeException("Failed to start adapter")).when(mockAdapter).start();
+
+ assertThrows(IllegalStateException.class, () -> launcher.run(config));
+ verify(mockHealthCheckServer).start();
+ verify(mockHealthCheckServer).setReady(false);
+ // Verify that the shutdown logic was called to clean up.
+ verify(mockHealthCheckServer, times(1)).stop();
+ verify(mockAdapter, times(1)).stop();
+ }
+
+ @Test
+ public void testShutdownHook_stopsAllInstances() throws Exception {
+ Map properties = new HashMap<>();
+ properties.put("databaseUri", DEFAULT_DATABASE_URI);
+ properties.put("healthCheckPort", "8080");
+ LauncherConfig config = LauncherConfig.fromProperties(properties);
+
+ launcher.run(config);
+ launcher.shutdown();
+
+ verify(mockAdapter, times(1)).stop();
+ verify(mockHealthCheckServer, times(1)).stop();
+ }
+}
diff --git a/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/configs/YamlConfigLoaderTest.java b/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/configs/YamlConfigLoaderTest.java
index 63f481e0..ede561bd 100644
--- a/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/configs/YamlConfigLoaderTest.java
+++ b/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/configs/YamlConfigLoaderTest.java
@@ -56,11 +56,9 @@ public void testLoad_validYamlFile_parsesCorrectly() throws IOException {
assertThat(listener1.getSpanner().getDatabaseUri())
.isEqualTo("projects/my-project/instances/my-instance/databases/my-database");
- assertThat(listener1.getSpanner().getSession()).isNotNull();
- assertThat(listener1.getSpanner().getSession().getNumGrpcChannels()).isEqualTo(4);
+ assertThat(listener1.getSpanner().getNumGrpcChannels()).isEqualTo(4);
- assertThat(listener1.getSpanner().getOperation()).isNotNull();
- assertThat(listener1.getSpanner().getOperation().getMaxCommitDelayMillis()).isEqualTo(100);
+ assertThat(listener1.getSpanner().getMaxCommitDelayMillis()).isEqualTo(100);
// Verify listener_2
ListenerConfigs listener2 = listeners.get(1);
@@ -73,10 +71,9 @@ public void testLoad_validYamlFile_parsesCorrectly() throws IOException {
assertThat(listener2.getSpanner().getDatabaseUri())
.isEqualTo("projects/my-project/instances/my-instance/databases/my-database-2");
- assertThat(listener2.getSpanner().getSession()).isNotNull();
- assertThat(listener2.getSpanner().getSession().getNumGrpcChannels()).isEqualTo(8);
+ assertThat(listener2.getSpanner().getNumGrpcChannels()).isEqualTo(8);
- assertThat(listener2.getSpanner().getOperation()).isNull();
+ assertThat(listener2.getSpanner().getMaxCommitDelayMillis()).isNull();
}
}
@@ -122,8 +119,8 @@ public void testLoad_missingOptionalFields_parsesCorrectly() throws IOException
assertThat(listener.getSpanner()).isNotNull();
assertThat(listener.getSpanner().getDatabaseUri()).isEqualTo("test");
assertThat(listener.getPort()).isNull();
- assertThat(listener.getSpanner().getSession()).isNull();
- assertThat(listener.getSpanner().getOperation()).isNull();
+ assertThat(listener.getSpanner().getNumGrpcChannels()).isNull();
+ assertThat(listener.getSpanner().getMaxCommitDelayMillis()).isNull();
}
}
diff --git a/google-cloud-spanner-cassandra/src/test/resources/valid-config.yaml b/google-cloud-spanner-cassandra/src/test/resources/valid-config.yaml
index 7b7b24ed..bd4b66f5 100644
--- a/google-cloud-spanner-cassandra/src/test/resources/valid-config.yaml
+++ b/google-cloud-spanner-cassandra/src/test/resources/valid-config.yaml
@@ -12,15 +12,12 @@ listeners:
port: 9042
spanner:
databaseUri: "projects/my-project/instances/my-instance/databases/my-database"
- session:
- numGrpcChannels: 4
- operation:
- maxCommitDelayMillis: 100
+ numGrpcChannels: 4
+ maxCommitDelayMillis: 100
- # Configuration for listener_2
name: "listener_2"
host: "127.0.0.2"
port: 9043
spanner:
databaseUri: "projects/my-project/instances/my-instance/databases/my-database-2"
- session:
- numGrpcChannels: 8
+ numGrpcChannels: 8