Connect example

.gitignore

@@ -7,4 +7,5 @@ ca.pem
 intermediate.cert.pem
 pki_intermediate.csr
 pki-ca-root.json
-cert_key_list
+cert_key_list
+nohup.out

examples/basic/README.md

@@ -0,0 +1,12 @@
+# Basic Deployment
+This example deploys a basic deployment.  No RBAC/LDAP.  Just a single topic 'foobar' is added as part of the pipeline.
+### Deploy CRDs
+Deploy the CRDS using the standard way:
+```shell
+kubectl apply -k ../../kustomize/crds
+```
+### Deploy Confluent Operator and Confluent Services
+Deploy the confluent operator and services:
+```shell
+kubectl apply -k .
+```

examples/basic/confluent/kafka-connect.yaml

@@ -5,7 +5,7 @@ metadata:
   name: connect
 spec:
   tls:
-    secretRef: tls-group1
+    autoGeneratedCerts: true
   dependencies:
     kafka:
       bootstrapEndpoint: kafka.sandbox.svc.cluster.local:9071

examples/basic/confluent/kustomization.yaml

@@ -1,8 +1,7 @@
 namespace: sandbox
 resources:
-- ./namespace.yaml
+- namespace.yaml
 - ../../../kustomize/base/confluent
-- ../../../kustomize/base/secrets-tls
 patchesStrategicMerge:
   - zookeeper.yaml
   - kafka.yaml

examples/basic/operator/kustomization.yaml

@@ -2,5 +2,4 @@ namespace: sandbox
 resources:
 - ../../../kustomize/base/operator
 patchesStrategicMerge:
-  - clusterrolebinding.yaml
   - deployment.yaml

examples/custom-connect-sql/Dockerfile

@@ -0,0 +1,6 @@
+FROM confluentinc/cp-server-connect-operator:6.1.2.1
+USER root
+ENV CONNECT_PLUGIN_PATH="/usr/share/java,/usr/share/confluent-hub-components"
+RUN confluent-hub install --no-prompt debezium/debezium-connector-sqlserver:1.6.0
+RUN confluent-hub install --no-prompt confluentinc/kafka-connect-jdbc:10.1.1
+USER 1001

examples/custom-connect-sql/README.md

