feat: DH-18888 Support Connector Plugin (#396)

Author: stanbrub

pom.xml

@@ -78,8 +78,8 @@
 				<artifactId>maven-compiler-plugin</artifactId>
 				<version>3.13.0</version>
 				<configuration>
-					<source>21</source>
-					<target>21</target>
+					<source>17</source>
+					<target>17</target>
 				</configuration>
 			</plugin>
 			<plugin>
@@ -146,7 +146,7 @@
 			<plugin>
 				<groupId>org.apache.maven.plugins</groupId>
 				<artifactId>maven-javadoc-plugin</artifactId>
-				<version>3.10.1</version>
+				<version>3.11.2</version>
 				<configuration>
 					<excludePackageNames>*.connect,*.controller,*.generator,*.jfr,*.metric,*.run,*.util</excludePackageNames>
 					<show>public</show>
@@ -167,7 +167,7 @@
 				'core.autocrlf' to true -->
 				<groupId>com.diffplug.spotless</groupId>
 				<artifactId>spotless-maven-plugin</artifactId>
-				<version>2.43.0</version>
+				<version>2.44.2</version>
 				<configuration>
 					<java>
 						<includes>
@@ -193,7 +193,7 @@
 			<plugin>
 				<groupId>org.apache.maven.plugins</groupId>
 				<artifactId>maven-surefire-plugin</artifactId>
-				<version>3.5.1</version>
+				<version>3.5.2</version>
 				<configuration>
 					<excludes>
 						<exclude>/io/deephaven/benchmark/tests/**/*Test</exclude>
@@ -203,14 +203,14 @@
 					<dependency>
 						<groupId>org.junit.jupiter</groupId>
 						<artifactId>junit-jupiter-engine</artifactId>
-						<version>5.9.2</version>
+						<version>5.9.3</version>
 					</dependency>
 				</dependencies>
 			</plugin>
 			<plugin>
 				<groupId>org.apache.maven.plugins</groupId>
 				<artifactId>maven-failsafe-plugin</artifactId>
-				<version>3.5.1</version>
+				<version>3.5.2</version>
 				<configuration>
 					<forkCount>0</forkCount>
 					<includes>
@@ -221,7 +221,7 @@
 					<dependency>
 						<groupId>org.junit.jupiter</groupId>
 						<artifactId>junit-jupiter-engine</artifactId>
-						<version>5.9.2</version>
+						<version>5.9.3</version>
 					</dependency>
 				</dependencies>
 				<executions>
@@ -247,39 +247,39 @@
 		<dependency>
 			<groupId>io.netty</groupId>
 			<artifactId>netty-all</artifactId>
-			<version>4.1.100.Final</version>
+			<version>4.1.114.Final</version>
 		</dependency>
 		<!-- Added because of conflicts -->
 		<dependency>
 			<groupId>io.grpc</groupId>
 			<artifactId>grpc-all</artifactId>
-			<version>1.58.0</version>
+			<version>1.65.1</version>
 		</dependency>
 		<!-- Added because of conflicts -->
 		<dependency>
 			<groupId>com.google.protobuf</groupId>
 			<artifactId>protobuf-java</artifactId>
-			<version>3.25.3</version>
+			<version>3.25.4</version>
 		</dependency>
 		<dependency>
 			<groupId>io.confluent</groupId>
 			<artifactId>kafka-avro-serializer</artifactId>
-			<version>7.7.1</version>
+			<version>7.9.0</version>
 		</dependency>
 		<dependency>
 			<groupId>io.confluent</groupId>
 			<artifactId>kafka-protobuf-serializer</artifactId>
-			<version>7.7.1</version>
+			<version>7.9.0</version>
 		</dependency>
 		<dependency>
 			<groupId>io.deephaven</groupId>
 			<artifactId>deephaven-java-client-barrage-dagger</artifactId>
-			<version>0.36.1</version>
+			<version>0.37.6</version>
 		</dependency>
 		<dependency>
 			<groupId>io.deephaven</groupId>
 			<artifactId>deephaven-log-to-slf4j</artifactId>
-			<version>0.36.1</version>
+			<version>0.37.6</version>
 		</dependency>
 		<dependency>
 			<groupId>org.junit.platform</groupId>

src/main/java/io/deephaven/benchmark/api/BenchQuery.java

@@ -1,13 +1,15 @@
-/* Copyright (c) 2022-2023 Deephaven Data Labs and Patent Pending */
+/* Copyright (c) 2022-2025 Deephaven Data Labs and Patent Pending */
 package io.deephaven.benchmark.api;
 
 import java.io.Closeable;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.Properties;
 import java.util.concurrent.Future;
 import java.util.function.Consumer;
 import java.util.function.Function;
-import io.deephaven.benchmark.connect.BarrageConnector;
+import io.deephaven.benchmark.connect.Connector;
+import io.deephaven.benchmark.connect.ConnectorFactory;
 import io.deephaven.benchmark.connect.ResultTable;
 import io.deephaven.benchmark.metric.Metrics;
 import io.deephaven.benchmark.util.Timer;
@@ -23,7 +25,8 @@ final public class BenchQuery implements Closeable {
     final QueryLog queryLog;
     final Map<String, Consumer<ResultTable>> snapshotFetchers = new LinkedHashMap<>();
     final Map<String, Function<ResultTable, Boolean>> tickingFetchers = new LinkedHashMap<>();
-    private BarrageConnector session = null;
+    final Properties props = new Properties();
+    private Connector session = null;
 
     BenchQuery(Bench bench, String logic, QueryLog queryLog) {
         this.bench = bench;
@@ -57,6 +60,16 @@ public BenchQuery fetchDuring(String table, Function<ResultTable, Boolean> table
         return this;
     }
 
+    /**
+     * Add properties to be passed to the <code>Connector</code> used in the query
+     */
+    public BenchQuery withProperty(String name, String value) {
+        if (value != null && !value.isBlank()) {
+            props.setProperty(name, value);
+        }
+        return this;
+    }
+
     /**
      * Execute the query logic through a session
      */
@@ -96,8 +109,10 @@ public void close() {
     // Add function defs in separate query so if there are errors in the "logic" part, the line numbers match up
     private void executeBarrageQuery(String logic) {
         if (session == null) {
-            String deephavenServer = bench.property("deephaven.addr", "localhost:10000");
-            session = new BarrageConnector(deephavenServer);
+            var connectorClass = bench.property("connector.class", "io.deephaven.benchmark.connect.BarrageConnector");
+            var localProps = Bench.profile.getProperties();
+            localProps.putAll(props);
+            session = ConnectorFactory.create(connectorClass, localProps);
         }
         String snippetsLogic = Bench.profile.replaceProperties(Snippets.getFunctions(logic));
         if (!snippetsLogic.isBlank()) {

src/main/java/io/deephaven/benchmark/api/Profile.java

@@ -1,4 +1,4 @@
-/* Copyright (c) 2022-2024 Deephaven Data Labs and Patent Pending */
+/* Copyright (c) 2022-2025 Deephaven Data Labs and Patent Pending */
 package io.deephaven.benchmark.api;
 
 import java.io.InputStream;
@@ -13,19 +13,20 @@
 import io.deephaven.benchmark.util.Log;
 
 /**
- * Represents properties for a the Benchmark API .profile file in addition to allowing retrieval of System and
- * environment properties.
+ * Represents properties for the Benchmark API profile file in addition to allowing retrieval of System and environment
+ * properties. The default property file (e.g. default.properties) for the profile can be overridden with
+ * <code>System.setProperty("benchmark.profile.default", "my-properties-path")</code>.
  */
 class Profile {
     final private URL url;
     final private Properties props;
 
     /**
-     * Initialize the profile from the supplied benchmark.profile property. That that property is absent, use the
-     * default properties file.
+     * Initialize the profile from the supplied "benchmark.profile" property. If that property is absent, use the
+     * "benchmark.profile.default" property. If that property is absent, use the "default.properties" resource file.
      */
     Profile() {
-        this(System.getProperty("benchmark.profile", "default.properties"));
+        this(System.getProperty("benchmark.profile", getProfileDefaultFile()));
     }
 
     /**
@@ -218,4 +219,8 @@ private URL findProfileAsResource(Class<?> parentClass, String uri) {
         }
     }
 
+    static String getProfileDefaultFile() {
+        return System.getProperty("benchmark.profile.default", "default.properties");
+    }
+
 }

src/main/java/io/deephaven/benchmark/connect/BarrageConnector.java

@@ -1,4 +1,4 @@
-/* Copyright (c) 2022-2023 Deephaven Data Labs and Patent Pending */
+/* Copyright (c) 2022-2025 Deephaven Data Labs and Patent Pending */
 package io.deephaven.benchmark.connect;
 
 import java.util.*;
@@ -30,7 +30,7 @@
  * The typical workflow will be initialize connection, execute query, fetch results, close. Note: This class is meant to
  * be used through the Bench api rather than directly.
  */
-public class BarrageConnector implements AutoCloseable {
+class BarrageConnector implements Connector {
     static {
         System.setProperty("thread.initialization", ""); // Remove server side initializers (e.g. DebuggingInitializer)
     }
@@ -51,15 +51,17 @@ public class BarrageConnector implements AutoCloseable {
      * 
      * @param hostPort a host and port string for connecting to a Deephaven worker (ex. localhost:10000)
      */
-    public BarrageConnector(String hostPort) {
-        String[] split;
+    BarrageConnector(Properties props) {
+        var hostPort = props.getProperty("deephaven.addr", "localhost:10000");
+        var userPass = props.getProperty("deephaven.auth", "");
+        var host = hostPort.replaceAll(":.*", "");
+        var port = hostPort.replaceAll(".*:", "");
+        if (host.isEmpty() || port.isEmpty())
+            throw new RuntimeException("Missing Connector host or port");
+        if (!userPass.isBlank())
+            System.out.println("Ignoring supplied User and Pass");
         try {
-            split = hostPort.split(":");
-        } catch (Exception ex) {
-            throw new RuntimeException("Failed to parse connection hostPort: " + hostPort, ex);
-        }
-        try {
-            this.channel = getManagedChannel(split[0], Integer.parseInt(split[1]));
+            this.channel = getManagedChannel(host, Integer.parseInt(port));
             this.session = getSession(channel);
             this.console = session.session().console("python").get();
         } catch (Exception ex) {

src/main/java/io/deephaven/benchmark/connect/Connector.java

@@ -0,0 +1,56 @@
+/* Copyright (c) 2022-2025 Deephaven Data Labs and Patent Pending */
+package io.deephaven.benchmark.connect;
+
+import java.util.Set;
+import java.util.concurrent.Future;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import io.deephaven.benchmark.metric.Metrics;
+
+/**
+ * An Object that provides a connection to a service under test, execution of statements, and retrieval of results
+ * through snapshots or live data. What this all means is up to the implementer, but it's execution is supported by
+ * Benchmark API in {@link io.deephaven.benchmark.api.BenchQuery}.
+ */
+public interface Connector extends AutoCloseable {
+
+    /**
+     * Execute a query or statement against the connected service
+     * 
+     * @param query the statement to be executed
+     */
+    public void executeQuery(String query);
+
+    /**
+     * Get a list a variable or table names used in the query. This is optional and may return an empty set.
+     * 
+     * @return a set of variable names or empty set
+     */
+    public Set<String> getUsedVariableNames();
+
+    /**
+     * Fetch data as a snapshot of the given table and return a result table according to the given consumer. This fetch
+     * is made after the execution of the query.
+     * 
+     * @param table the table to fetch data from
+     * @param tableHandler a consumer to supplied to receive the fetched data
+     * @return a future that waits until the fetch is complete
+     */
+    public Future<Metrics> fetchSnapshotData(String table, Consumer<ResultTable> tableHandler);
+
+    /**
+     * Fetch data for the given table as it is updated while the query is running. The given function is called at a
+     * refresh rate determined by the <code>Connector</code> implementation.
+     * 
+     * @param table the table to fetch data from
+     * @param tableHandler a function to call on a cycle
+     * @return a future that waits until the fetch initiates
+     */
+    public Future<Metrics> fetchTickingData(String table, Function<ResultTable, Boolean> tableHandler);
+
+    /**
+     * Close the connector and clean up resources
+     */
+    public void close();
+
+}

src/main/java/io/deephaven/benchmark/connect/ConnectorFactory.java

@@ -0,0 +1,20 @@
+/* Copyright (c) 2022-2025 Deephaven Data Labs and Patent Pending */
+package io.deephaven.benchmark.connect;
+
+import java.util.Properties;
+
+/**
+ * Instantiate the <code>Connector</code> for the given full-qualified class name with the given properties.
+ */
+public class ConnectorFactory {
+    static public Connector create(String connectorClassName, Properties props) {
+        try {
+            var myClass = Class.forName(connectorClassName);
+            var constructor = myClass.getDeclaredConstructor(Properties.class);
+            return (Connector) constructor.newInstance(props);
+        } catch (Exception ex) {
+            throw new RuntimeException("Failed to instantiate Connector: " + connectorClassName, ex);
+        }
+    }
+
+}

src/main/java/io/deephaven/benchmark/util/Numbers.java

@@ -88,18 +88,21 @@ static public String formatBytesToGigs(Object val) {
      * @return the negated number or null if <code>val</code> was null
      */
     static public Number negate(Object val) {
-        if (val instanceof Number && ((Number) val).doubleValue() == 0.0)
+        if (val == null || (val instanceof Number && ((Number) val).doubleValue() == 0.0))
             return (Number) val;
-        return switch (val) {
-            case Integer v -> -v.intValue();
-            case Float v -> -v.floatValue();
-            case Long v -> -v.longValue();
-            case Double v -> -v.doubleValue();
-            case Short v -> (short) (-v.shortValue());
-            case Byte v -> (byte) (-v.byteValue());
-            case null -> null;
-            default -> throw new RuntimeException("Unsupported Type: " + val.getClass().getSimpleName());
-        };
+        if (val instanceof Integer)
+            return -((Integer) val).intValue();
+        if (val instanceof Float)
+            return -((Float) val).floatValue();
+        if (val instanceof Long)
+            return -((Long) val).longValue();
+        if (val instanceof Double)
+            return -((Double) val).doubleValue();
+        if (val instanceof Short)
+            return (short) -((Short) val).shortValue();
+        if (val instanceof Byte)
+            return (byte) -((Byte) val).byteValue();
+        throw new RuntimeException("Unsupported Type: " + val.getClass().getSimpleName());
     }
 
     /**
@@ -111,11 +114,11 @@ static public Number negate(Object val) {
      * @return true if an even number, otherwise false
      */
     static public boolean isEven(Object val) {
-        return switch (val) {
-            case Number v -> v.longValue() % 2 == 0;
-            case null -> false;
-            default -> throw new RuntimeException("Unsupported Type: " + val.getClass().getSimpleName());
-        };
+        if (val == null)
+            return false;
+        if (val instanceof Number)
+            return ((Number) val).longValue() % 2 == 0;
+        throw new RuntimeException("Unsupported Type: " + val.getClass().getSimpleName());
     }
 
     /**

src/main/resources/io/deephaven/benchmark/run/profile/default.properties

@@ -1,4 +1,10 @@
 
+# The fully-qualified class name of the connector used in the tests
+connector.class=
+
+# Description of the authentication to use (e.g. user:pass)
+deephaven.auth=
+
 # Deephaven engine address (same one the UI uses)
 deephaven.addr=localhost:10000
 

src/test/java/io/deephaven/benchmark/util/NumbersTest.java

@@ -2,7 +2,6 @@
 package io.deephaven.benchmark.util;
 
 import static org.junit.jupiter.api.Assertions.*;
-import java.math.BigInteger;
 import org.junit.jupiter.api.*;
 
 public class NumbersTest {