MySQL CDC adapter and Kafka upserts (#2)

Author: ryannedolan

Dockerfile

@@ -5,6 +5,6 @@ COPY ./hoptimator-operator/run.sh ./hoptimator-operator
 COPY ./hoptimator-cli/build/libs/hoptimator-cli-all.jar ./hoptimator-cli-all.jar
 COPY ./hoptimator-operator/build/libs/hoptimator-operator-all.jar ./hoptimator-operator-all.jar
 COPY ./etc/* ./
-COPY ./test-model.yaml ./test-model.yaml
 ENTRYPOINT ["/bin/sh", "-c"]
-CMD ["./hoptimator"]
+CMD ["./hoptimator -n '' -p '' -u jdbc:calcite:model=model.yaml"]
+

Makefile

@@ -1,12 +1,10 @@
 
-bounce: build undeploy deploy
-
 build:
 	./gradlew build
 	docker build . -t hoptimator
 	docker build hoptimator-flink-runner -t hoptimator-flink-runner
 
-bounce: build undeploy deploy deploy-samples
+bounce: build undeploy deploy deploy-samples deploy-config
 
 integration-tests:
 	./bin/hoptimator --run=./integration-tests.sql
@@ -15,30 +13,35 @@ integration-tests:
 clean:
 	./gradlew clean
 
+deploy: deploy-config
+	kubectl apply -f ./deploy
+
 undeploy:
 	kubectl delete -f ./deploy || echo "skipping"
+	kubectl delete configmap hoptimator-configmap || echo "skipping"
 
 quickstart: build deploy-dev-environment deploy
 
 deploy-dev-environment: 
 	kubectl create -f https://github.com/jetstack/cert-manager/releases/download/v1.8.2/cert-manager.yaml || echo "skipping"
-	helm repo add flink-operator-repo https://downloads.apache.org/flink/flink-kubernetes-operator-1.4.0/
-	helm upgrade --atomic --set webhook.create=false flink-kubernetes-operator flink-operator-repo/flink-kubernetes-operator || helm install --atomic --set webhook.create=false flink-kubernetes-operator flink-operator-repo/flink-kubernetes-operator
 	kubectl create namespace kafka || echo "skipping"
+	kubectl create namespace mysql || echo "skipping"
+	helm repo add flink-operator-repo https://downloads.apache.org/flink/flink-kubernetes-operator-1.4.0/
+	helm upgrade --install --atomic --set webhook.create=false flink-kubernetes-operator flink-operator-repo/flink-kubernetes-operator
 	kubectl apply -f "https://strimzi.io/install/latest?namespace=kafka" -n kafka
 	kubectl apply -f "https://strimzi.io/examples/latest/kafka/kafka-ephemeral-single.yaml" -n kafka
 	kubectl apply -f ./deploy/dev
 
-deploy:
-	kubectl apply -f ./deploy/
-
 deploy-samples:
 	kubectl apply -f ./deploy/samples
 
-release:
-	./gradlew publish
+deploy-config:
+	kubectl create configmap hoptimator-configmap --from-file=model.yaml=test-model.yaml --dry-run=client -o yaml | kubectl apply -f -
 
 generate-models:
 	./models/generate-models.sh
 
-.PHONY: build clean quickstart deploy-dev-environment deploy deploy-samples integration-tests bounce generate-models
+release:
+	./gradlew publish
+
+.PHONY: build clean quickstart deploy-dev-environment deploy deploy-samples deploy-config integration-tests bounce generate-models release

bin/hoptimator

@@ -1,5 +1,3 @@
 #!/bin/sh
 
-docker build . -t hoptimator
-kubectl exec -it hoptimator -c hoptimator -- ./hoptimator -n "" -p "" -u "jdbc:calcite:model=./test-model.yaml" "$@"
-
+kubectl exec -it hoptimator -c hoptimator -- ./hoptimator -n "" -p "" -u "jdbc:calcite:model=/etc/config/model.yaml" "$@"

deploy/dev/mysql.yaml

@@ -0,0 +1,44 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: mysql
+  namespace: mysql
+spec:
+  ports:
+  - port: 3306
+  selector:
+    app: mysql
+  clusterIP: None
+
+---
+
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: mysql
+  namespace: mysql
+spec:
+  selector:
+    matchLabels:
+      app: mysql
+  strategy:
+    type: Recreate
+  template:
+    metadata:
+      labels:
+        app: mysql
+    spec:
+      containers:
+      - image: quay.io/debezium/example-mysql:2.2
+        name: mysql
+        env:
+        - name: MYSQL_ROOT_PASSWORD
+          value: debezium
+        - name: MYSQL_USER
+          value: mysqluser
+        - name: MYSQL_PASSWORD
+          value: mysqlpw
+        ports:
+        - containerPort: 3306
+          name: mysql
+

deploy/hoptimator-operator-deployment.yaml

@@ -18,5 +18,12 @@ spec:
       - name: hoptimator-operator
         image: docker.io/library/hoptimator
         imagePullPolicy: Never
-        command: ["./hoptimator-operator"]
+        command: ["./hoptimator-operator", "/etc/config/model.yaml"]
+        volumeMounts:
+        - name: config-volume
+          mountPath: /etc/config
+      volumes:
+      - name: config-volume
+        configMap:
+          name: hoptimator-configmap
 

deploy/hoptimator-pod.yaml

@@ -8,8 +8,14 @@ spec:
     image: docker.io/library/hoptimator
     imagePullPolicy: Never
     command: ["sleep", "infinity"]
+    volumeMounts:
+    - name: config-volume
+      mountPath: /etc/config
     readinessProbe:
       exec:
         command: ["./hoptimator", "--run=./readiness-probe.sql"]
       timeoutSeconds: 30
-
+  volumes:
+  - name: config-volume
+    configMap:
+      name: hoptimator-configmap

deploy/samples/subscriptions.yaml

@@ -1,29 +1,29 @@
 apiVersion: hoptimator.linkedin.com/v1alpha1
 kind: Subscription
 metadata:
-  name: people
+  name: ceos
 spec:
-  sql: SELECT NAME FROM DATAGEN.PERSON
+  sql: SELECT CEO, NAME AS KEY FROM DATAGEN.COMPANY
   database: RAWKAFKA
 
 ---
 
 apiVersion: hoptimator.linkedin.com/v1alpha1
 kind: Subscription
 metadata:
-  name: companies
+  name: companies 
 spec:
-  sql: SELECT NAME FROM DATAGEN.COMPANY
+  sql: SELECT KEY AS PAYLOAD, PAYLOAD AS KEY FROM RAWKAFKA."ceos"
   database: RAWKAFKA
 
 ---
 
 apiVersion: hoptimator.linkedin.com/v1alpha1
 kind: Subscription
 metadata:
-  name: names
+  name: products
 spec:
-  sql: SELECT p.PAYLOAD || c.PAYLOAD FROM RAWKAFKA."people" p, RAWKAFKA."companies" c
+  sql: SELECT "quantity", "product_id" AS KEY FROM INVENTORY."products_on_hand"
   database: RAWKAFKA
 
 

etc/integration-tests.sql

@@ -1,5 +1,5 @@
 
-!connect "jdbc:calcite:model=./test-model.yaml" "" ""
+!connect "jdbc:calcite:model=./model.yaml" "" ""
 
 !set maxWidth 80
 !table

etc/model.yaml

@@ -0,0 +1,10 @@
+# A default model with only built-in adapters
+
+version: 1.0
+defaultSchema: DATAGEN
+schemas:
+
+- name: DATAGEN
+  type: custom
+  factory: com.linkedin.hoptimator.catalog.builtin.DatagenSchemaFactory
+

etc/readiness-probe.sql

@@ -1,5 +1,5 @@
 
-!connect "jdbc:calcite:model=./test-model.yaml" "" ""
+!connect "jdbc:calcite:model=./model.yaml" "" ""
 
 !set maxWidth 80
 !table