@@ -0,0 +1,59 @@
+## Custom Connect Image / Debezium SQL Server
+In this example we go through the following process:
+
+* Build a 'custom' kafka connect image with [Debezium](https://debezium.io/) plug-in, and make available to the internal (minikube) kubernetes cluster
+* Initiate a SQLServer stub populated with the traditional '[AdventureWorks](https://docs.microsoft.com/en-us/sql/samples/adventureworks-install-configure?view=sql-server-ver15&tabs=ssms)' database, and enable [CDC](https://en.wikipedia.org/wiki/Change_data_capture) on select tables
+* Deploy a connector via a cURL command
+* Observe how changes to CDC enabled tables will trigger events in Kafka
+
+NOTE: For ease of readability, we will simply reference the scripts that perform the actions of the following stages.  For better understanding of what is actually being done, please review the scripts themselves which will have their own comments/notations.  **Assumptions are that you will be running all commands from the present directory**
+
+### Building the custom docker image
+The Dockerfile installs a custom plugin with the following line: `RUN confluent-hub install --no-prompt debezium/debezium-connector-sqlserver:1.6.0`.  To build, run:
+
+```shell
+./build-inside.sh
+```
+### Deploy CRDs
+Deploy the CRDS using the standard way:
+```shell
+kubectl apply -k ../../kustomize/crds
+```
+### Deploy Confluent Operator and Confluent Services 
+Deploy the confluent operator and services:
+```shell
+kubectl apply -k .
+```
+### Enable CDC on 'person' table of AdventureWorks Database
+CDC needs to be enabled on a table by table basis.  This table is also referenced in the prod-mssql-connnector.json file. 
+```shell
+./enable_cdc.sh
+Context "minikube" modified.
+Changed database context to 'AdventureWorks'.
+Job 'cdc.AdventureWorks_capture' started successfully.
+Job 'cdc.AdventureWorks_cleanup' started successfully.
+```
+### Deploy Debezium Connector
+A curl request is sent to the 'connect pod' to install the connector.
+```shell
+./deploy_connector.sh
+```
+
+At this stage, if you log onto Control Center, you should see a running connector:
+
+![connector](./connect_image.png)
+
+### Update CDC enabled 'Person' table
+Now we will send a SQL Command that will update all users in the person.Person table on the AdventureWorks database:
+
+```shell
+./update_person.sh
+Context "minikube" modified.
+Changed database context to 'AdventureWorks'.
+
+(19972 rows affected)
+```
+If you observe the automatically created topic `adventureworks-connect.Person.Person` you will see the update event messages streaming through
+
+
+![topic_update](./topic_update.png)

examples/custom-connect-sql/build-inside.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+eval $(minikube docker-env)
+docker build -t sandbox-example-connect .

examples/openshift-basic/confluent/control-centre.yaml → examples/custom-connect-sql/confluent/control-centre.yaml

@@ -4,8 +4,6 @@ kind: ControlCenter
 metadata:
   name: controlcenter
 spec:
-  podTemplate:
-    podSecurityContext: {}
   tls:
     autoGeneratedCerts: true
   dependencies:
@@ -18,11 +16,6 @@ spec:
         url:  https://connect.sandbox.svc.cluster.local:8083
         tls:
           enabled: true
-    ksqldb:
-      - name: ksqldb
-        url:  https://ksqldb.sandbox.svc.cluster.local:8088
-        tls:
-          enabled: true
     schemaRegistry:
       url: https://schemaregistry.sandbox.svc.cluster.local:8081
       tls:

examples/custom-connect-sql/confluent/custom-kafka-connect.yaml

@@ -0,0 +1,31 @@
+apiVersion: platform.confluent.io/v1beta1
+kind: Connect
+metadata:
+  name: connect
+spec:
+  replicas: 1
+  image:
+    application: sandbox-example-connect:latest
+    init: confluentinc/cp-init-container-operator:6.1.2.0
+  tls:
+    autoGeneratedCerts: true
+  dependencies:
+    kafka:
+      bootstrapEndpoint: kafka.sandbox.svc.cluster.local:9071
+      tls:
+        enabled: true
+  configOverrides:
+    server:
+      - group.id=debezium-cluster
+      - offset.storage.replication.factor=3
+      - config.storage.replication.factor=3
+      - status.storage.replication.factor=3
+      - config.storage.topic=_confluent-connect-configs
+      - offset.storage.topic=_confluent-connect-offsets
+      - status.storage.topic=_confluent-connect-status
+      - confluent.topic.security.protocol=SASL_SSL
+      - confluent.topic.sasl.mechanism=OAUTHBEARER
+      - confluent.topic.sasl.login.callback.handler.class=io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler
+      - confluent.topic.ssl.truststore.location=/mnt/sslcerts/truststore.jks
+      - confluent.topic.ssl.truststore.password=${file:/mnt/sslcerts/jksPassword.txt:jksPassword}
+

examples/openshift-basic/confluent/kafka.yaml → examples/custom-connect-sql/confluent/kafka.yaml

@@ -3,17 +3,9 @@ kind: Kafka
 metadata:
   name: kafka
 spec:
-  podTemplate:
-    podSecurityContext: {}
-    resources:
-      requests:
-        memory: 256Mi
-        cpu: 100m
-      limits:
-        memory: 1Gi
-        cpu: 500m
   configOverrides:
     server:
+      - auto.create.topics.enable = true
       - confluent.schema.registry.url=https://schemaregistry.sandbox.svc.cluster.local:8081
   tls:
     autoGeneratedCerts: true

examples/openshift-basic/confluent/kustomization.yaml → examples/custom-connect-sql/confluent/kustomization.yaml

@@ -1,11 +1,11 @@
 namespace: sandbox
 resources:
+- ./namespace.yaml
 - ../../../kustomize/base/confluent
 - ../../../kustomize/base/secrets-tls
 patchesStrategicMerge:
+  - custom-kafka-connect.yaml
   - zookeeper.yaml
   - kafka.yaml
   - control-centre.yaml
-  - kafka-connect.yaml
-  - ksqldb.yaml
   - schema-registry.yaml

examples/openshift-basic/confluent/schema-registry.yaml → examples/custom-connect-sql/confluent/schema-registry.yaml

@@ -3,8 +3,6 @@ kind: SchemaRegistry
 metadata:
   name: schemaregistry
 spec:
-  podTemplate:
-    podSecurityContext: {}
   tls:
     autoGeneratedCerts: true
   dependencies:

examples/custom-connect-sql/confluent/zookeeper.yaml

@@ -0,0 +1,7 @@
+apiVersion: platform.confluent.io/v1beta1
+kind: Zookeeper
+metadata:
+  name: zookeeper
+spec:
+  tls:
+    autoGeneratedCerts: true

examples/custom-connect-sql/custom-connect/adventure-works-sql-server.yaml

@@ -0,0 +1,41 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: sql-server
+  namespace: sandbox
+  labels:
+    role: sql-server
+    app: sql-server
+spec:
+  containers:
+    - name: sql-server
+      image:  boysherman/sqlserver-adventureworks:latest
+      ports:
+        - name: sql
+          containerPort: 1433
+      env:
+        - name: ACCEPT_EULA
+          value: "Y"
+        - name: SA_PASSWORD
+          value: "nbBg8G4DkR83Xs"
+        - name: MSSQL_AGENT_ENABLED
+          value: "true"
+  restartPolicy: Always
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: sql-server
+  labels:
+    app: sql-server
+  namespace: sandbox
+spec:
+  ports:
+    - port: 1433
+      name: sql
+  clusterIP: None
+  selector:
+    app: sql-server
+
+
+

examples/custom-connect-sql/custom-connect/kustomization.yaml

@@ -0,0 +1,3 @@
+namespace: sandbox
+resources:
+- adventure-works-sql-server.yaml

examples/custom-connect-sql/deploy_connector.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+export CONNECTOR_CONFIG=$(cat ./prod-mssql-connector.json)
+kubectl exec -i connect-0 -c connect -- curl -k -X PUT -H 'Content-Type:application/json' -d "$CONNECTOR_CONFIG" https://localhost:8083/connectors/debezium-sql-server/config

examples/custom-connect-sql/enable_cdc.sh

@@ -0,0 +1,26 @@
+#!/bin/bash
+kubectl config set-context --current --namespace=sandbox
+kubectl exec -i sql-server -- \
+tee -a /tmp/person.sql > /dev/null <<EOT
+USE AdventureWorks;
+GO
+EXEC sp_changedbowner 'sa'
+GO
+EXEC sys.sp_cdc_enable_db;
+GO
+EXEC sys.sp_cdc_enable_table
+     @source_schema = N'person',
+     @source_name   = N'person',
+     @role_name     = N'MyRole',
+     @filegroup_name = N'Primary',
+     @supports_net_changes = 0
+GO
+EOT
+
+kubectl exec -it sql-server -- \
+/opt/mssql-tools/bin/sqlcmd -S localhost -U sa -P "nbBg8G4DkR83Xs" -i  /tmp/person.sql
+
+
+
+
+

examples/custom-connect-sql/kustomization.yaml

@@ -0,0 +1,8 @@
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+namespace: sandbox
+resources:
+  - confluent
+  - operator
+  - custom-connect
+  - topics

examples/openshift-basic/operator/deployment.yaml → examples/custom-connect-sql/operator/deployment.yaml

@@ -10,8 +10,3 @@ spec:
             - --debug=false
             - --namespaces=sandbox
           name: confluent-operator
-          image: public.ecr.aws/h0s0v2g0/confluennt-operator-mccullya:latest
-      securityContext:
-        fsGroup: 1000610000
-        runAsNonRoot: true
-        runAsUser: 1000610000

examples/openshift-basic/operator/kustomization.yaml → examples/custom-connect-sql/operator/kustomization.yaml

@@ -2,5 +2,4 @@ namespace: sandbox
 resources:
 - ../../../kustomize/base/operator
 patchesStrategicMerge:
-  - clusterrolebinding.yaml
   - deployment.yaml

examples/custom-connect-sql/prod-mssql-connector.json

@@ -0,0 +1,45 @@
+{
+    "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector", 
+
+    "database.hostname": "sql-server.sandbox.svc.cluster.local", 
+    "database.port": "1433", 
+
+    "database.user": "sa", 
+    "database.password": "nbBg8G4DkR83Xs",
+
+    "database.dbname": "AdventureWorks", 
+    "database.server.name": "adventureworks-connect",
+
+    "database.history.kafka.bootstrap.servers": "kafka.sandbox.svc.cluster.local:9071",
+    "database.history.kafka.topic": "adventureworks-connect-history",
+    "database.history.kafka.security.protocol": "SASL_SSL",
+    "database.history.kafka.ssl.truststore.location": "/mnt/sslcerts/truststore.jks",
+    "database.history.kafka.truststore.password": "mystorepassword",
+    "database.history.kafka.sasl.mechanism": "PLAIN",
+
+    "database.history.producer.bootstrap.servers": "kafka.sandbox.svc.cluster.local:9071",
+    "database.history.producer.security.protocol": "SSL",
+    "database.history.producer.ssl.keystore.location": "/mnt/sslcerts/keystore.jks",
+    "database.history.producer.ssl.keystore.password": "mystorepassword",
+    "database.history.producer.ssl.truststore.location": "/mnt/sslcerts/truststore.jks",
+    "database.history.producer.ssl.truststore.password": "mystorepassword",
+    "database.history.producer.sasl.jaas.config": "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"<kafka-user>\" password=\"<kafka-password>\";",
+    "database.history.producer.sasl.mechanism": "PLAIN",
+
+    "database.history.consumer.bootstrap.servers": "kafka.sandbox.svc.cluster.local:9071",
+    "database.history.consumer.security.protocol": "SSL",
+    "database.history.consumer.ssl.keystore.location": "/mnt/sslcerts/keystore.jks",
+    "database.history.consumer.ssl.keystore.password": "mystorepassword",
+    "database.history.consumer.ssl.truststore.location": "/mnt/sslcerts/truststore.jks",
+    "database.history.consumer.ssl.truststore.password": "mystorepassword",
+    "database.history.consumer.sasl.mechanism": "PLAIN",
+
+	"decimal.handling.mode": "double",
+	"time.precision.mode": "connect",
+	"database.history.skip.unparseable.ddl": false,
+	"database.history.store.only.monitored.tables.ddl": false,
+	"table.include.list": "person.Person",
+	"table.ignore.builtin": false,
+	"include.schema.changes": true
+}
+