Fix various issues (see description) (#89)

* Expand ConfigProvider to load from a namespace

* Fix snakeyaml issue

* sqljob update

Author: jogrogan

deploy/samples/demodb.yaml

@@ -48,4 +48,3 @@ spec:
   - Modify
   connector: |
     connector = blackhole
-

deploy/sqljobs.crd.yaml

@@ -11,10 +11,11 @@ docker run \
   ghcr.io/kubernetes-client/java/crd-model-gen:v1.0.6 \
   /generate.sh -o "$(pwd)/hoptimator-k8s" -n "" -p "com.linkedin.hoptimator.k8s" \
   -u "$(pwd)/hoptimator-k8s/src/main/resources/databases.crd.yaml" \
+  -u "$(pwd)/hoptimator-k8s/src/main/resources/engines.crd.yaml" \
   -u "$(pwd)/hoptimator-k8s/src/main/resources/jobtemplates.crd.yaml" \
   -u "$(pwd)/hoptimator-k8s/src/main/resources/pipelines.crd.yaml" \
+  -u "$(pwd)/hoptimator-k8s/src/main/resources/sqljobs.crd.yaml" \
+  -u "$(pwd)/hoptimator-k8s/src/main/resources/subscriptions.crd.yaml" \
   -u "$(pwd)/hoptimator-k8s/src/main/resources/tabletemplates.crd.yaml" \
   -u "$(pwd)/hoptimator-k8s/src/main/resources/views.crd.yaml" \
-  -u "$(pwd)/hoptimator-k8s/src/main/resources/subscriptions.crd.yaml" \
-  -u "$(pwd)/hoptimator-k8s/src/main/resources/engines.crd.yaml" \
   && echo "done."

generate-models.sh

@@ -4,8 +4,11 @@ avro = "org.apache.avro:avro:1.10.2"
 calcite-avatica = "org.apache.calcite.avatica:avatica:1.23.0"
 calcite-core = "org.apache.calcite:calcite-core:1.34.0"
 calcite-server = "org.apache.calcite:calcite-server:1.34.0"
+commons-cli = "commons-cli:commons-cli:1.4"
 flink-clients = "org.apache.flink:flink-clients:1.18.1"
 flink-connector-base = "org.apache.flink:flink-connector-base:1.18.1"
+flink-connector-kafka = "org.apache.flink:flink-sql-connector-kafka:3.2.0-1.18"
+flink-connector-mysql-cdc = "com.ververica:flink-sql-connector-mysql-cdc:2.3.0"
 flink-core = "org.apache.flink:flink-core:1.18.1"
 flink-csv = "org.apache.flink:flink-csv:1.18.1"
 flink-jdbc = "org.apache.flink:flink-sql-jdbc-driver-bundle:1.18.1"
@@ -15,8 +18,6 @@ flink-table-api-java-bridge = "org.apache.flink:flink-table-api-java-bridge:1.18
 flink-table-common = "org.apache.flink:flink-table-common:1.18.1"
 flink-table-planner = "org.apache.flink:flink-table-planner_2.12:1.18.1"
 flink-table-runtime = "org.apache.flink:flink-table-runtime:1.18.1"
-flink-connector-kafka = "org.apache.flink:flink-sql-connector-kafka:3.2.0-1.18"
-flink-connector-mysql-cdc = "com.ververica:flink-sql-connector-mysql-cdc:2.3.0"
 gson = "com.google.code.gson:gson:2.9.0"
 jackson = "com.fasterxml.jackson.core:jackson-core:2.14.1"
 jackson-dataformat-yaml = "com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:2.14.1"
@@ -27,8 +28,8 @@ kubernetes-client = "io.kubernetes:client-java:16.0.2"
 kubernetes-extended-client = "io.kubernetes:client-java-extended:16.0.2"
 slf4j-simple = "org.slf4j:slf4j-simple:1.7.30"
 slf4j-api = "org.slf4j:slf4j-api:1.7.30"
+snakeyaml = "org.yaml:snakeyaml:1.33"
 sqlline = "sqlline:sqlline:1.12.0"
-commons-cli = "commons-cli:commons-cli:1.4"
 quidem = "net.hydromatic:quidem:0.11"
 venice = "com.linkedin.venice:venice-common:0.4.376"
 venice-client = "com.linkedin.venice:venice-thin-client:0.4.376"

gradle/libs.versions.toml

@@ -4,5 +4,5 @@
 
 public interface ConfigProvider {
 
-  Properties loadConfig() throws Exception;
+  Properties loadConfig(String namespace) throws Exception;
 }

hoptimator-api/src/main/java/com/linkedin/hoptimator/ConfigProvider.java

@@ -6,7 +6,7 @@ plugins {
 
 dependencies {
   implementation project(':hoptimator-catalog')
-  implementation project(':hoptimator-models')
+  implementation project(':hoptimator-k8s')
   implementation project(':hoptimator-operator')
   implementation libs.kubernetesClient
   implementation libs.kubernetesExtendedClient

hoptimator-flink-adapter/build.gradle

@@ -1,7 +1,7 @@
 package com.linkedin.hoptimator.operator.flink;
 
-import com.linkedin.hoptimator.models.V1alpha1SqlJob;
-import com.linkedin.hoptimator.models.V1alpha1SqlJobList;
+import com.linkedin.hoptimator.k8s.models.V1alpha1SqlJob;
+import com.linkedin.hoptimator.k8s.models.V1alpha1SqlJobList;
 import com.linkedin.hoptimator.operator.ControllerProvider;
 import com.linkedin.hoptimator.operator.Operator;
 

hoptimator-flink-adapter/src/main/java/com/linkedin/hoptimator/operator/flink/FlinkControllerProvider.java

@@ -2,11 +2,11 @@
 
 import com.linkedin.hoptimator.catalog.Resource;
 import com.linkedin.hoptimator.catalog.flink.FlinkStreamingSqlJob;
+import com.linkedin.hoptimator.k8s.models.V1alpha1SqlJob;
+import com.linkedin.hoptimator.k8s.models.V1alpha1SqlJobSpec.DialectEnum;
+import com.linkedin.hoptimator.k8s.models.V1alpha1SqlJobSpec.ExecutionModeEnum;
+import com.linkedin.hoptimator.k8s.models.V1alpha1SqlJobStatus;
 import com.linkedin.hoptimator.operator.Operator;
-import com.linkedin.hoptimator.models.V1alpha1SqlJob;
-import com.linkedin.hoptimator.models.V1alpha1SqlJobSpec.DialectEnum;
-import com.linkedin.hoptimator.models.V1alpha1SqlJobSpec.ExecutionModeEnum;
-import com.linkedin.hoptimator.models.V1alpha1SqlJobStatus;
 
 import io.kubernetes.client.extended.controller.reconciler.Reconciler;
 import io.kubernetes.client.extended.controller.reconciler.Request;

hoptimator-flink-adapter/src/main/java/com/linkedin/hoptimator/operator/flink/FlinkStreamingSqlJobReconciler.java

@@ -6,7 +6,7 @@
 
 public class SystemPropertiesConfigProvider implements ConfigProvider {
 
-  public Properties loadConfig() {
+  public Properties loadConfig(String namespace) {
     return System.getProperties();
   }
 }

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/SystemPropertiesConfigProvider.java

@@ -14,16 +14,19 @@ public class K8sConfigProvider implements ConfigProvider {
 
   public static final String HOPTIMATOR_CONFIG_MAP = "hoptimator-configmap";
 
-  public Properties loadConfig() throws SQLException {
-    Map<String, String> topLevelConfigs = loadTopLevelConfig(HOPTIMATOR_CONFIG_MAP);
+  public Properties loadConfig(String namespace) throws SQLException {
+    Map<String, String> topLevelConfigs = loadTopLevelConfig(HOPTIMATOR_CONFIG_MAP, namespace);
     Properties p = new Properties();
     p.putAll(topLevelConfigs);
     return p;
   }
 
   // Load top-level config map properties
-  private Map<String, String> loadTopLevelConfig(String configMapName) throws SQLException {
+  private Map<String, String> loadTopLevelConfig(String configMapName, String namespace) throws SQLException {
     K8sApi<V1ConfigMap, V1ConfigMapList> configMapApi = new K8sApi<>(K8sContext.currentContext(), K8sApiEndpoints.CONFIG_MAPS);
-    return configMapApi.get(configMapName).getData();
+    if (namespace == null || namespace.isEmpty()) {
+      return configMapApi.get(configMapName).getData();
+    }
+    return configMapApi.get(namespace, configMapName).getData();
   }
 }