Deploying Debezium on OpenShift

Debezium Deployment

For setting up Kafka and Kafka Connect on OpenShift, a set of images provided by the Strimzi project can be used, which offers “Kafka as a Service”. It consists of enterprise grade configuration files and images that bring Kafka to OpenShift.

First, install the operators and templates for the Kafka broker and Kafka Connect into our OpenShift project:

  1. export STRIMZI_VERSION=0.13.0
  2. git clone -b $STRIMZI_VERSION https://github.com/strimzi/strimzi-kafka-operator
  3. cd strimzi-kafka-operator
  5. # We need to create security objects as part of installation so it is necessary to switch to admin user
  6. oc login -u system:admin
  7. oc create -f install/cluster-operator && oc create -f examples/templates/cluster-operator

Next, deploy a Kafka broker cluster and a Kafka Connect cluster and then create a Kafka Connect image with the Debezium connectors installed:

  1. # Deploy an ephemeral single instance Kafka broker
  2. oc process strimzi-ephemeral -p CLUSTER_NAME=broker -p ZOOKEEPER_NODE_COUNT=1 -p KAFKA_NODE_COUNT=1 -p KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 -p KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 | oc apply -f -
  4. # Deploy a single instance of Kafka Connect with no plug-in installed
  5. oc process strimzi-connect-s2i -p CLUSTER_NAME=debezium -p KAFKA_CONNECT_BOOTSTRAP_SERVERS=broker-kafka-bootstrap:9092 -p KAFKA_CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR=1 -p KAFKA_CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR=1 -p KAFKA_CONNECT_STATUS_STORAGE_REPLICATION_FACTOR=1 -p KAFKA_CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE=false -p KAFKA_CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE=false | oc apply -f -
  7. # Build a Debezium image
  8. export DEBEZIUM_VERSION=1.1.2.Final
  9. mkdir -p plugins && cd plugins && \
  10. for PLUGIN in {mongodb,mysql,postgres}; do \
  11.     curl http://central.maven.org/maven2/io/debezium/debezium-connector-$PLUGIN/$DEBEZIUM_VERSION/debezium-connector-$PLUGIN-$DEBEZIUM_VERSION-plugin.tar.gz | tar xz; \
  12. done && \
  13. oc start-build debezium-connect --from-dir=. --follow && \
  14. cd .. && rm -rf plugins

After a while all parts should be up and running:

  1. oc get pods
  3. NAME                                       READY     STATUS        RESTARTS   AGE
  4. broker-entity-operator-5fb7bc8b9b-r86nz    3/3       Running       1          4m
  5. broker-kafka-0                             2/2       Running       0          4m
  6. broker-zookeeper-0                         2/2       Running       0          5m
  7. debezium-connect-3-4sdjr                   1/1       Running       0          1m
  8. strimzi-cluster-operator-d77476b8f-rblqf   1/1       Running       0          5m

Alternatively, you can go to the “Pods” view of your OpenShift Web Console (https://myhost:8443/console/project/myproject/browse/pods) to confirm all pods are up and running:

openshift pods

Verifying the Deployment

Verify whether the deployment is correct by emulating the Debezium Tutorial in the OpenShift environment.

Begin by starting a MySQL server instance that contains some example tables:

  1. # Deploy pre-populated MySQL instance
  2. oc new-app --name=mysql debezium/example-mysql:1.1
  4. # Configure credentials for the database
  5. oc set env dc/mysql MYSQL_ROOT_PASSWORD=debezium  MYSQL_USER=mysqluser MYSQL_PASSWORD=mysqlpw

A new pod with MySQL server should be up and running:

  1. oc get pods
  2. NAME                             READY     STATUS      RESTARTS   AGE
  3. ...
  4. mysql-1-4503l                    1/1       Running     0          2s
  5. mysql-1-deploy                   1/1       Running     0          4s
  6. ...

Register the Debezium MySQL connector to run against the deployed MySQL instance:

  1. oc exec -i -c kafka broker-kafka-0 -- curl -X POST \
  2.     -H "Accept:application/json" \
  3.     -H "Content-Type:application/json" \
  4.     http://debezium-connect-api:8083/connectors -d @- <<'EOF'
  6. {
  7.     "name": "inventory-connector",
  8.     "config": {
  9.         "connector.class": "io.debezium.connector.mysql.MySqlConnector",
  10.         "tasks.max": "1",
  11.         "database.hostname": "mysql",
  12.         "database.port": "3306",
  13.         "database.user": "debezium",
  14.         "database.password": "dbz",
  15.         "database.server.id": "184054",
  16.         "database.server.name": "dbserver1",
  17.         "database.whitelist": "inventory",
  18.         "database.history.kafka.bootstrap.servers": "broker-kafka-bootstrap:9092",
  19.         "database.history.kafka.topic": "schema-changes.inventory"
  20.     }
  21. }
  22. EOF

Kafka Connect’s log file should contain messages regarding execution of the initial snapshot:

  1. oc logs $(oc get pods -o name -l strimzi.io/name=debezium-connect)

Now you can read change events for the customers table from the corresponding Kafka topic:

  1. oc exec -it broker-kafka-0 -- /opt/kafka/bin/kafka-console-consumer.sh \
  2.     --bootstrap-server localhost:9092 \
  3.     --from-beginning \
  4.     --property print.key=true \
  5.     --topic dbserver1.inventory.customers

You should see an output like the following (formatted for the sake of readability):

  1. # Message 1
  2. {
  3.     "id": 1001
  4. }
  6. # Message 1 Value
  7. {
  8.     "before": null,
  9.     "after": {
  10.         "id": 1001,
  11.         "first_name": "Sally",
  12.         "last_name": "Thomas",
  13.         "email": "sally.thomas@acme.com"
  14.     },
  15.     "source": {
  16.         "name": "dbserver1",
  17.         "server_id": 0,
  18.         "ts_sec": 0,
  19.         "gtid": null,
  20.         "file": "mysql-bin.000003",
  21.         "pos": 154,
  22.         "row": 0,
  23.         "snapshot": true,
  24.         "thread": null,
  25.         "db": "inventory",
  26.         "table": "customers"
  27.     },
  28.     "op": "c",
  29.     "ts_ms": 1509530901446
  30. }
  32. # Message 2 Key
  33. {
  34.     "id": 1002
  35. }
  37. # Message 2 Value
  38. {
  39.     "before": null,
  40.     "after": {
  41.         "id": 1002,
  42.         "first_name": "George",
  43.         "last_name": "Bailey",
  44.         "email": "gbailey@foobar.com"
  45.     },
  46.     "source": {
  47.         "name": "dbserver1",
  48.         "server_id": 0,
  49.         "ts_sec": 0,
  50.         "gtid": null,
  51.         "file": "mysql-bin.000003",
  52.         "pos": 154,
  53.         "row": 0,
  54.         "snapshot": true,
  55.         "thread": null,
  56.         "db": "inventory",
  57.         "table": "customers"
  58.     },
  59.     "op": "c",
  60.     "ts_ms": 1509530901446
  61. }
  62. ...

Finally, the next example modifies some records in the customers table of the database:

  1. oc exec -it $(oc get pods -o custom-columns=NAME:.metadata.name --no-headers -l app=mysql) \
  2.     -- bash -c 'mysql -u $MYSQL_USER -p$MYSQL_PASSWORD inventory'
  4. # For example, run UPDATE customers SET email="sally.thomas@example.com" WHERE ID = 1001;

You should now see additional change messages in the consumer started before.

If you have any questions or requests related to running Debezium on OpenShift, let us know via our user group or in the Debezium developer’s chat.