Add simple MySQL CDC integration test (linkedin#9)

* Add simple MySQL CDC integration test

* Increase integration test timeout

* Log cluste state in integration tests

* Reduce Flink cpu requests

.github/workflows/integration-tests.yml

@@ -34,9 +34,35 @@ jobs:
     - name: Deploy Dev Environment
       run: make deploy-dev-environment
     - name: Deploy Hoptimator
-      run: make deploy deploy-samples
+      run: make deploy
     - name: Wait for Readiness
-      run: kubectl wait pod hoptimator --for condition=Ready --timeout=5m || kubectl describe pod hoptimator
+      run: kubectl wait pod hoptimator --for condition=Ready --timeout=10m
+    - name: Deploy Samples
+      run: make deploy-samples
+    - name: Wait for Flink Jobs
+      run: |
+        i=0
+        while [ $i -ne 10 ]
+        do
+          kubectl wait flinkdeployments --all --for=jsonpath={.status.lifecycleState}=STABLE --timeout=1m && break || sleep 60
+          i=$(($i+1))
+          echo "No stable Flink jobs after $i tries..."
+        done
     - name: Integration Tests
       run: make integration-tests
+    - name: Capture Cluster State
+      if: always()
+      run: |
+        kubectl describe pods
+        kubectl describe deployments
+        kubectl describe kafkas -n kafka
+        kubectl describe flinkdeployments
+        kubectl describe subscriptions
+    - name: Capture Hoptimator Operator Logs
+      if: always()
+      run: kubectl logs $(kubectl get pods -l app=hoptimator-operator -o name)
+    - name: Capture Flink Operator Logs
+      if: always()
+      run: kubectl logs $(kubectl get pods -l app.kubernetes.io/name=flink-kubernetes-operator -o name)
+
 

bin/hoptimator

@@ -1,3 +1,3 @@
 #!/bin/sh
 
-kubectl exec -it hoptimator -c hoptimator -- ./hoptimator -n "" -p "" -u "jdbc:calcite:model=/etc/config/model.yaml" "$@"
+kubectl exec -it hoptimator -c hoptimator -- ./hoptimator --isolation=TRANSACTION_NONE -n "" -p "" -u "jdbc:calcite:model=/etc/config/model.yaml" -nn hoptimator "$@"

deploy/samples/subscriptions.yaml

@@ -1,22 +1,3 @@
-apiVersion: hoptimator.linkedin.com/v1alpha1
-kind: Subscription
-metadata:
-  name: ceos
-spec:
-  sql: SELECT CEO, NAME AS KEY FROM DATAGEN.COMPANY
-  database: RAWKAFKA
-
----
-
-apiVersion: hoptimator.linkedin.com/v1alpha1
-kind: Subscription
-metadata:
-  name: companies 
-spec:
-  sql: SELECT KEY AS PAYLOAD, PAYLOAD AS KEY FROM RAWKAFKA."ceos"
-  database: RAWKAFKA
-
----
 
 apiVersion: hoptimator.linkedin.com/v1alpha1
 kind: Subscription
@@ -26,4 +7,3 @@ spec:
   sql: SELECT "quantity", "product_id" AS KEY FROM INVENTORY."products_on_hand"
   database: RAWKAFKA
 
-

etc/integration-tests.sql

@@ -1,10 +1,15 @@
 
-!connect "jdbc:calcite:model=./model.yaml" "" ""
-
 !set maxWidth 80
 !table
 !schemas
 
+-- built-in bounded tables
 SELECT * FROM DATAGEN.PERSON;
 SELECT * FROM DATAGEN.COMPANY;
 
+-- MySQL CDC tables
+SELECT * FROM INVENTORY."products_on_hand" LIMIT 1;
+
+-- MySQL CDC -> Kafka
+SELECT * FROM RAWKAFKA."products" LIMIT 1;
+

etc/readiness-probe.sql

@@ -1,10 +1,6 @@
 
-!connect "jdbc:calcite:model=./model.yaml" "" ""
-
-!set maxWidth 80
-!table
-!schemas
+!connect "jdbc:calcite:model=/etc/config/model.yaml" "" ""
 
 SELECT * FROM DATAGEN.PERSON;
 SELECT * FROM DATAGEN.COMPANY;
-
+SELECT * FROM INVENTORY."products_on_hand" LIMIT 1;

hoptimator-operator/src/main/resources/FlinkDeployment.yaml.template

@@ -8,16 +8,16 @@ spec:
   imagePullPolicy: Never
   flinkVersion: v1_16
   flinkConfiguration:
-    taskmanager.numberOfTaskSlots: "2"
+    taskmanager.numberOfTaskSlots: "1"
   serviceAccount: flink
   jobManager:
     resource:
       memory: "2048m"
-      cpu: 1
+      cpu: .1
   taskManager:
     resource:
       memory: "2048m"
-      cpu: 1
+      cpu: .1
   job:
     entryClass: com.linkedin.hoptimator.flink.runner.FlinkRunner
     args: