Drop SqlJob and FlinkSqlJobReconciler (#35)

* Drop SqlJob and FlinkSqlJobReconciler

* Change name of Resource templates back to 'template'

* Enable flink adapter in CLI

* Update readme

Author: ryannedolan

Makefile

@@ -36,8 +36,7 @@ deploy-dev-environment:
 deploy-samples: deploy
 	kubectl wait --for=condition=Established=True	\
 	  crds/subscriptions.hoptimator.linkedin.com \
-	  crds/kafkatopics.hoptimator.linkedin.com \
-	  crds/sqljobs.hoptimator.linkedin.com
+	  crds/kafkatopics.hoptimator.linkedin.com
 	kubectl apply -f ./deploy/samples
 
 deploy-config:

README.md

@@ -59,7 +59,6 @@ In response, the operator will deploy a Flink job and other resources:
 
 ```
   $ kubectl get subscriptions
-  $ kubectl get sqljobs
   $ kubectl get flinkdeployments
   $ kubectl get kafkatopics
 ```

deploy/rbac.yaml

@@ -5,7 +5,7 @@ metadata:
   name: hoptimator-controller
 rules:
 - apiGroups: ["hoptimator.linkedin.com"]
-  resources: ["sqljobs", "kafkatopics", "subscriptions"]
+  resources: ["kafkatopics", "subscriptions"]
   verbs: ["get", "watch", "list", "update", "create"]
 - apiGroups: ["flink.apache.org"]
   resources: ["flinkdeployments"]

deploy/sqljobs.crd.yaml

@@ -33,24 +33,23 @@
  * for informational/debugging purposes.
  */
 public abstract class Resource {
-  private final String kind;
+  private final String template;
   private final SortedMap<String, Supplier<String>> properties = new TreeMap<>();
   private final List<Resource> inputs = new ArrayList<>();
 
-  /** A Resource of some kind. */
-  public Resource(String kind) {
-    this.kind = kind;
+  public Resource(String template) {
+    this.template = template;
   }
 
   /** Copy constructor */
   public Resource(Resource other) {
-    this.kind = other.kind;
+    this.template = other.template;
     this.properties.putAll(other.properties);
     this.inputs.addAll(other.inputs);
   }
 
-  public String kind() {
-    return kind;
+  public String template() {
+    return template;
   }
 
   /** Export a computed value to the template */
@@ -109,7 +108,7 @@ public int hashCode() {
   @Override
   public String toString() {
     StringBuilder sb = new StringBuilder();
-    sb.append("[ kind: " + kind() + " ");
+    sb.append("[ template: " + template() + " ");
     for (Map.Entry<String, Supplier<String>> entry : properties.entrySet()) {
       String value = entry.getValue().get();
       if (value != null && !value.isEmpty()) {
@@ -280,10 +279,10 @@ public SimpleTemplateFactory(Environment env) {
 
     @Override
     public Template get(Resource resource) {
-      String kind = resource.kind();
-      InputStream in = getClass().getClassLoader().getResourceAsStream(kind + ".yaml.template");
+      String template = resource.template();
+      InputStream in = getClass().getClassLoader().getResourceAsStream(template + ".yaml.template");
       if (in == null) {
-        throw new IllegalArgumentException("No template '" + kind + "' found in jar resources");
+        throw new IllegalArgumentException("No template '" + template + "' found in jar resources");
       }
       StringBuilder sb = new StringBuilder();
       Scanner scanner = new Scanner(in);

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

@@ -17,6 +17,7 @@ dependencies {
   // include adapters for integration tests
   integration project(':hoptimator-kafka-adapter')
   integration project(':hoptimator-mysql-adapter')
+  integration project(':hoptimator-flink-adapter')
   integration libs.flinkCsv
   integration libs.flinkConnectorKafka
   integration libs.slf4jSimple

hoptimator-cli/build.gradle

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

hoptimator-cli/src/main/resources/SqlJob.yaml.template

@@ -2,31 +2,21 @@
 
 import com.linkedin.hoptimator.operator.ControllerProvider;
 import com.linkedin.hoptimator.operator.Operator;
-import com.linkedin.hoptimator.models.V1alpha1SqlJob;
 
 import io.kubernetes.client.extended.controller.Controller;
-import io.kubernetes.client.extended.controller.builder.ControllerBuilder;
-import io.kubernetes.client.extended.controller.reconciler.Reconciler;
 
 import java.util.Collection;
 import java.util.Collections;
 
-/** Provides a Controller plugin for FlinkDeployments. */
+/** A bridge to flink-kubernetes-operator */
 public class FlinkControllerProvider implements ControllerProvider {
 
   @Override
   public Collection<Controller> controllers(Operator operator) {
     operator.registerApi("FlinkDeployment", "flinkdeployment", "flinkdeployments",
       "flink.apache.org", "v1beta1");
 
-    Reconciler reconciler = new FlinkSqlJobReconciler(operator);
-    Controller controller = ControllerBuilder.defaultBuilder(operator.informerFactory())
-      .withReconciler(reconciler)
-      .withName("flink-deployment-controller")
-      .withWorkerCount(1)
-      .watch(x -> ControllerBuilder.controllerWatchBuilder(V1alpha1SqlJob.class, x).build())
-      .build();
-
-    return Collections.singleton(controller);
+    // We don't need a controller
+    return Collections.emptyList();
   }
 }