Support connector hint passthrough (#127)

* Support connector hint passthrough

* touch up README

* revert not needed change

* Use connector name for hints

Author: jogrogan

README.md

@@ -148,7 +148,7 @@ In this case, any jobs created with this template will get deployed as `FlinkSes
 
 ### Configuration
 
-The ``{{ }}`` sections you see in the templates are variable placeholders that will be filled in by the Deployer.
+The `{{ }}` sections you see in the templates are variable placeholders that will be filled in by the Deployer.
 See [Template.java](hoptimator-util/src/main/java/com/linkedin/hoptimator/util/Template.java) for how to specify templates.
 
 While Deployers are extensible, today the primary deployer is to Kubernetes. These deployers 
@@ -180,10 +180,25 @@ This can be done by adding hints as JDBC properties.
 
 Hints are key-value pairs separated by an equals sign. Multiple hints are separated by a comma.
 
-For example, to specify the number of kafka partitions and the flink parallelism, you could add the following hints to the query:
+There are two ways to use hints.
+
+1. Template hints can be used to override the `{{ }}` template specifications. 
+
+For example, to specify the number of kafka partitions and the flink parallelism, you could add the following hints to the connection:
 ```
 jdbc:hoptimator://hints=kafka.partitions=4,flink.parallelism=2
 ```
 These fields can then be added to templates as `{{kafka.partitions}}` or `{{flink.parallelism}}` where applicable.
 
+2. Connector hints allow the user to pass configurations directly through to an Engine (e.g. Flink).
+
+Connector hints must be formatted as follows `<engine-connector-name>.<source|sink>.<configName>`
+
+For example, to set a Kafka group id and startup mode to be used by Flink, you could add the following hints to the connection:
+```
+jdbc:hoptimator://hints=kafka.source.properties.group.id=4,kafka.sink.sink.parallelism=2
+```
+Field `properties.group.id` will be applied if the `kafka` connector is used by a source, and `sink.parallelism`
+if the `kafka` connector is used by a sink.
+
 Note that hints are simply recommendations, if the planner plans a different pipeline, they will be ignored.

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sConnector.java

@@ -10,6 +10,7 @@
 import java.util.stream.Collectors;
 
 import com.linkedin.hoptimator.Connector;
+import com.linkedin.hoptimator.Sink;
 import com.linkedin.hoptimator.Source;
 import com.linkedin.hoptimator.k8s.models.V1alpha1TableTemplate;
 import com.linkedin.hoptimator.k8s.models.V1alpha1TableTemplateList;
@@ -46,6 +47,16 @@ public Map<String, String> configure() throws SQLException {
         .collect(Collectors.joining("\n"));
     Properties props = new Properties();
     try {
+      // Preload configs in order to check for 'connector'
+      props.load(new StringReader(configs));
+
+      // The order here is intentional. Connection options that allow overrides will have already been overridden above
+      // by hints. Connection options that do not allow overrides, should not be overridden by hints.
+      // If this were allowed, this can lead to incorrect behavior if hints attempt to override essential properties,
+      // e.g. 'connector' or 'topic' for kafka
+      if (props.containsKey("connector")) {
+        props.putAll(getConnectorHints(props.getProperty("connector")));
+      }
       props.load(new StringReader(configs));
     } catch (IOException e) {
       throw new SQLException(e);
@@ -55,4 +66,11 @@ public Map<String, String> configure() throws SQLException {
         map.put(k, props.getProperty(k)));
     return map;
   }
+
+  private Map<String, String> getConnectorHints(String connectorName) {
+    String connectorHintPrefix = connectorName + "." + (source instanceof Sink ? "sink" : "source");
+    return source.options().entrySet().stream()
+        .filter(e -> e.getKey().startsWith(connectorHintPrefix))
+        .collect(Collectors.toMap(e -> e.getKey().substring(connectorHintPrefix.length() + 1), Map.Entry::getValue));
+  }
 }

hoptimator-kafka/src/test/java/com/linkedin/hoptimator/kafka/TestSqlScripts.java

@@ -11,6 +11,6 @@ public class TestSqlScripts extends QuidemTestBase {
   @Test
   @Tag("integration")
   public void kafkaDdlScript() throws Exception {
-    run("kafka-ddl.id", "hints=kafka.partitions=4,flink.parallelism=2");
+    run("kafka-ddl.id", "hints=kafka.partitions=4,flink.parallelism=2,kafka.source.k1=v1,kafka.sink.k2=v2");
   }
 }

hoptimator-kafka/src/test/resources/kafka-ddl.id

@@ -12,9 +12,9 @@ spec:
     entryClass: com.linkedin.hoptimator.flink.runner.FlinkRunner
     args:
     - CREATE DATABASE IF NOT EXISTS `KAFKA` WITH ()
-    - CREATE TABLE IF NOT EXISTS `KAFKA`.`existing-topic-2` (`KEY` VARCHAR, `VALUE` BINARY) WITH ('connector'='kafka', 'key.fields'='KEY', 'key.format'='raw', 'properties.bootstrap.servers'='one-kafka-bootstrap.kafka.svc.cluster.local:9094', 'scan.startup.mode'='earliest-offset', 'topic'='existing-topic-2', 'value.fields-include'='EXCEPT_KEY', 'value.format'='json')
+    - CREATE TABLE IF NOT EXISTS `KAFKA`.`existing-topic-2` (`KEY` VARCHAR, `VALUE` BINARY) WITH ('connector'='kafka', 'k1'='v1', 'key.fields'='KEY', 'key.format'='raw', 'properties.bootstrap.servers'='one-kafka-bootstrap.kafka.svc.cluster.local:9094', 'scan.startup.mode'='earliest-offset', 'topic'='existing-topic-2', 'value.fields-include'='EXCEPT_KEY', 'value.format'='json')
     - CREATE DATABASE IF NOT EXISTS `KAFKA` WITH ()
-    - CREATE TABLE IF NOT EXISTS `KAFKA`.`existing-topic-1` (`KEY` VARCHAR, `VALUE` BINARY) WITH ('connector'='kafka', 'key.fields'='KEY', 'key.format'='raw', 'properties.bootstrap.servers'='one-kafka-bootstrap.kafka.svc.cluster.local:9094', 'scan.startup.mode'='earliest-offset', 'topic'='existing-topic-1', 'value.fields-include'='EXCEPT_KEY', 'value.format'='json')
+    - CREATE TABLE IF NOT EXISTS `KAFKA`.`existing-topic-1` (`KEY` VARCHAR, `VALUE` BINARY) WITH ('connector'='kafka', 'k2'='v2', 'key.fields'='KEY', 'key.format'='raw', 'properties.bootstrap.servers'='one-kafka-bootstrap.kafka.svc.cluster.local:9094', 'scan.startup.mode'='earliest-offset', 'topic'='existing-topic-1', 'value.fields-include'='EXCEPT_KEY', 'value.format'='json')
     - INSERT INTO `KAFKA`.`existing-topic-1` (`KEY`, `VALUE`) SELECT * FROM `KAFKA`.`existing-topic-2`
     jarURI: file:///opt/hoptimator-flink-runner.jar
     parallelism: 2

hoptimator-util/src/main/java/com/linkedin/hoptimator/util/Template.java

@@ -107,7 +107,7 @@ private String formatMapAsString(Map<String, String> configMap) {
     private String formatPropertiesAsString(Properties props) {
       StringBuilder stringBuilder = new StringBuilder();
       for (String key : props.stringPropertyNames()) {
-        stringBuilder.append(key).append(": ").append(props.getProperty(key)).append("\n");
+        stringBuilder.append(key).append(": '").append(props.getProperty(key)).append("'\n");
       }
       return stringBuilder.toString();
     }