Add ownership references (#114)

Author: ryannedolan

Makefile

@@ -69,7 +69,7 @@ deploy-kafka: deploy deploy-flink
 	kubectl apply -f ./deploy/samples/kafkadb.yaml
 
 undeploy-kafka:
-	kubectl delete kafkatopic.kafka.strimzi.io -n kafka --all || echo "skipping"
+	kubectl delete kafkatopic.kafka.strimzi.io --all || echo "skipping"
 	kubectl delete strimzi -n kafka --all || echo "skipping"
 	kubectl delete pvc -l strimzi.io/name=one-kafka -n kafka || echo "skipping"
 	kubectl delete -f "https://strimzi.io/install/latest?namespace=kafka" -n kafka || echo "skipping"
@@ -103,8 +103,8 @@ undeploy-dev-environment: undeploy-venice undeploy-kafka undeploy-flink undeploy
 # Integration test setup intended to be run locally
 integration-tests: deploy-dev-environment
 	kubectl wait kafka.kafka.strimzi.io/one --for=condition=Ready --timeout=10m -n kafka
-	kubectl wait kafkatopic.kafka.strimzi.io/existing-topic-1 --for=condition=Ready --timeout=10m -n kafka
-	kubectl wait kafkatopic.kafka.strimzi.io/existing-topic-2 --for=condition=Ready --timeout=10m -n kafka
+	kubectl wait kafkatopic.kafka.strimzi.io/kafka-database-existing-topic-1 --for=condition=Ready --timeout=10m
+	kubectl wait kafkatopic.kafka.strimzi.io/kafka-database-existing-topic-2 --for=condition=Ready --timeout=10m
 	kubectl port-forward -n kafka svc/one-kafka-external-bootstrap 9092 & echo $$! > port-forward.pid
 	kubectl port-forward -n flink svc/flink-sql-gateway 8083 & echo $$! > port-forward-2.pid
 	kubectl port-forward -n flink svc/basic-session-deployment-rest 8081 & echo $$! > port-forward-3.pid
@@ -116,8 +116,8 @@ integration-tests: deploy-dev-environment
 # kind cluster used in github workflow needs to have different routing set up, avoiding the need to forward kafka ports
 integration-tests-kind: deploy-dev-environment
 	kubectl wait kafka.kafka.strimzi.io/one --for=condition=Ready --timeout=10m -n kafka
-	kubectl wait kafkatopic.kafka.strimzi.io/existing-topic-1 --for=condition=Ready --timeout=10m -n kafka
-	kubectl wait kafkatopic.kafka.strimzi.io/existing-topic-2 --for=condition=Ready --timeout=10m -n kafka
+	kubectl wait kafkatopic.kafka.strimzi.io/kafka-database-existing-topic-1 --for=condition=Ready --timeout=10m 
+	kubectl wait kafkatopic.kafka.strimzi.io/kafka-database-existing-topic-2 --for=condition=Ready --timeout=10m 
 	./gradlew intTest -i
 
 generate-models:

deploy/dev/kafka.yaml

@@ -65,5 +65,41 @@ spec:
     storage:
       type: ephemeral
   entityOperator:
-    topicOperator: {}
+    topicOperator:
+      watchedNamespace: default
+
+---
+
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  namespace: kafka
+  name: kafka-operator
+rules:
+- apiGroups: ["kafka.strimzi.io"]
+  resources: ["*"]
+  verbs: ["*"]
+- apiGroups: ["rbac.authorization.k8s.io"]
+  resources: ["roles", "rolebindings"]
+  verbs: ["*"]
+
+
+---
+
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: kafka-operator
+  namespace: kafka
+subjects:
+- kind: ServiceAccount
+  name: one-entity-operator
+  namespace: kafka
+- kind: ServiceAccount
+  name: strimzi-cluster-operator
+  namespace: kafka
+roleRef:
+  kind: ClusterRole
+  name: kafka-operator
+  apiGroup: rbac.authorization.k8s.io
 

deploy/samples/flink-template.yaml

@@ -10,7 +10,6 @@ spec:
     kind: FlinkSessionJob
     metadata:
       name: {{name}}
-      namespace: flink
     spec:
       deploymentName: basic-session-deployment
       job:

deploy/samples/kafkadb.yaml

@@ -21,7 +21,6 @@ spec:
     kind: KafkaTopic
     metadata:
       name: {{name}}
-      namespace: kafka
       labels:
         strimzi.io/cluster: one
     spec:
@@ -46,11 +45,11 @@ spec:
 apiVersion: kafka.strimzi.io/v1beta2
 kind: KafkaTopic
 metadata:
-  name: existing-topic-1
-  namespace: kafka
+  name: kafka-database-existing-topic-1
   labels:
     strimzi.io/cluster: one
 spec:
+  topicName: existing-topic-1
   partitions: 1
   replicas: 1
   config:
@@ -62,11 +61,11 @@ spec:
 apiVersion: kafka.strimzi.io/v1beta2
 kind: KafkaTopic
 metadata:
-  name: existing-topic-2
-  namespace: kafka
+  name: kafka-database-existing-topic-2
   labels:
     strimzi.io/cluster: one
 spec:
+  topicName: existing-topic-2
   partitions: 1
   replicas: 1
   config:

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

@@ -4,7 +4,7 @@
 import java.util.Map;
 
 
-public interface Connector<T> {
+public interface Connector {
 
-  Map<String, String> configure(T t) throws SQLException;
+  Map<String, String> configure() throws SQLException;
 }

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

@@ -6,5 +6,6 @@
 
 public interface ConnectorProvider {
 
-  <T> Collection<Connector<T>> connectors(Class<T> clazz, Properties connectionProperties);
+  /** Find connectors capable of configuring data plane connectors for the obj. */
+  <T> Collection<Connector> connectors(T obj, Properties connectionProperties);
 }

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

@@ -1,17 +1,6 @@
 package com.linkedin.hoptimator;
 
-import java.sql.SQLException;
-import java.util.List;
-
 
 public interface Deployable {
 
-  void create() throws SQLException;
-
-  void delete() throws SQLException;
-
-  void update() throws SQLException;
-
-  /** Render a list of specs, usually YAML. */
-  List<String> specify() throws SQLException;
 }

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

@@ -4,13 +4,15 @@
 import java.util.List;
 
 
-public interface Deployer<T> {
+/** Deploys something. */
+public interface Deployer {
 
-  void create(T t) throws SQLException;
+  void create() throws SQLException;
 
-  void update(T t) throws SQLException;
+  void delete() throws SQLException;
 
-  void delete(T t) throws SQLException;
+  void update() throws SQLException;
 
-  List<String> specify(T t) throws SQLException;
+  /** Render a list of specs, usually YAML. */
+  List<String> specify() throws SQLException;
 }

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

@@ -6,5 +6,6 @@
 
 public interface DeployerProvider {
 
-  <T> Collection<Deployer<T>> deployers(Class<T> clazz, Properties connectionProperties);
+  /** Find deployers capable of deploying the obj. */
+  <T extends Deployable> Collection<Deployer> deployers(T obj, Properties connectionProperties);
 }

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

@@ -3,7 +3,7 @@
 import java.util.function.Function;
 
 
-public class Job {
+public class Job implements Deployable {
 
   private final String name;
   private final Sink sink;

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

@@ -5,28 +5,20 @@
 import java.util.stream.Collectors;
 
 
-public class MaterializedView {
+public class MaterializedView extends View implements Deployable {
 
   private final String database;
-  private final List<String> path;
-  private final String viewSql;
   private final Function<SqlDialect, String> pipelineSql;
   private final Pipeline pipeline;
 
   public MaterializedView(String database, List<String> path, String viewSql, Function<SqlDialect, String> pipelineSql,
       Pipeline pipeline) {
+    super(path, viewSql);
     this.database = database;
-    this.path = path;
-    this.viewSql = viewSql;
     this.pipelineSql = pipelineSql;
     this.pipeline = pipeline;
   }
 
-  /** SQL query which defines this view, e.g. SELECT ... FROM ... */
-  public String viewSql() {
-    return viewSql;
-  }
-
   public Pipeline pipeline() {
     return pipeline;
   }
@@ -40,22 +32,6 @@ public String database() {
     return database;
   }
 
-  public String table() {
-    return path.get(path.size() - 1);
-  }
-
-  public String schema() {
-    return path.get(path.size() - 2);
-  }
-
-  public List<String> path() {
-    return path;
-  }
-
-  protected String pathString() {
-    return path.stream().collect(Collectors.joining("."));
-  }
-
   @Override
   public String toString() {
     return "MaterializedView[" + pathString() + "]";

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

@@ -1,48 +1,33 @@
 package com.linkedin.hoptimator;
 
 import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.List;
+import java.util.Collection;
 
 
 /**
- * A set of Deployable objects that work together to deliver data.
+ * A job, along with its sources and sink.
  */
-public class Pipeline implements Deployable {
+public class Pipeline {
 
-  private List<Deployable> deployables;
+  private Collection<Source> sources;
+  private Sink sink;
+  private Job job;
 
-  public Pipeline(List<Deployable> deployables) {
-    this.deployables = deployables;
+  public Pipeline(Collection<Source> sources, Sink sink, Job job) {
+    this.sources = sources;
+    this.sink = sink;
+    this.job = job;
   }
 
-  @Override
-  public void create() throws SQLException {
-    for (Deployable deployable : deployables) {
-      deployable.create();
-    }
+  public Collection<Source> sources() {
+    return sources;
   }
 
-  @Override
-  public void delete() throws SQLException {
-    for (Deployable deployable : deployables) {
-      deployable.delete();
-    }
+  public Sink sink() {
+    return sink;
   }
 
-  @Override
-  public void update() throws SQLException {
-    for (Deployable deployable : deployables) {
-      deployable.update();
-    }
-  }
-
-  @Override
-  public List<String> specify() throws SQLException {
-    List<String> specs = new ArrayList<>();
-    for (Deployable deployable : deployables) {
-      specs.addAll(deployable.specify());
-    }
-    return specs;
+  public Job job() {
+    return job;
   }
 }

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

@@ -4,7 +4,8 @@
 import java.util.Map;
 
 
-public class Sink extends Source {
+// Unclear if Sink will always extend Source
+public class Sink extends Source implements Deployable {
 
   public Sink(String database, List<String> path, Map<String, String> options) {
     super(database, path, options);

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

@@ -5,7 +5,7 @@
 import java.util.stream.Collectors;
 
 
-public class Source {
+public class Source implements Deployable {
 
   private final String database;
   private final List<String> path;

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

@@ -11,9 +11,7 @@
 import java.util.function.Function;
 
 
-public interface Validator<T> {
-
-  void validate(T t, Issues issues);
+public interface Validator extends Validated {
 
   static void validateSubdomainName(String s, Issues issues) {
     // N.B. we don't aim to be efficient here; rather, as verbose as possible.
@@ -44,15 +42,18 @@ static void notYetImplemented(Issues issues) {
     issues.warn("Validation not implemented for this object");
   }
 
-  /** Validator that invokes `Validated.validate()`. */
-  class DefaultValidator<T> implements Validator<T> {
+  /** Validator that invokes `validate()` on the target object. */
+  class DefaultValidator<T extends Validated> implements Validator {
+
+    private final T t;
+
+    public DefaultValidator(T t) {
+      this.t = t;
+    }
 
     @Override
-    public void validate(T t, Issues issues) {
-      if (t instanceof Validated) {
-        Validated v = (Validated) t;
-        v.validate(issues.child(t.getClass().getSimpleName()));
-      }
+    public void validate(Issues issues) {
+      t.validate(issues.child(t.getClass().getSimpleName()));
     }
   }
 

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

@@ -5,5 +5,5 @@
 
 public interface ValidatorProvider {
 
-  <T> Collection<Validator<T>> validators(Class<T> clazz);
+  <T> Collection<Validator> validators(T obj);
 }