[grid] External datastore JDBC-backed for Session Queue
diff --git a/java/src/org/openqa/selenium/grid/sessionqueue/jdbc/BUILD.bazel b/java/src/org/openqa/selenium/grid/sessionqueue/jdbc/BUILD.bazel
new file mode 100644
index 0000000..e5cd4f6
--- /dev/null
+++ b/java/src/org/openqa/selenium/grid/sessionqueue/jdbc/BUILD.bazel
@@ -0,0 +1,27 @@
+load("@rules_jvm_external//:defs.bzl", "artifact")
+load("//java:defs.bzl", "java_export")
+load("//java:version.bzl", "SE_VERSION")
+
+java_export(
+    name = "jdbc",
+    srcs = glob(["*.java"]),
+    maven_coordinates = "org.seleniumhq.selenium:selenium-session-queue-jdbc:%s" % SE_VERSION,
+    pom_template = "//java/src/org/openqa/selenium:template-pom",
+    tags = [
+        "release-artifact",
+    ],
+    visibility = [
+        "//visibility:public",
+    ],
+    exports = [
+        "//java/src/org/openqa/selenium/grid",
+    ],
+    deps = [
+        "//java:auto-service",
+        "//java/src/org/openqa/selenium/grid",
+        "//java/src/org/openqa/selenium/json",
+        "//java/src/org/openqa/selenium/remote",
+        artifact("com.beust:jcommander"),
+        artifact("com.google.guava:guava"),
+    ],
+)
diff --git a/java/src/org/openqa/selenium/grid/sessionqueue/jdbc/JdbcBackedSessionQueue.java b/java/src/org/openqa/selenium/grid/sessionqueue/jdbc/JdbcBackedSessionQueue.java
new file mode 100644
index 0000000..59db4b3
--- /dev/null
+++ b/java/src/org/openqa/selenium/grid/sessionqueue/jdbc/JdbcBackedSessionQueue.java
@@ -0,0 +1,435 @@
+// Licensed to the Software Freedom Conservancy (SFC) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The SFC 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.openqa.selenium.grid.sessionqueue.jdbc;
+
+import static org.openqa.selenium.remote.tracing.Tags.EXCEPTION;
+
+import java.io.Closeable;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import org.openqa.selenium.Capabilities;
+import org.openqa.selenium.SessionNotCreatedException;
+import org.openqa.selenium.grid.config.Config;
+import org.openqa.selenium.grid.config.ConfigException;
+import org.openqa.selenium.grid.data.CreateSessionResponse;
+import org.openqa.selenium.grid.data.RequestId;
+import org.openqa.selenium.grid.data.SessionRequest;
+import org.openqa.selenium.grid.data.SessionRequestCapability;
+import org.openqa.selenium.grid.log.LoggingOptions;
+import org.openqa.selenium.grid.security.Secret;
+import org.openqa.selenium.grid.security.SecretOptions;
+import org.openqa.selenium.grid.sessionqueue.NewSessionQueue;
+import org.openqa.selenium.internal.Either;
+import org.openqa.selenium.internal.Require;
+import org.openqa.selenium.json.Json;
+import org.openqa.selenium.remote.http.HttpResponse;
+import org.openqa.selenium.remote.tracing.AttributeKey;
+import org.openqa.selenium.remote.tracing.AttributeMap;
+import org.openqa.selenium.remote.tracing.Span;
+import org.openqa.selenium.remote.tracing.Status;
+import org.openqa.selenium.remote.tracing.Tracer;
+
+public class JdbcBackedSessionQueue extends NewSessionQueue implements Closeable {
+  private static final Logger LOG = Logger.getLogger(JdbcBackedSessionQueue.class.getName());
+  private static final String TABLE_NAME = "session_queue";
+  private static final Json JSON = new Json();
+  private static final String DATABASE_STATEMENT = AttributeKey.DATABASE_STATEMENT.getKey();
+  private static final String DATABASE_OPERATION = AttributeKey.DATABASE_OPERATION.getKey();
+  private static final String DATABASE_USER = AttributeKey.DATABASE_USER.getKey();
+  private static final String DATABASE_CONNECTION_STRING =
+      AttributeKey.DATABASE_CONNECTION_STRING.getKey();
+  private static String jdbcUser;
+  private static String jdbcUrl;
+  private final Connection connection;
+
+  public JdbcBackedSessionQueue(
+      Tracer tracer, Secret registrationSecret, Connection jdbcConnection) {
+    super(tracer, registrationSecret);
+    this.connection = Require.nonNull("JDBC Connection Object", jdbcConnection);
+    ensureTableExists();
+  }
+
+  public static NewSessionQueue create(Config config) {
+    Tracer tracer = new LoggingOptions(config).getTracer();
+    Secret secret = new SecretOptions(config).getRegistrationSecret();
+    Connection connection;
+    try {
+      JdbcSessionQueueOptions sessionQueueOptions = new JdbcSessionQueueOptions(config);
+      jdbcUser = sessionQueueOptions.getJdbcUser();
+      jdbcUrl = sessionQueueOptions.getJdbcUrl();
+      connection = sessionQueueOptions.getJdbcConnection();
+    } catch (SQLException e) {
+      throw new ConfigException(e);
+    }
+    return new JdbcBackedSessionQueue(tracer, secret, connection);
+  }
+
+  @Override
+  public boolean isReady() {
+    try {
+      return !connection.isClosed();
+    } catch (SQLException e) {
+      return false;
+    }
+  }
+
+  @Override
+  public boolean peekEmpty() {
+    try (Span span = tracer.getCurrentContext().createSpan("SELECT COUNT(*) FROM session_queue")) {
+      AttributeMap attributeMap = tracer.createAttributeMap();
+      setCommonSpanAttributes(span);
+      setCommonEventAttributes(attributeMap);
+
+      String sql = "SELECT COUNT(*) FROM " + TABLE_NAME;
+      try (PreparedStatement stmt = connection.prepareStatement(sql)) {
+        String statementStr = stmt.toString();
+        span.setAttribute(DATABASE_STATEMENT, statementStr);
+        span.setAttribute(DATABASE_OPERATION, "select");
+        attributeMap.put(DATABASE_STATEMENT, statementStr);
+        attributeMap.put(DATABASE_OPERATION, "select");
+
+        ResultSet rs = stmt.executeQuery();
+        if (rs.next()) {
+          boolean isEmpty = rs.getInt(1) == 0;
+          attributeMap.put("queue.empty", isEmpty);
+          span.addEvent("Checked queue emptiness", attributeMap);
+          return isEmpty;
+        }
+      } catch (SQLException e) {
+        span.setAttribute("error", true);
+        span.setStatus(Status.CANCELLED);
+        EXCEPTION.accept(attributeMap, e);
+        attributeMap.put(
+            AttributeKey.EXCEPTION_MESSAGE.getKey(),
+            "Unable to check if queue is empty: " + e.getMessage());
+        span.addEvent(AttributeKey.EXCEPTION_EVENT.getKey(), attributeMap);
+        LOG.log(Level.SEVERE, "Failed to check if queue is empty", e);
+      }
+    }
+    return false;
+  }
+
+  @Override
+  public HttpResponse addToQueue(SessionRequest request) {
+    Require.nonNull("SessionRequest to add", request);
+
+    try (Span span =
+        tracer
+            .getCurrentContext()
+            .createSpan(
+                "INSERT into session_queue (request_id, payload, enqueue_time) values (?, ?, ?)")) {
+      AttributeMap attributeMap = tracer.createAttributeMap();
+      setCommonSpanAttributes(span);
+      setCommonEventAttributes(attributeMap);
+
+      String sql =
+          "INSERT INTO " + TABLE_NAME + " (request_id, payload, enqueue_time) VALUES (?, ?, ?)";
+      try (PreparedStatement stmt = connection.prepareStatement(sql)) {
+        stmt.setString(1, request.getRequestId().toString());
+        stmt.setString(2, JSON.toJson(request));
+        stmt.setTimestamp(3, Timestamp.from(request.getEnqueued()));
+
+        String statementStr = stmt.toString();
+        span.setAttribute(DATABASE_STATEMENT, statementStr);
+        span.setAttribute(DATABASE_OPERATION, "insert");
+        attributeMap.put(DATABASE_STATEMENT, statementStr);
+        attributeMap.put(DATABASE_OPERATION, "insert");
+
+        int rowCount = stmt.executeUpdate();
+        attributeMap.put("rows.added", rowCount);
+        span.addEvent("Inserted into the database", attributeMap);
+
+        HttpResponse resp = new HttpResponse();
+        resp.setStatus(rowCount > 0 ? 200 : 500);
+        return resp;
+      } catch (SQLException e) {
+        span.setAttribute("error", true);
+        span.setStatus(Status.CANCELLED);
+        EXCEPTION.accept(attributeMap, e);
+        attributeMap.put(
+            AttributeKey.EXCEPTION_MESSAGE.getKey(),
+            "Unable to add session request to the database: " + e.getMessage());
+        span.addEvent(AttributeKey.EXCEPTION_EVENT.getKey(), attributeMap);
+
+        HttpResponse resp = new HttpResponse();
+        resp.setStatus(500);
+        return resp;
+      }
+    }
+  }
+
+  @Override
+  public boolean retryAddToQueue(SessionRequest request) {
+    HttpResponse response = addToQueue(request);
+    return response.getStatus() == 200;
+  }
+
+  @Override
+  public Optional<SessionRequest> remove(RequestId requestId) {
+    Require.nonNull("RequestId to remove", requestId);
+
+    try (Span span =
+        tracer
+            .getCurrentContext()
+            .createSpan("SELECT and DELETE session request from session_queue")) {
+      AttributeMap attributeMap = tracer.createAttributeMap();
+      setCommonSpanAttributes(span);
+      setCommonEventAttributes(attributeMap);
+
+      String select = "SELECT payload FROM " + TABLE_NAME + " WHERE request_id = ?";
+      String delete = "DELETE FROM " + TABLE_NAME + " WHERE request_id = ?";
+
+      try (PreparedStatement selectStmt = connection.prepareStatement(select)) {
+        selectStmt.setString(1, requestId.toString());
+
+        String statementStr = selectStmt.toString();
+        span.setAttribute(DATABASE_STATEMENT, statementStr);
+        span.setAttribute(DATABASE_OPERATION, "select");
+        attributeMap.put(DATABASE_STATEMENT, statementStr);
+        attributeMap.put(DATABASE_OPERATION, "select");
+
+        ResultSet rs = selectStmt.executeQuery();
+        if (rs.next()) {
+          String payload = rs.getString("payload");
+
+          try (PreparedStatement deleteStmt = connection.prepareStatement(delete)) {
+            deleteStmt.setString(1, requestId.toString());
+
+            String deleteStatementStr = deleteStmt.toString();
+            span.setAttribute(DATABASE_STATEMENT, deleteStatementStr);
+            span.setAttribute(DATABASE_OPERATION, "delete");
+            attributeMap.put(DATABASE_STATEMENT, deleteStatementStr);
+            attributeMap.put(DATABASE_OPERATION, "delete");
+
+            int rowCount = deleteStmt.executeUpdate();
+            attributeMap.put("rows.deleted", rowCount);
+            span.addEvent("Removed session request from queue", attributeMap);
+
+            return Optional.of(JSON.toType(payload, SessionRequest.class));
+          }
+        } else {
+          attributeMap.put("request.found", false);
+          span.addEvent("Session request not found in queue", attributeMap);
+        }
+      } catch (SQLException e) {
+        span.setAttribute("error", true);
+        span.setStatus(Status.CANCELLED);
+        EXCEPTION.accept(attributeMap, e);
+        attributeMap.put(
+            AttributeKey.EXCEPTION_MESSAGE.getKey(),
+            "Unable to remove session request from queue: " + e.getMessage());
+        span.addEvent(AttributeKey.EXCEPTION_EVENT.getKey(), attributeMap);
+        LOG.log(Level.SEVERE, "Failed to remove session request from queue", e);
+      }
+    }
+    return Optional.empty();
+  }
+
+  @Override
+  public List<SessionRequest> getNextAvailable(Map<Capabilities, Long> stereotypes) {
+    try (Span span =
+        tracer
+            .getCurrentContext()
+            .createSpan("SELECT next available session request from session_queue")) {
+      AttributeMap attributeMap = tracer.createAttributeMap();
+      setCommonSpanAttributes(span);
+      setCommonEventAttributes(attributeMap);
+
+      String sql = "SELECT payload FROM " + TABLE_NAME + " ORDER BY enqueue_time ASC LIMIT 1";
+      try (PreparedStatement stmt = connection.prepareStatement(sql)) {
+        String statementStr = stmt.toString();
+        span.setAttribute(DATABASE_STATEMENT, statementStr);
+        span.setAttribute(DATABASE_OPERATION, "select");
+        attributeMap.put(DATABASE_STATEMENT, statementStr);
+        attributeMap.put(DATABASE_OPERATION, "select");
+
+        ResultSet rs = stmt.executeQuery();
+        if (rs.next()) {
+          String payload = rs.getString("payload");
+          SessionRequest request = JSON.toType(payload, SessionRequest.class);
+          attributeMap.put("requests.found", 1);
+          span.addEvent("Retrieved next available session request", attributeMap);
+          return List.of(request);
+        } else {
+          attributeMap.put("requests.found", 0);
+          span.addEvent("No session requests available", attributeMap);
+        }
+      } catch (SQLException e) {
+        span.setAttribute("error", true);
+        span.setStatus(Status.CANCELLED);
+        EXCEPTION.accept(attributeMap, e);
+        attributeMap.put(
+            AttributeKey.EXCEPTION_MESSAGE.getKey(),
+            "Unable to get next available session request: " + e.getMessage());
+        span.addEvent(AttributeKey.EXCEPTION_EVENT.getKey(), attributeMap);
+        LOG.log(Level.SEVERE, "Failed to get next available session request", e);
+      }
+    }
+    return List.of();
+  }
+
+  @Override
+  public boolean complete(
+      RequestId reqId, Either<SessionNotCreatedException, CreateSessionResponse> result) {
+    return remove(reqId).isPresent();
+  }
+
+  @Override
+  public int clearQueue() {
+    try (Span span =
+        tracer.getCurrentContext().createSpan("DELETE all session requests from session_queue")) {
+      AttributeMap attributeMap = tracer.createAttributeMap();
+      setCommonSpanAttributes(span);
+      setCommonEventAttributes(attributeMap);
+
+      String sql = "DELETE FROM " + TABLE_NAME;
+      try (Statement stmt = connection.createStatement()) {
+        String statementStr = sql;
+        span.setAttribute(DATABASE_STATEMENT, statementStr);
+        span.setAttribute(DATABASE_OPERATION, "delete");
+        attributeMap.put(DATABASE_STATEMENT, statementStr);
+        attributeMap.put(DATABASE_OPERATION, "delete");
+
+        int rowCount = stmt.executeUpdate(sql);
+        attributeMap.put("rows.deleted", rowCount);
+        span.addEvent("Cleared all session requests from queue", attributeMap);
+        return rowCount;
+      } catch (SQLException e) {
+        span.setAttribute("error", true);
+        span.setStatus(Status.CANCELLED);
+        EXCEPTION.accept(attributeMap, e);
+        attributeMap.put(
+            AttributeKey.EXCEPTION_MESSAGE.getKey(),
+            "Unable to clear session queue: " + e.getMessage());
+        span.addEvent(AttributeKey.EXCEPTION_EVENT.getKey(), attributeMap);
+        LOG.log(Level.SEVERE, "Failed to clear session queue", e);
+        return 0;
+      }
+    }
+  }
+
+  @Override
+  public List<SessionRequestCapability> getQueueContents() {
+    try (Span span =
+        tracer.getCurrentContext().createSpan("SELECT all session requests from session_queue")) {
+      AttributeMap attributeMap = tracer.createAttributeMap();
+      setCommonSpanAttributes(span);
+      setCommonEventAttributes(attributeMap);
+
+      String sql = "SELECT request_id, payload FROM " + TABLE_NAME + " ORDER BY enqueue_time ASC";
+      List<SessionRequestCapability> contents = new ArrayList<>();
+
+      try (PreparedStatement stmt = connection.prepareStatement(sql)) {
+        String statementStr = stmt.toString();
+        span.setAttribute(DATABASE_STATEMENT, statementStr);
+        span.setAttribute(DATABASE_OPERATION, "select");
+        attributeMap.put(DATABASE_STATEMENT, statementStr);
+        attributeMap.put(DATABASE_OPERATION, "select");
+
+        ResultSet rs = stmt.executeQuery();
+        while (rs.next()) {
+          String requestIdStr = rs.getString("request_id");
+          String payload = rs.getString("payload");
+
+          try {
+            RequestId requestId = new RequestId(UUID.fromString(requestIdStr));
+            SessionRequest request = JSON.toType(payload, SessionRequest.class);
+
+            SessionRequestCapability capability =
+                new SessionRequestCapability(requestId, request.getDesiredCapabilities());
+            contents.add(capability);
+          } catch (Exception e) {
+            LOG.log(
+                Level.WARNING, "Failed to parse session request from queue: " + requestIdStr, e);
+          }
+        }
+
+        attributeMap.put("queue.contents.size", contents.size());
+        span.addEvent("Retrieved queue contents", attributeMap);
+        return contents;
+      } catch (SQLException e) {
+        span.setAttribute("error", true);
+        span.setStatus(Status.CANCELLED);
+        EXCEPTION.accept(attributeMap, e);
+        attributeMap.put(
+            AttributeKey.EXCEPTION_MESSAGE.getKey(),
+            "Unable to get queue contents: " + e.getMessage());
+        span.addEvent(AttributeKey.EXCEPTION_EVENT.getKey(), attributeMap);
+        LOG.log(Level.SEVERE, "Failed to get queue contents", e);
+      }
+    }
+    return List.of();
+  }
+
+  @Override
+  public void close() {
+    try {
+      if (!connection.isClosed()) {
+        connection.close();
+      }
+    } catch (SQLException e) {
+      LOG.log(Level.WARNING, "Failed to close JDBC connection for SessionQueue", e);
+    }
+  }
+
+  private void ensureTableExists() {
+    String sql =
+        "CREATE TABLE IF NOT EXISTS "
+            + TABLE_NAME
+            + " ("
+            + "request_id VARCHAR(64) PRIMARY KEY,"
+            + "payload CLOB NOT NULL,"
+            + "enqueue_time TIMESTAMP NOT NULL"
+            + ")";
+    try (Statement stmt = connection.createStatement()) {
+      stmt.execute(sql);
+    } catch (SQLException e) {
+      LOG.log(Level.SEVERE, "Failed to create session_queue table", e);
+    }
+  }
+
+  private void setCommonSpanAttributes(Span span) {
+    span.setAttribute("span.kind", Span.Kind.CLIENT.toString());
+    if (jdbcUser != null) {
+      span.setAttribute(DATABASE_USER, jdbcUser);
+    }
+    if (jdbcUrl != null) {
+      span.setAttribute(DATABASE_CONNECTION_STRING, jdbcUrl);
+    }
+  }
+
+  private void setCommonEventAttributes(AttributeMap attributeMap) {
+    if (jdbcUser != null) {
+      attributeMap.put(DATABASE_USER, jdbcUser);
+    }
+    if (jdbcUrl != null) {
+      attributeMap.put(DATABASE_CONNECTION_STRING, jdbcUrl);
+    }
+  }
+}
diff --git a/java/src/org/openqa/selenium/grid/sessionqueue/jdbc/JdbcException.java b/java/src/org/openqa/selenium/grid/sessionqueue/jdbc/JdbcException.java
new file mode 100644
index 0000000..cd67f8d
--- /dev/null
+++ b/java/src/org/openqa/selenium/grid/sessionqueue/jdbc/JdbcException.java
@@ -0,0 +1,38 @@
+// Licensed to the Software Freedom Conservancy (SFC) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The SFC 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.openqa.selenium.grid.sessionqueue.jdbc;
+
+import org.openqa.selenium.WebDriverException;
+
+public class JdbcException extends WebDriverException {
+  public JdbcException() {
+    super();
+  }
+
+  public JdbcException(String message) {
+    super(message);
+  }
+
+  public JdbcException(Throwable cause) {
+    super(cause);
+  }
+
+  public JdbcException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}
diff --git a/java/src/org/openqa/selenium/grid/sessionqueue/jdbc/JdbcSessionQueueFlags.java b/java/src/org/openqa/selenium/grid/sessionqueue/jdbc/JdbcSessionQueueFlags.java
new file mode 100644
index 0000000..edbbe2e
--- /dev/null
+++ b/java/src/org/openqa/selenium/grid/sessionqueue/jdbc/JdbcSessionQueueFlags.java
@@ -0,0 +1,58 @@
+// Licensed to the Software Freedom Conservancy (SFC) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The SFC 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.openqa.selenium.grid.sessionqueue.jdbc;
+
+import static org.openqa.selenium.grid.config.StandardGridRoles.SESSION_QUEUE_ROLE;
+
+import com.beust.jcommander.Parameter;
+import com.google.auto.service.AutoService;
+import java.util.Collections;
+import java.util.Set;
+import org.openqa.selenium.grid.config.ConfigValue;
+import org.openqa.selenium.grid.config.HasRoles;
+import org.openqa.selenium.grid.config.Role;
+
+@AutoService(HasRoles.class)
+public class JdbcSessionQueueFlags implements HasRoles {
+
+  @Parameter(
+      names = "--sessionqueue-jdbc-url",
+      description = "Database URL for session queue JDBC connection.")
+  @ConfigValue(
+      section = "sessionqueue",
+      name = "jdbc-url",
+      example = "\"jdbc:postgresql://localhost:5432/selenium_queue\"")
+  private String jdbcUrl;
+
+  @Parameter(
+      names = "--sessionqueue-jdbc-user",
+      description = "Username for the session queue JDBC connection")
+  @ConfigValue(section = "sessionqueue", name = "jdbc-user", example = "selenium_user")
+  private String username;
+
+  @Parameter(
+      names = "--sessionqueue-jdbc-password",
+      description = "Password for the session queue JDBC connection")
+  @ConfigValue(section = "sessionqueue", name = "jdbc-password", example = "secure_password")
+  private String password;
+
+  @Override
+  public Set<Role> getRoles() {
+    return Collections.singleton(SESSION_QUEUE_ROLE);
+  }
+}
diff --git a/java/src/org/openqa/selenium/grid/sessionqueue/jdbc/JdbcSessionQueueOptions.java b/java/src/org/openqa/selenium/grid/sessionqueue/jdbc/JdbcSessionQueueOptions.java
new file mode 100644
index 0000000..4ef782a
--- /dev/null
+++ b/java/src/org/openqa/selenium/grid/sessionqueue/jdbc/JdbcSessionQueueOptions.java
@@ -0,0 +1,68 @@
+// Licensed to the Software Freedom Conservancy (SFC) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The SFC 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.openqa.selenium.grid.sessionqueue.jdbc;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.util.NoSuchElementException;
+import org.openqa.selenium.grid.config.Config;
+import org.openqa.selenium.internal.Require;
+
+public class JdbcSessionQueueOptions {
+
+  static final String SESSION_QUEUE_SECTION = "sessionqueue";
+
+  private final String jdbcUrl;
+  private final String jdbcUser;
+  private final String jdbcPassword;
+
+  public JdbcSessionQueueOptions(Config config) {
+    Require.nonNull("Config", config);
+
+    try {
+      this.jdbcUrl = config.get(SESSION_QUEUE_SECTION, "jdbc-url").orElse("");
+      this.jdbcUser = config.get(SESSION_QUEUE_SECTION, "jdbc-user").orElse("");
+      this.jdbcPassword = config.get(SESSION_QUEUE_SECTION, "jdbc-password").orElse("");
+
+      if (jdbcUrl.isEmpty()) {
+        throw new JdbcException(
+            "Missing JDBC Url value. Add sessionqueue option value --sessionqueue-jdbc-url"
+                + " <url-value>");
+      }
+    } catch (NoSuchElementException e) {
+      throw new JdbcException(
+          "Missing sessionqueue options. Check and add all the following options \n"
+              + " --sessionqueue-jdbc-url <url> \n"
+              + " --sessionqueue-jdbc-user <user> \n"
+              + " --sessionqueue-jdbc-password <password>");
+    }
+  }
+
+  public Connection getJdbcConnection() throws SQLException {
+    return DriverManager.getConnection(jdbcUrl, jdbcUser, jdbcPassword);
+  }
+
+  public String getJdbcUrl() {
+    return jdbcUrl;
+  }
+
+  public String getJdbcUser() {
+    return jdbcUser;
+  }
+}
diff --git a/java/test/org/openqa/selenium/grid/sessionqueue/jdbc/BUILD.bazel b/java/test/org/openqa/selenium/grid/sessionqueue/jdbc/BUILD.bazel
new file mode 100644
index 0000000..f5f3702
--- /dev/null
+++ b/java/test/org/openqa/selenium/grid/sessionqueue/jdbc/BUILD.bazel
@@ -0,0 +1,19 @@
+load("@rules_jvm_external//:defs.bzl", "artifact")
+load("//java:defs.bzl", "JUNIT5_DEPS", "java_test_suite")
+
+java_test_suite(
+    name = "MediumTests",
+    size = "medium",
+    srcs = glob(["*Test.java"]),
+    deps = [
+        "//java/src/org/openqa/selenium/events/local",
+        "//java/src/org/openqa/selenium/grid/sessionqueue/jdbc",
+        "//java/src/org/openqa/selenium/remote",
+        "//java/test/org/openqa/selenium/remote/tracing:tracing-support",
+        "//java/test/org/openqa/selenium/testing:test-base",
+        artifact("io.opentelemetry:opentelemetry-api"),
+        artifact("org.junit.jupiter:junit-jupiter-api"),
+        artifact("org.assertj:assertj-core"),
+        artifact("org.hsqldb:hsqldb"),
+    ] + JUNIT5_DEPS,
+)
diff --git a/java/test/org/openqa/selenium/grid/sessionqueue/jdbc/JdbcBackedSessionQueueTest.java b/java/test/org/openqa/selenium/grid/sessionqueue/jdbc/JdbcBackedSessionQueueTest.java
new file mode 100644
index 0000000..d4b575e
--- /dev/null
+++ b/java/test/org/openqa/selenium/grid/sessionqueue/jdbc/JdbcBackedSessionQueueTest.java
@@ -0,0 +1,416 @@
+// Licensed to the Software Freedom Conservancy (SFC) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The SFC 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.openqa.selenium.grid.sessionqueue.jdbc;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.openqa.selenium.remote.http.HttpMethod.POST;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.openqa.selenium.SessionNotCreatedException;
+import org.openqa.selenium.grid.config.Config;
+import org.openqa.selenium.grid.config.ConfigException;
+import org.openqa.selenium.grid.config.MapConfig;
+import org.openqa.selenium.grid.data.RequestId;
+import org.openqa.selenium.grid.data.SessionRequest;
+import org.openqa.selenium.grid.data.SessionRequestCapability;
+import org.openqa.selenium.grid.security.Secret;
+import org.openqa.selenium.grid.sessionqueue.NewSessionQueue;
+import org.openqa.selenium.internal.Either;
+import org.openqa.selenium.remote.http.Contents;
+import org.openqa.selenium.remote.http.HttpRequest;
+import org.openqa.selenium.remote.http.HttpResponse;
+import org.openqa.selenium.remote.tracing.DefaultTestTracer;
+import org.openqa.selenium.remote.tracing.Tracer;
+
+class JdbcBackedSessionQueueTest {
+  private static Connection connection;
+  private static final Tracer tracer = DefaultTestTracer.createTracer();
+  private static final Secret secret = new Secret("test-secret");
+
+  @BeforeAll
+  public static void createDB() throws SQLException {
+    connection = DriverManager.getConnection("jdbc:hsqldb:mem:sessionqueue", "SA", "");
+    Statement createStatement = connection.createStatement();
+    createStatement.executeUpdate(
+        "CREATE TABLE session_queue (request_id VARCHAR(64) PRIMARY KEY, payload CLOB NOT NULL,"
+            + " enqueue_time TIMESTAMP NOT NULL)");
+  }
+
+  @AfterAll
+  public static void killDBConnection() throws SQLException {
+    connection.close();
+  }
+
+  @Test
+  void shouldThrowIllegalArgumentExceptionIfConnectionObjectIsNull() {
+    assertThatThrownBy(() -> new JdbcBackedSessionQueue(tracer, secret, null))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  void canAddAndRemoveSessionRequest() {
+    JdbcBackedSessionQueue queue = getSessionQueue();
+    RequestId requestId = new RequestId(UUID.randomUUID());
+
+    // Create a proper HttpRequest for SessionRequest constructor
+    HttpRequest httpRequest = new HttpRequest(POST, "/session");
+    httpRequest.setContent(Contents.utf8String("{\"capabilities\":{\"browserName\":\"chrome\"}}"));
+
+    SessionRequest request = new SessionRequest(requestId, httpRequest, Instant.now());
+
+    HttpResponse response = queue.addToQueue(request);
+    assertThat(response.getStatus()).isEqualTo(200);
+
+    Optional<SessionRequest> removed = queue.remove(requestId);
+    assertThat(removed).isPresent();
+    assertThat(removed.get().getRequestId()).isEqualTo(requestId);
+  }
+
+  @Test
+  void getNextAvailableReturnsOldest() {
+    JdbcBackedSessionQueue queue = getSessionQueue();
+    queue.clearQueue();
+
+    RequestId requestId1 = new RequestId(UUID.randomUUID());
+    HttpRequest httpRequest1 = new HttpRequest(POST, "/session");
+    httpRequest1.setContent(
+        Contents.utf8String("{\"capabilities\":{\"browserName\":\"firefox\"}}"));
+    SessionRequest request1 = new SessionRequest(requestId1, httpRequest1, Instant.now());
+
+    RequestId requestId2 = new RequestId(UUID.randomUUID());
+    HttpRequest httpRequest2 = new HttpRequest(POST, "/session");
+    httpRequest2.setContent(Contents.utf8String("{\"capabilities\":{\"browserName\":\"chrome\"}}"));
+    SessionRequest request2 =
+        new SessionRequest(requestId2, httpRequest2, Instant.now().plusSeconds(1));
+
+    queue.addToQueue(request1);
+    queue.addToQueue(request2);
+
+    // Use getNextAvailable instead of getNextMatchingRequest
+    var next = queue.getNextAvailable(Map.of());
+    assertThat(next).isNotEmpty();
+    assertThat(next.get(0).getRequestId()).isEqualTo(requestId1);
+  }
+
+  @Test
+  void clearRemovesAllRequests() {
+    JdbcBackedSessionQueue queue = getSessionQueue();
+    queue.clearQueue();
+
+    RequestId requestId = new RequestId(UUID.randomUUID());
+    HttpRequest httpRequest = new HttpRequest(POST, "/session");
+    httpRequest.setContent(Contents.utf8String("{\"capabilities\":{\"browserName\":\"chrome\"}}"));
+    SessionRequest request = new SessionRequest(requestId, httpRequest, Instant.now());
+
+    queue.addToQueue(request);
+    queue.clearQueue();
+
+    var next = queue.getNextAvailable(Map.of());
+    assertThat(next).isEmpty();
+  }
+
+  @Test
+  void getQueueContentsReturnsAllRequests() {
+    JdbcBackedSessionQueue queue = getSessionQueue();
+    queue.clearQueue();
+
+    // Add multiple requests
+    RequestId requestId1 = new RequestId(UUID.randomUUID());
+    HttpRequest httpRequest1 = new HttpRequest(POST, "/session");
+    httpRequest1.setContent(
+        Contents.utf8String("{\"capabilities\":{\"firstMatch\":[{\"browserName\":\"firefox\"}]}}"));
+    SessionRequest request1 = new SessionRequest(requestId1, httpRequest1, Instant.now());
+
+    RequestId requestId2 = new RequestId(UUID.randomUUID());
+    HttpRequest httpRequest2 = new HttpRequest(POST, "/session");
+    httpRequest2.setContent(
+        Contents.utf8String("{\"capabilities\":{\"firstMatch\":[{\"browserName\":\"chrome\"}]}}"));
+    SessionRequest request2 =
+        new SessionRequest(requestId2, httpRequest2, Instant.now().plusSeconds(1));
+
+    queue.addToQueue(request1);
+    queue.addToQueue(request2);
+
+    // Get queue contents
+    var contents = queue.getQueueContents();
+    assertThat(contents).hasSize(2);
+
+    // Verify first request (oldest)
+    SessionRequestCapability first = contents.get(0);
+    assertThat(first.getRequestId()).isEqualTo(requestId1);
+    assertThat(first.getDesiredCapabilities()).isNotNull();
+
+    // Verify second request
+    SessionRequestCapability second = contents.get(1);
+    assertThat(second.getRequestId()).isEqualTo(requestId2);
+    assertThat(second.getDesiredCapabilities()).isNotNull();
+  }
+
+  @Test
+  void getQueueContentsReturnsEmptyListWhenQueueIsEmpty() {
+    JdbcBackedSessionQueue queue = getSessionQueue();
+    queue.clearQueue();
+
+    var contents = queue.getQueueContents();
+    assertThat(contents).isEmpty();
+  }
+
+  @Test
+  void peekEmptyReturnsTrueWhenQueueIsEmpty() {
+    JdbcBackedSessionQueue queue = getSessionQueue();
+    queue.clearQueue();
+
+    assertThat(queue.peekEmpty()).isTrue();
+  }
+
+  @Test
+  void peekEmptyReturnsFalseWhenQueueHasRequests() {
+    JdbcBackedSessionQueue queue = getSessionQueue();
+    queue.clearQueue();
+
+    RequestId requestId = new RequestId(UUID.randomUUID());
+    HttpRequest httpRequest = new HttpRequest(POST, "/session");
+    httpRequest.setContent(Contents.utf8String("{\"capabilities\":{\"browserName\":\"chrome\"}}"));
+    SessionRequest request = new SessionRequest(requestId, httpRequest, Instant.now());
+
+    queue.addToQueue(request);
+    assertThat(queue.peekEmpty()).isFalse();
+  }
+
+  @Test
+  void isReadyReturnsTrueWhenConnectionIsOpen() {
+    JdbcBackedSessionQueue queue = getSessionQueue();
+    assertThat(queue.isReady()).isTrue();
+  }
+
+  @Test
+  void removeReturnsEmptyWhenRequestNotFound() {
+    JdbcBackedSessionQueue queue = getSessionQueue();
+    queue.clearQueue();
+
+    RequestId nonExistentId = new RequestId(UUID.randomUUID());
+    Optional<SessionRequest> removed = queue.remove(nonExistentId);
+    assertThat(removed).isEmpty();
+  }
+
+  @Test
+  void retryAddToQueueDelegatesToAddToQueue() {
+    JdbcBackedSessionQueue queue = getSessionQueue();
+    RequestId requestId = new RequestId(UUID.randomUUID());
+
+    HttpRequest httpRequest = new HttpRequest(POST, "/session");
+    httpRequest.setContent(Contents.utf8String("{\"capabilities\":{\"browserName\":\"chrome\"}}"));
+    SessionRequest request = new SessionRequest(requestId, httpRequest, Instant.now());
+
+    boolean result = queue.retryAddToQueue(request);
+    assertThat(result).isTrue();
+
+    // Verify it was actually added
+    Optional<SessionRequest> removed = queue.remove(requestId);
+    assertThat(removed).isPresent();
+  }
+
+  @Test
+  void completeDoesNotThrowException() {
+    JdbcBackedSessionQueue queue = getSessionQueue();
+    RequestId requestId = new RequestId(UUID.randomUUID());
+
+    // complete() method should not throw any exception - requires Either parameter
+    assertThatCode(
+            () -> queue.complete(requestId, Either.left(new SessionNotCreatedException("test"))))
+        .doesNotThrowAnyException();
+  }
+
+  @Test
+  void getNextAvailableReturnsEmptyWhenQueueIsEmpty() {
+    JdbcBackedSessionQueue queue = getSessionQueue();
+    queue.clearQueue();
+
+    var next = queue.getNextAvailable(Map.of());
+    assertThat(next).isEmpty();
+  }
+
+  @Test
+  void clearQueueReturnsCorrectRowCount() {
+    JdbcBackedSessionQueue queue = getSessionQueue();
+    queue.clearQueue(); // Start with empty queue
+
+    // Add multiple requests
+    RequestId requestId1 = new RequestId(UUID.randomUUID());
+    HttpRequest httpRequest1 = new HttpRequest(POST, "/session");
+    httpRequest1.setContent(Contents.utf8String("{\"capabilities\":{\"browserName\":\"chrome\"}}"));
+    SessionRequest request1 = new SessionRequest(requestId1, httpRequest1, Instant.now());
+
+    RequestId requestId2 = new RequestId(UUID.randomUUID());
+    HttpRequest httpRequest2 = new HttpRequest(POST, "/session");
+    httpRequest2.setContent(
+        Contents.utf8String("{\"capabilities\":{\"browserName\":\"firefox\"}}"));
+    SessionRequest request2 = new SessionRequest(requestId2, httpRequest2, Instant.now());
+
+    queue.addToQueue(request1);
+    queue.addToQueue(request2);
+
+    // Clear queue and verify row count
+    int deletedRows = queue.clearQueue();
+    assertThat(deletedRows).isEqualTo(2);
+
+    // Verify queue is actually empty
+    assertThat(queue.peekEmpty()).isTrue();
+  }
+
+  @Test
+  void addToQueueHandlesDuplicateRequestIds() {
+    JdbcBackedSessionQueue queue = getSessionQueue();
+    RequestId requestId = new RequestId(UUID.randomUUID());
+
+    HttpRequest httpRequest1 = new HttpRequest(POST, "/session");
+    httpRequest1.setContent(Contents.utf8String("{\"capabilities\":{\"browserName\":\"chrome\"}}"));
+    SessionRequest request1 = new SessionRequest(requestId, httpRequest1, Instant.now());
+
+    HttpRequest httpRequest2 = new HttpRequest(POST, "/session");
+    httpRequest2.setContent(
+        Contents.utf8String("{\"capabilities\":{\"browserName\":\"firefox\"}}"));
+    SessionRequest request2 = new SessionRequest(requestId, httpRequest2, Instant.now());
+
+    // First add should succeed
+    HttpResponse response1 = queue.addToQueue(request1);
+    assertThat(response1.getStatus()).isEqualTo(200);
+
+    // Second add with same ID should fail due to primary key constraint
+    HttpResponse response2 = queue.addToQueue(request2);
+    assertThat(response2.getStatus()).isEqualTo(500);
+  }
+
+  @Test
+  void getQueueContentsHandlesLargeQueue() {
+    JdbcBackedSessionQueue queue = getSessionQueue();
+    queue.clearQueue();
+
+    // Add multiple requests to test ordering
+    int numRequests = 5;
+    RequestId[] requestIds = new RequestId[numRequests];
+
+    for (int i = 0; i < numRequests; i++) {
+      requestIds[i] = new RequestId(UUID.randomUUID());
+      HttpRequest httpRequest = new HttpRequest(POST, "/session");
+      httpRequest.setContent(
+          Contents.utf8String("{\"capabilities\":{\"browserName\":\"browser" + i + "\"}}"));
+      SessionRequest request =
+          new SessionRequest(requestIds[i], httpRequest, Instant.now().plusSeconds(i));
+      queue.addToQueue(request);
+    }
+
+    var contents = queue.getQueueContents();
+    assertThat(contents).hasSize(numRequests);
+
+    // Verify ordering (oldest first)
+    for (int i = 0; i < numRequests; i++) {
+      assertThat(contents.get(i).getRequestId()).isEqualTo(requestIds[i]);
+    }
+  }
+
+  @Test
+  void closeConnectionDoesNotThrowException() {
+    JdbcBackedSessionQueue queue = getSessionQueue();
+
+    // close() method should not throw any exception
+    assertThatCode(() -> queue.close()).doesNotThrowAnyException();
+  }
+
+  @Test
+  void createWithValidConfigReturnsNewSessionQueue() {
+    // Create a config with JDBC settings
+    Map<String, Object> configMap =
+        Map.of(
+            "sessionqueue",
+                Map.of(
+                    "jdbc-url", "jdbc:hsqldb:mem:testqueue",
+                    "jdbc-user", "SA",
+                    "jdbc-password", ""),
+            "logging", Map.of("tracing", false),
+            "server", Map.of("registration-secret", "test-secret"));
+
+    Config config = new MapConfig(configMap);
+
+    // Test that create method returns a NewSessionQueue instance
+    NewSessionQueue queue = JdbcBackedSessionQueue.create(config);
+
+    assertThat(queue).isNotNull();
+    assertThat(queue).isInstanceOf(JdbcBackedSessionQueue.class);
+    assertThat(queue.isReady()).isTrue();
+
+    // Test basic functionality
+    assertThat(queue.peekEmpty()).isTrue();
+
+    // Clean up
+    if (queue instanceof JdbcBackedSessionQueue) {
+      ((JdbcBackedSessionQueue) queue).close();
+    }
+  }
+
+  @Test
+  void createWithInvalidJdbcUrlThrowsConfigException() {
+    // Create a config with invalid JDBC URL
+    Map<String, Object> configMap =
+        Map.of(
+            "sessionqueue",
+                Map.of(
+                    "jdbc-url", "invalid:jdbc:url",
+                    "jdbc-user", "SA",
+                    "jdbc-password", ""),
+            "logging", Map.of("tracing", false),
+            "server", Map.of("registration-secret", "test-secret"));
+
+    Config config = new MapConfig(configMap);
+
+    // Test that create method throws ConfigException for invalid JDBC URL
+    assertThatThrownBy(() -> JdbcBackedSessionQueue.create(config))
+        .isInstanceOf(ConfigException.class)
+        .hasCauseInstanceOf(SQLException.class);
+  }
+
+  @Test
+  void createWithMissingConfigThrowsException() {
+    // Create a config missing required JDBC settings
+    Map<String, Object> configMap =
+        Map.of(
+            "logging", Map.of("tracing", false),
+            "server", Map.of("registration-secret", "test-secret"));
+
+    Config config = new MapConfig(configMap);
+
+    // Test that create method throws JdbcException for missing config
+    assertThatThrownBy(() -> JdbcBackedSessionQueue.create(config))
+        .isInstanceOf(JdbcException.class)
+        .hasMessageContaining("Missing JDBC Url value");
+  }
+
+  private JdbcBackedSessionQueue getSessionQueue() {
+    return new JdbcBackedSessionQueue(tracer, secret, connection);
+  }
+}