Subscription "hints"

bin/hoptimator

@@ -1,3 +1,3 @@
 #!/bin/sh
 
-kubectl exec -it hoptimator -c hoptimator -- ./hoptimator --isolation=TRANSACTION_NONE -n "" -p "" -u "jdbc:calcite:model=/etc/config/model.yaml" -nn hoptimator "$@"
+kubectl exec -it hoptimator -c hoptimator -- ./hoptimator -n "" -p "" -u "jdbc:calcite:model=/etc/config/model.yaml" "$@"

deploy/samples/subscriptions.yaml

@@ -6,4 +6,5 @@ metadata:
 spec:
   sql: SELECT "quantity", "product_id" AS KEY FROM INVENTORY."products_on_hand"
   database: RAWKAFKA
-
+  hints:
+    kafka.numPartitions: "2"

deploy/subscriptions.crd.yaml

@@ -38,6 +38,11 @@ spec:
                 database:
                   description: The database in which to create the output/sink table.
                   type: string
+                hints:
+                  description: Hints to adapters, which may disregard them.
+                  type: object
+                  additionalProperties:
+                    type: string
               required:
               - sql
               - database

hoptimator-catalog/src/main/java/com/linkedin/hoptimator/catalog/Resource.java

@@ -148,15 +148,19 @@ public interface Environment {
     Environment EMPTY = new SimpleEnvironment();
     Environment PROCESS = new ProcessEnvironment();
 
-    String get(String key);
+    String getOrDefault(String key, String defaultValue);
   }
 
   /** Basic Environment implementation */
   public static class SimpleEnvironment implements Environment {
     private final Map<String, String> vars;
 
     public SimpleEnvironment(Map<String, String> vars) {
-      this.vars = vars;
+      if (vars != null) {
+        this.vars = vars;
+      } else {
+        this.vars = new HashMap<>();
+      }
     }
 
     public SimpleEnvironment() {
@@ -182,31 +186,38 @@ public SimpleEnvironment with(String key, String value) {
     }
 
     @Override
-    public String get(String key) {
-      if (!vars.containsKey(key)) {
+    public String getOrDefault(String key, String defaultValue) {
+      if (defaultValue == null && !vars.containsKey(key)) {
         throw new IllegalArgumentException("No variable '" + key + "' found in the environment");
       }
-      return vars.get(key);
+      return vars.getOrDefault(key, defaultValue);
     }
   }
 
-  /** Returns "{{key}}" for any key */
+  /** Returns "{{key}}" for any key without a default */
   public static class DummyEnvironment implements Environment {
     @Override
-    public String get(String key) {
-      return "{{" + key + "}}";
+    public String getOrDefault(String key, String defaultValue) {
+      if (defaultValue != null) {
+        return defaultValue;
+      } else {
+        return "{{" + key + "}}";
+      }
     }
   }
 
   /** Provides access to the process's environment variables */
   public static class ProcessEnvironment implements Environment {
 
     @Override
-    public String get(String key) {
+    public String getOrDefault(String key, String defaultValue) {
       String value = System.getenv(key);
       if (value == null) {
         value = System.getProperty(key);
       }
+      if (value == null) {
+        value = defaultValue;
+      }
       if (value == null) {
         throw new IllegalArgumentException("Missing system property `" + key + "`");
       }
@@ -222,11 +233,12 @@ public interface Template {
   /**
    * Replaces `{{var}}` in a template file with the corresponding variable.
    *
-   * Resource-scoped variables take precedence over Environment-scoped variables.
+   * Resource-scoped variables take precedence over Environment-scoped
+   * variables. Default values can supplied with `{{var:default}}`.
    *
-   * If `var` contains multiple lines, the behavior depends on context; specifically,
-   * whether the pattern appears within a list or comment (prefixed with `-` or `#`).
-   * For example, if the template includes:
+   * If `var` contains multiple lines, the behavior depends on context;
+   * specifically, whether the pattern appears within a list or comment
+   * (prefixed with `-` or `#`). For example, if the template includes:
    *
    *   - {{var}}
    *
@@ -235,8 +247,8 @@ public interface Template {
    *   - value line 1
    *   - value line 2
    *
-   * To avoid this behavior (and just get a multiline string), use one of YAML's multiline
-   * markers, e.g.
+   * To avoid this behavior (and just get a multiline string), use one of
+   * YAML's multiline markers, e.g.
    *
    *   - |
    *       {{var}}
@@ -255,17 +267,18 @@ public SimpleTemplate(Environment env, String template) {
     @Override
     public String render(Resource resource) {
       StringBuffer sb = new StringBuffer();
-      Pattern p = Pattern.compile("([\\s\\-\\#]*)\\{\\{\\s*([\\w_\\-\\.]+)\\s*\\}\\}");
+      Pattern p = Pattern.compile("([\\s\\-\\#]*)\\{\\{\\s*([\\w_\\-\\.]+)\\s*(:([\\w_\\-\\.]+))?\\s*\\}\\}");
       Matcher m = p.matcher(template);
       while (m.find()) {
         String prefix = m.group(1);
         if (prefix == null) {
           prefix = "";
         }
         String key = m.group(2);
-        String value = resource.getOrDefault(key, () -> env.get(key));
+        String defaultValue = m.group(4);
+        String value = resource.getOrDefault(key, () -> env.getOrDefault(key, defaultValue));
         if (value == null) {
-          throw new IllegalArgumentException("No value for key " + key);
+          throw new IllegalArgumentException(template + " has no value for key " + key + ".");
         }
         String quotedPrefix = Matcher.quoteReplacement(prefix);
         String quotedValue = Matcher.quoteReplacement(value);

hoptimator-cli/run.sh

@@ -5,5 +5,6 @@ java \
   --add-opens java.base/java.util=ALL-UNNAMED \
   --add-opens java.base/java.time=ALL-UNNAMED \
   -classpath "/opt/plugins/*/lib/*:./hoptimator-cli-all.jar" \
+  -Dorg.slf4j.simpleLogger.defaultLogLevel=error \
   $JAVA_OPTS \
-  com.linkedin.hoptimator.HoptimatorCliApp --verbose=true -nn hoptimator "$@"
+  com.linkedin.hoptimator.HoptimatorCliApp --verbose=true -nn hoptimator --isolation=TRANSACTION_NONE "$@"

hoptimator-flink-adapter/src/main/resources/SqlJob.yaml.template

@@ -1,29 +1,29 @@
 apiVersion: flink.apache.org/v1beta1
 kind: FlinkDeployment
 metadata:
-  namespace: {{namespace}}
-  name: {{name}}-flink-job
+  name: {{pipeline.name}}-flink-job
+  namespace: {{pipeline.namespace}}
 spec:
   image: docker.io/library/hoptimator-flink-runner
   imagePullPolicy: Never
   flinkVersion: v1_16
   flinkConfiguration:
-    taskmanager.numberOfTaskSlots: "1"
+    taskmanager.numberOfTaskSlots: "{{flink.taskmanager.numberOfTaskSlots:1}}"
   serviceAccount: flink
   jobManager:
     resource:
-      memory: "2048m"
-      cpu: .1
+      memory: "{{flink.job.memory:2048m}}"
+      cpu: {{flink.job.cpu:.1}}
   taskManager:
     resource:
-      memory: "2048m"
-      cpu: .1
+      memory: "{{flink.task.memory:2048m}}"
+      cpu: {{flink.task.cpu:.1}}
   job:
     entryClass: com.linkedin.hoptimator.flink.runner.FlinkRunner
     args:
-    - {{sql}}
+    - {{pipeline.sql}}
     jarURI: local:///opt/hoptimator-flink-runner.jar
-    parallelism: 1
+    parallelism: {{flink.parallelism:1}}
     upgradeMode: stateless
     state: running
 

hoptimator-kafka-adapter/src/main/java/com/linkedin/hoptimator/catalog/kafka/KafkaTopic.java

@@ -7,12 +7,9 @@
 import java.util.Map;
 
 class KafkaTopic extends Resource {
-  public KafkaTopic(String name, Integer numPartitions,
-      Map<String, String> clientOverrides) {
+  public KafkaTopic(String name, Map<String, String> clientOverrides) {
     super("KafkaTopic");
     export("name", name);
-    export("numPartitions", Optional.ofNullable(numPartitions)
-      .map(x -> Integer.toString(x)).orElse("null"));
     export("clientOverrides", clientOverrides);
   }
 }

hoptimator-kafka-adapter/src/main/java/com/linkedin/hoptimator/catalog/kafka/RawKafkaSchemaFactory.java

@@ -44,11 +44,12 @@ public Schema create(SchemaPlus parentSchema, String name, Map<String, Object> o
     };
     ConfigProvider topicConfigProvider = ConfigProvider.from(clientConfig);
     TableResolver resolver = x -> rowType.rel();
-    Integer numPartitions = (Integer) operand.get("numPartitions");
+
     ResourceProvider resources = ResourceProvider.empty()
-      .with(x -> new KafkaTopic(x, numPartitions, topicConfigProvider.config(x)))
+      .with(x -> new KafkaTopic(x, topicConfigProvider.config(x)))
       .readWith(x -> new KafkaTopicAcl(x, principal, "Read"))
       .writeWith(x -> new KafkaTopicAcl(x, principal, "Write"));
+
     Database database = new Database(name, tableLister, resolver, connectorConfigProvider,
       resources);
     return new DatabaseSchema(database);

hoptimator-kafka-adapter/src/main/resources/KafkaTopic.yaml.template

@@ -1,10 +1,10 @@
 apiVersion: hoptimator.linkedin.com/v1alpha1
 kind: KafkaTopic
 metadata:
-  name: {{name}}
-  namespace: {{namespace}}
+  name: {{pipeline.name}}-kafka-topic-{{id}}
+  namespace: {{pipeline.namespace}}
 spec:
   topicName: {{name}}
-  numPartitions: {{numPartitions}}
+  numPartitions: {{kafka.numPartitions:null}}
   clientOverrides:
     {{clientOverrides}}

hoptimator-kafka-adapter/src/main/resources/KafkaTopicAcl.yaml.template

@@ -2,7 +2,7 @@ apiVersion: hoptimator.linkedin.com/v1alpha1
 kind: Acl
 metadata:
   name: {{name}}-acl-{{id}}
-  namespace: {{namespace}}
+  namespace: {{pipeline.namespace}}
 spec:
   resource:
     kind: KafkaTopic

hoptimator-models/src/main/java/com/linkedin/hoptimator/models/V1alpha1Acl.java

@@ -31,7 +31,7 @@
  * Access control rule (colloquially, an Acl)
  */
 @ApiModel(description = "Access control rule (colloquially, an Acl)")
-@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-19T17:12:58.614Z[Etc/UTC]")
+@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-21T22:04:16.918Z[Etc/UTC]")
 public class V1alpha1Acl implements io.kubernetes.client.common.KubernetesObject {
   public static final String SERIALIZED_NAME_API_VERSION = "apiVersion";
   @SerializedName(SERIALIZED_NAME_API_VERSION)

hoptimator-models/src/main/java/com/linkedin/hoptimator/models/V1alpha1AclList.java

@@ -32,7 +32,7 @@
  * AclList is a list of Acl
  */
 @ApiModel(description = "AclList is a list of Acl")
-@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-19T17:12:58.614Z[Etc/UTC]")
+@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-21T22:04:16.918Z[Etc/UTC]")
 public class V1alpha1AclList implements io.kubernetes.client.common.KubernetesListObject {
   public static final String SERIALIZED_NAME_API_VERSION = "apiVersion";
   @SerializedName(SERIALIZED_NAME_API_VERSION)

hoptimator-models/src/main/java/com/linkedin/hoptimator/models/V1alpha1AclSpec.java

@@ -29,7 +29,7 @@
  * A set of related ACL rules.
  */
 @ApiModel(description = "A set of related ACL rules.")
-@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-19T17:12:58.614Z[Etc/UTC]")
+@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-21T22:04:16.918Z[Etc/UTC]")
 public class V1alpha1AclSpec {
   /**
    * The resource access method.

hoptimator-models/src/main/java/com/linkedin/hoptimator/models/V1alpha1AclSpecResource.java

@@ -28,7 +28,7 @@
  * The resource being controlled.
  */
 @ApiModel(description = "The resource being controlled.")
-@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-19T17:12:58.614Z[Etc/UTC]")
+@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-21T22:04:16.918Z[Etc/UTC]")
 public class V1alpha1AclSpecResource {
   public static final String SERIALIZED_NAME_KIND = "kind";
   @SerializedName(SERIALIZED_NAME_KIND)

hoptimator-models/src/main/java/com/linkedin/hoptimator/models/V1alpha1AclStatus.java

@@ -28,7 +28,7 @@
  * Status, as set by the operator.
  */
 @ApiModel(description = "Status, as set by the operator.")
-@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-19T17:12:58.614Z[Etc/UTC]")
+@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-21T22:04:16.918Z[Etc/UTC]")
 public class V1alpha1AclStatus {
   public static final String SERIALIZED_NAME_MESSAGE = "message";
   @SerializedName(SERIALIZED_NAME_MESSAGE)

hoptimator-models/src/main/java/com/linkedin/hoptimator/models/V1alpha1KafkaTopic.java

@@ -31,7 +31,7 @@
  * Kafka Topic
  */
 @ApiModel(description = "Kafka Topic")
-@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-19T17:12:58.614Z[Etc/UTC]")
+@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-21T22:04:16.918Z[Etc/UTC]")
 public class V1alpha1KafkaTopic implements io.kubernetes.client.common.KubernetesObject {
   public static final String SERIALIZED_NAME_API_VERSION = "apiVersion";
   @SerializedName(SERIALIZED_NAME_API_VERSION)

hoptimator-models/src/main/java/com/linkedin/hoptimator/models/V1alpha1KafkaTopicList.java

@@ -32,7 +32,7 @@
  * KafkaTopicList is a list of KafkaTopic
  */
 @ApiModel(description = "KafkaTopicList is a list of KafkaTopic")
-@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-19T17:12:58.614Z[Etc/UTC]")
+@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-21T22:04:16.918Z[Etc/UTC]")
 public class V1alpha1KafkaTopicList implements io.kubernetes.client.common.KubernetesListObject {
   public static final String SERIALIZED_NAME_API_VERSION = "apiVersion";
   @SerializedName(SERIALIZED_NAME_API_VERSION)

hoptimator-models/src/main/java/com/linkedin/hoptimator/models/V1alpha1KafkaTopicSpec.java

@@ -33,7 +33,7 @@
  * Desired Kafka topic configuration.
  */
 @ApiModel(description = "Desired Kafka topic configuration.")
-@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-19T17:12:58.614Z[Etc/UTC]")
+@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-21T22:04:16.918Z[Etc/UTC]")
 public class V1alpha1KafkaTopicSpec {
   public static final String SERIALIZED_NAME_CLIENT_CONFIGS = "clientConfigs";
   @SerializedName(SERIALIZED_NAME_CLIENT_CONFIGS)

hoptimator-models/src/main/java/com/linkedin/hoptimator/models/V1alpha1KafkaTopicSpecClientConfigs.java

@@ -28,7 +28,7 @@
 /**
  * V1alpha1KafkaTopicSpecClientConfigs
  */
-@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-19T17:12:58.614Z[Etc/UTC]")
+@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-21T22:04:16.918Z[Etc/UTC]")
 public class V1alpha1KafkaTopicSpecClientConfigs {
   public static final String SERIALIZED_NAME_CONFIG_MAP_REF = "configMapRef";
   @SerializedName(SERIALIZED_NAME_CONFIG_MAP_REF)

hoptimator-models/src/main/java/com/linkedin/hoptimator/models/V1alpha1KafkaTopicSpecConfigMapRef.java

@@ -28,7 +28,7 @@
  * Reference to a ConfigMap to use for AdminClient configuration.
  */
 @ApiModel(description = "Reference to a ConfigMap to use for AdminClient configuration.")
-@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-19T17:12:58.614Z[Etc/UTC]")
+@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-21T22:04:16.918Z[Etc/UTC]")
 public class V1alpha1KafkaTopicSpecConfigMapRef {
   public static final String SERIALIZED_NAME_NAME = "name";
   @SerializedName(SERIALIZED_NAME_NAME)

hoptimator-models/src/main/java/com/linkedin/hoptimator/models/V1alpha1KafkaTopicStatus.java

@@ -28,7 +28,7 @@
  * Current state of the topic.
  */
 @ApiModel(description = "Current state of the topic.")
-@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-19T17:12:58.614Z[Etc/UTC]")
+@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-21T22:04:16.918Z[Etc/UTC]")
 public class V1alpha1KafkaTopicStatus {
   public static final String SERIALIZED_NAME_MESSAGE = "message";
   @SerializedName(SERIALIZED_NAME_MESSAGE)

hoptimator-models/src/main/java/com/linkedin/hoptimator/models/V1alpha1Subscription.java

@@ -31,7 +31,7 @@
  * Hoptimator Subscription
  */
 @ApiModel(description = "Hoptimator Subscription")
-@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-19T17:12:58.614Z[Etc/UTC]")
+@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-21T22:04:16.918Z[Etc/UTC]")
 public class V1alpha1Subscription implements io.kubernetes.client.common.KubernetesObject {
   public static final String SERIALIZED_NAME_API_VERSION = "apiVersion";
   @SerializedName(SERIALIZED_NAME_API_VERSION)

hoptimator-models/src/main/java/com/linkedin/hoptimator/models/V1alpha1SubscriptionList.java

@@ -32,7 +32,7 @@
  * SubscriptionList is a list of Subscription
  */
 @ApiModel(description = "SubscriptionList is a list of Subscription")
-@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-19T17:12:58.614Z[Etc/UTC]")
+@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2023-07-21T22:04:16.918Z[Etc/UTC]")
 public class V1alpha1SubscriptionList implements io.kubernetes.client.common.KubernetesListObject {
   public static final String SERIALIZED_NAME_API_VERSION = "apiVersion";
   @SerializedName(SERIALIZED_NAME_API_VERSION)