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;