feat: add configurable health check point

README.md

@@ -155,11 +155,13 @@ For non-Java applications or tools like `cqlsh`, you can run the Spanner Cassand
     -Dhost=127.0.0.1 \
     -Dport=9042 \
     -DnumGrpcChannels=4 \
+    -DhealthCheckPort=8080 \
     -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`.
+    * You can optionally specify a `-DhealthCheckPort`. If you specify a port number, a health check endpoint will be started on that port. The IP address of the health check endpoint will be the same as the IP address that is used for the client. The health check endpoint will return an HTTP status 200 OK if the client is up and running, and a 503 Service Unavailable if it is not. The health check endpoint is not enabled by default.
 
 ## View and manage client-side metrics
 

google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Adapter.java

@@ -15,7 +15,18 @@
 */
 package com.google.cloud.spanner.adapter;
 
-import static com.google.cloud.spanner.adapter.util.ThreadFactoryUtil.tryCreateVirtualThreadPerTaskExecutor;
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketException;
+import java.util.Collections;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import javax.annotation.concurrent.NotThreadSafe;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import com.google.api.gax.core.CredentialsProvider;
 import com.google.api.gax.core.FixedCredentialsProvider;
@@ -28,20 +39,11 @@
 import com.google.auth.Credentials;
 import com.google.auth.oauth2.GoogleCredentials;
 import com.google.cloud.NoCredentials;
+import static com.google.cloud.spanner.adapter.util.ThreadFactoryUtil.tryCreateVirtualThreadPerTaskExecutor;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableSet;
 import com.google.spanner.adapter.v1.AdapterClient;
 import com.google.spanner.adapter.v1.AdapterSettings;
-import java.io.IOException;
-import java.net.ServerSocket;
-import java.net.Socket;
-import java.net.SocketException;
-import java.util.Collections;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import javax.annotation.concurrent.NotThreadSafe;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /** Manages client connections, acting as an intermediary for communication with Spanner. */
 @NotThreadSafe
@@ -157,7 +159,6 @@ void start() {
 
       started = true;
       LOG.info("Adapter started for database '{}'.", options.getDatabaseUri());
-
     } catch (IOException | RuntimeException e) {
       throw new AdapterStartException(e);
     }

google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/HealthServer.java

@@ -0,0 +1,77 @@
+/*
+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 java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+
+class HealthServer {
+  private static final Logger LOG = LoggerFactory.getLogger(HealthServer.class);
+  private static final int HTTP_OK_STATUS = 200;
+  private static final int HTTP_UNAVAILABLE_STATUS = 503;
+
+  private final HttpServer server;
+  private final AtomicBoolean isReady = new AtomicBoolean(false);
+
+  HealthServer(InetAddress address, int port) throws IOException {
+    this.server = HttpServer.create(new InetSocketAddress(address, port), 0);
+    this.server.createContext("/debug/health", new HealthHandler());
+    this.server.setExecutor(null); // creates a default executor
+  }
+
+  void start() {
+    server.start();
+    LOG.info("Health server started on {}", server.getAddress());
+  }
+
+  void stop() {
+    server.stop(0);
+    LOG.info("Health server stopped.");
+  }
+
+  void setReady(boolean ready) {
+    isReady.set(ready);
+  }
+
+  private class HealthHandler implements HttpHandler {
+    @Override
+    public void handle(HttpExchange exchange) throws IOException {
+      if (isReady.get()) {
+        sendResponse(exchange, HTTP_OK_STATUS, "All listeners are up and running");
+      } else {
+        sendResponse(exchange, HTTP_UNAVAILABLE_STATUS, "Service Unavailable");
+      }
+    }
+
+    private void sendResponse(HttpExchange exchange, int statusCode, String response)
+        throws IOException {
+      exchange.sendResponseHeaders(statusCode, response.length());
+      try (OutputStream os = exchange.getResponseBody()) {
+        os.write(response.getBytes());
+      }
+    }
+  }
+}

google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Launcher.java

@@ -13,18 +13,21 @@
 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;
-import com.google.cloud.spanner.adapter.metrics.BuiltInMetricsRecorder;
-import com.google.spanner.adapter.v1.DatabaseName;
-import io.opentelemetry.api.OpenTelemetry;
 import java.net.InetAddress;
 import java.time.Duration;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.cloud.spanner.adapter.metrics.BuiltInMetricsProvider;
+import com.google.cloud.spanner.adapter.metrics.BuiltInMetricsRecorder;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.spanner.adapter.v1.DatabaseName;
+
+import io.opentelemetry.api.OpenTelemetry;
+
 /**
  * Main entry point for running a Spanner Cassandra Adapter as a stand-alone application.
  *
@@ -43,6 +46,8 @@
  *       with Spanner. Defaults to 4.
  *   <li>{@code maxCommitDelayMillis}: (Optional) The max commit delay to set in requests to
  *       optimize write throughput, in milliseconds. Defaults to none.
+ *   <li>{@code healthCheckPort}: (Optional) The port number for the health check server. Defaults
+ *       to none.
  * </ul>
  *
  * Example usage:
@@ -53,6 +58,7 @@
  * -Dport=9042 \
  * -DnumGrpcChannels=4 \
  * -DmaxCommitDelayMillis=5 \
+ * -DhealthCheckPort=8080 \
  * -cp path/to/your/spanner-cassandra-launcher.jar com.google.cloud.spanner.adapter.SpannerCassandraLauncher
  * </pre>
  *
@@ -72,6 +78,48 @@ public class Launcher {
   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;
+
+  @VisibleForTesting
+  Launcher(Adapter adapter) {
+    this.adapter = adapter;
+  }
+
+  void start(InetAddress address, int healthCheckPort) throws Exception {
+    HealthServer healthServer = null;
+    if (healthCheckPort > 0) {
+      healthServer = new HealthServer(address, healthCheckPort);
+      healthServer.start();
+    }
+
+    adapter.start();
+    if (healthServer != null) {
+      healthServer.setReady(true);
+    }
+
+    final HealthServer finalHealthServer = healthServer;
+    Runtime.getRuntime()
+        .addShutdownHook(
+            new Thread(
+                () -> {
+                  if (finalHealthServer != null) {
+                    finalHealthServer.stop();
+                  }
+                  try {
+                    adapter.stop();
+                  } catch (Exception e) {
+                    LOG.error("Failed to stop adapter", e);
+                  }
+                }));
+
+    try {
+      Thread.currentThread().join();
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+    }
+  }
 
   public static void main(String[] args) throws Exception {
     final String databaseUri = System.getProperty(DATABASE_URI_PROP_KEY);
@@ -83,6 +131,8 @@ public static void main(String[] args) throws Exception {
     final String maxCommitDelayProperty = System.getProperty(MAX_COMMIT_DELAY_PROP_KEY);
     final boolean enableBuiltInMetrics =
         Boolean.parseBoolean(System.getProperty(ENABLE_BUILTIN_METRICS_PROP_KEY, "false"));
+    final int healthCheckPort =
+        Integer.parseInt(System.getProperty(HEALTH_CHECK_PORT_PROP_KEY, "0"));
 
     if (databaseUri == null) {
       throw new IllegalArgumentException(
@@ -112,23 +162,16 @@ public static void main(String[] args) throws Exception {
       opBuilder.maxCommitDelay(Duration.ofMillis(Integer.parseInt(maxCommitDelayProperty)));
     }
 
-    Adapter adapter = new Adapter(opBuilder.build());
     LOG.info(
         "Starting Adapter for Spanner database {} on {}:{} with {} gRPC channels, max commit"
-            + " delay of {} and built-in metrics enabled: {}",
+            + " delay of {} and built-in metrics enabled: {},",
         databaseUri,
         inetAddress,
         port,
         numGrpcChannels,
         maxCommitDelayProperty,
         enableBuiltInMetrics);
 
-    adapter.start();
-
-    try {
-      Thread.currentThread().join();
-    } catch (InterruptedException e) {
-      Thread.currentThread().interrupt();
-    }
+    new Launcher(new Adapter(opBuilder.build())).start(inetAddress, healthCheckPort);
   }
 }

google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/AdapterTest.java

@@ -16,79 +16,93 @@
 
 package com.google.cloud.spanner.adapter;
 
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.HttpURLConnection;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.URL;
+import java.net.UnknownHostException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThrows;
+import org.junit.Before;
+import org.junit.Test;
 import static org.mockito.ArgumentMatchers.any;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.mockConstruction;
 import static org.mockito.Mockito.mockStatic;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import com.google.api.gax.core.NoCredentialsProvider;
 import com.google.auth.oauth2.GoogleCredentials;
 import com.google.spanner.adapter.v1.AdapterClient;
 import com.google.spanner.adapter.v1.AdapterSettings;
-import com.google.spanner.adapter.v1.CreateSessionRequest;
 import com.google.spanner.adapter.v1.Session;
-import java.net.InetAddress;
-import java.net.ServerSocket;
-import java.net.UnknownHostException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import org.junit.Before;
-import org.junit.Test;
-import org.mockito.MockedConstruction;
-import org.mockito.MockedStatic;
 
 public final class AdapterTest {
-  private static final String TEST_HOST = "localhost";
   private static final String TEST_DATABASE_URI =
       "projects/test-project/instances/test-instance/databases/test-db";
   private static final int TEST_PORT = 12345;
   private final InetAddress inetAddress;
   private Adapter adapter;
 
   public AdapterTest() throws UnknownHostException {
-    inetAddress = InetAddress.getByName("0.0.0.0");
+    inetAddress = InetAddress.getByName("localhost");
   }
 
   @Before
-  public void setUp() {
+  public void setUp() throws IOException {
     AdapterOptions options =
         new AdapterOptions.Builder()
-            .spannerEndpoint(TEST_HOST)
+            .spannerEndpoint("localhost:1234")
             .tcpPort(TEST_PORT)
             .databaseUri(TEST_DATABASE_URI)
             .inetAddress(inetAddress)
+            .credentials(NoCredentialsProvider.create().getCredentials())
             .build();
 
     adapter = new Adapter(options);
   }
 
+  private int getAvailablePort() throws IOException {
+    try (ServerSocket serverSocket = new ServerSocket(0)) {
+      return serverSocket.getLocalPort();
+    }
+  }
+
   @Test
   public void successfulStartStopFlow() throws Exception {
-
     try (MockedConstruction<ServerSocket> mockedServerSocketConstruction =
             mockConstruction(ServerSocket.class);
         MockedStatic<Executors> mockedExecutors = mockStatic(Executors.class);
         MockedStatic<AdapterClient> mockedStaticAdapterClient = mockStatic(AdapterClient.class);
         MockedStatic<GoogleCredentials> mockedGoogleCredentials =
-            mockStatic(GoogleCredentials.class)) {
+            mockStatic(GoogleCredentials.class);
+        MockedConstruction<SessionManager> mockedSessionManager =
+            mockConstruction(
+                SessionManager.class,
+                (mock, context) -> {
+                  when(mock.getSession()).thenReturn(mock(Session.class));
+                })) {
       AdapterClient mockAdapterClient = mock(AdapterClient.class);
-      Session mockSession = mock(Session.class);
       mockedGoogleCredentials.when(GoogleCredentials::getApplicationDefault).thenReturn(null);
       mockedStaticAdapterClient
           .when(() -> AdapterClient.create(any(AdapterSettings.class)))
           .thenReturn(mockAdapterClient);
-      when(mockAdapterClient.createSession(any())).thenReturn(mockSession);
       ExecutorService mockExecutor = mock(ExecutorService.class);
       mockedExecutors.when(Executors::newCachedThreadPool).thenReturn(mockExecutor);
 
       adapter.start();
       adapter.stop();
 
-      verify(mockAdapterClient, times(1)).createSession(any(CreateSessionRequest.class));
+      verify(mockedSessionManager.constructed().get(0), times(1)).getSession();
       verify(mockExecutor).execute(any(Runnable.class));
       // Verify ServerSocket was constructed
       assertEquals(1, mockedServerSocketConstruction.constructed().size());