Vert.x config cleanup (#1446)

Author: ferenc-csaky

sqrl-cli/src/main/java/com/datasqrl/cli/DatasqrlRun.java

@@ -16,11 +16,8 @@
 package com.datasqrl.cli;
 
 import static com.datasqrl.env.EnvVariableNames.KAFKA_BOOTSTRAP_SERVERS;
-import static com.datasqrl.env.EnvVariableNames.POSTGRES_DATABASE;
-import static com.datasqrl.env.EnvVariableNames.POSTGRES_HOST;
 import static com.datasqrl.env.EnvVariableNames.POSTGRES_JDBC_URL;
 import static com.datasqrl.env.EnvVariableNames.POSTGRES_PASSWORD;
-import static com.datasqrl.env.EnvVariableNames.POSTGRES_PORT;
 import static com.datasqrl.env.EnvVariableNames.POSTGRES_USERNAME;
 
 import com.datasqrl.config.PackageJson;
@@ -32,15 +29,14 @@
 import com.datasqrl.graphql.config.ServerConfigUtil;
 import com.datasqrl.graphql.server.RootGraphqlModel;
 import com.datasqrl.util.ConfigLoaderUtils;
-import com.google.common.io.Resources;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import io.micrometer.prometheusmetrics.PrometheusConfig;
 import io.micrometer.prometheusmetrics.PrometheusMeterRegistry;
 import io.vertx.core.Vertx;
 import io.vertx.core.VertxOptions;
 import io.vertx.core.json.JsonObject;
 import io.vertx.micrometer.MicrometerMetricsFactory;
 import java.net.URI;
-import java.net.URL;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -83,6 +79,7 @@ public class DatasqrlRun {
   private final Configuration flinkConfig;
   private final Map<String, String> env;
   private final boolean blocking;
+  private final ObjectMapper mapper;
   @Nullable private final CountDownLatch shutdownLatch;
 
   private Vertx vertx;
@@ -99,6 +96,7 @@ private DatasqrlRun(
     this.flinkConfig = flinkConfig;
     this.env = env;
     this.blocking = blocking;
+    mapper = SqrlObjectMapper.getMapperWithEnvVarResolver(env);
     shutdownLatch = blocking ? new CountDownLatch(1) : null;
   }
 
@@ -272,52 +270,34 @@ private String getenv(String key) {
 
   @SneakyThrows
   private void startVertx() {
-    if (!planDir.resolve("vertx.json").toFile().exists()) {
+    var vertxJson = planDir.resolve("vertx.json").toFile();
+    if (!vertxJson.exists()) {
       return;
     }
-    RootGraphqlModel rootGraphqlModel =
-        SqrlObjectMapper.MAPPER.readValue(
-                planDir.resolve("vertx.json").toFile(), ModelContainer.class)
-            .model;
+
+    RootGraphqlModel rootGraphqlModel = mapper.readValue(vertxJson, ModelContainer.class).model;
     if (rootGraphqlModel == null) {
       return; // no graphql server queries
     }
 
-    URL resource = Resources.getResource("server-config.json");
-    Map<String, Object> json = SqrlObjectMapper.MAPPER.readValue(resource, Map.class);
-    JsonObject config = new JsonObject(json);
-
-    ServerConfig serverConfig =
-        new ServerConfig(config) {
-          @Override
-          public String getEnvironmentVariable(String envVar) {
-            return getenv(envVar);
-          }
-        };
-
-    // Set Postgres connection options from environment variables
-    if (planDir.resolve("postgres.json").toFile().exists()) {
-      serverConfig
-          .getPgConnectOptions()
-          .setHost(getenv(POSTGRES_HOST))
-          .setPort(Integer.parseInt(getenv(POSTGRES_PORT)))
-          .setUser(getenv(POSTGRES_USERNAME))
-          .setPassword(getenv(POSTGRES_PASSWORD))
-          .setDatabase(getenv(POSTGRES_DATABASE));
+    var vertxConfigJson = planDir.resolve("vertx-config.json").toFile();
+    if (!vertxConfigJson.exists()) {
+      throw new IllegalStateException(
+          "Server config JSON '%s' does not exist".formatted(vertxConfigJson));
     }
 
-    serverConfig = ServerConfigUtil.mergeConfigs(serverConfig, vertxConfig());
-
-    var server = new HttpServerVerticle(serverConfig, rootGraphqlModel);
+    Map<String, Object> json = mapper.readValue(vertxConfigJson, Map.class);
+    var baseServerConfig = new ServerConfig(new JsonObject(json));
 
+    var serverConfig = ServerConfigUtil.mergeConfigs(baseServerConfig, vertxConfig());
+    var serverVerticle = new HttpServerVerticle(serverConfig, rootGraphqlModel);
     var prometheusMeterRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
     var metricsOptions =
         new MicrometerMetricsFactory(prometheusMeterRegistry).newOptions().setEnabled(true);
 
     vertx = Vertx.vertx(new VertxOptions().setMetricsOptions(metricsOptions));
-
     vertx
-        .deployVerticle(server)
+        .deployVerticle(serverVerticle)
         .onComplete(
             res -> {
               if (res.succeeded()) {

sqrl-cli/src/main/resources/templates/server-config.json

@@ -108,15 +108,15 @@
     "http2RstFloodWindowDuration" : 30,
     "http2RstFloodWindowDurationTimeUnit" : "SECONDS",
     "strictThreadMode" : false,
-    "useAlpn" : false,
+    "enabledSecureTransportProtocols" : [ "TLSv1", "TLSv1.1", "TLSv1.2" ],
     "keyCertOptions" : null,
     "trustOptions" : null,
     "clientAuth" : "NONE",
-    "enabledSecureTransportProtocols" : [ "TLSv1", "TLSv1.1", "TLSv1.2" ],
     "sslHandshakeTimeout" : 10,
     "sslHandshakeTimeoutUnit" : "SECONDS",
-    "sni" : false,
-    "fileRegionEnabled" : true
+    "fileRegionEnabled" : true,
+    "useAlpn" : false,
+    "sni" : false
   },
   "pgConnectOptions" : {
     "host" : "${POSTGRES_HOST}",
@@ -184,5 +184,16 @@
     "contactEmail" : "[email protected]",
     "license" : "Apache License 2.0",
     "licenseUrl" : "https://www.apache.org/licenses/LICENSE-2.0"
+  },
+  "kafkaMutationConfig" : {
+    "value.serializer" : "com.datasqrl.graphql.kafka.JsonSerializer",
+    "bootstrap.servers" : "${KAFKA_BOOTSTRAP_SERVERS}",
+    "key.serializer" : "com.datasqrl.graphql.kafka.JsonSerializer"
+  },
+  "kafkaSubscriptionConfig" : {
+    "key.deserializer" : "com.datasqrl.graphql.kafka.JsonDeserializer",
+    "value.deserializer" : "com.datasqrl.graphql.kafka.JsonDeserializer",
+    "bootstrap.servers" : "${KAFKA_BOOTSTRAP_SERVERS}",
+    "auto.offset.reset" : "latest"
   }
 }

sqrl-planner/src/main/resources/server-config.json

@@ -19,6 +19,7 @@
 import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.databind.DeserializationContext;
 import com.fasterxml.jackson.databind.JsonDeserializer;
+import jakarta.annotation.Nullable;
 import java.io.IOException;
 import java.util.Map;
 import java.util.regex.Matcher;
@@ -27,14 +28,14 @@
 /** Custom deserializer to replace environment variables in JSON strings. */
 public class JsonEnvVarDeserializer extends JsonDeserializer<String> {
 
-  private Map<String, String> env;
+  private final Map<String, String> env;
 
   public JsonEnvVarDeserializer() {
-    env = GlobalEnvironmentStore.getAll();
+    this(null);
   }
 
-  public JsonEnvVarDeserializer(Map<String, String> env) {
-    this.env = env;
+  public JsonEnvVarDeserializer(@Nullable Map<String, String> env) {
+    this.env = env != null ? env : GlobalEnvironmentStore.getAll();
   }
 
   @Override