Installing Debezium on RHEL

Procedure

1. Download the Debezium connector or connectors that you want to use from the Software Downloads site. For example, to use Debezium with a MySQL database, download the Debezium 2.7.3 MySQL Connector.

2. On the Red Hat Enterprise Linux host where you deployed Streams for Apache Kafka, open a terminal window and create a connector-plugins directory in /opt/kafka, if it does not already exist:

   $ sudo mkdir /opt/kafka/connector-plugins

3. Enter the following command to extract the contents of the Debezium connector archive that you downloaded to the /opt/kafka/connector-plugins directory.

   $ sudo unzip debezium-connector-mysql-2.7.3.Final.zip -d /opt/kafka/connector-plugins

4. Repeat Steps 1 -3 for each connector that you want to install.

5. From a terminal window, sign in as the kafka user:

   $ su - kafka
   $ Password:

6. Stop the Kafka Connect process if it is running.

   1. Check whether Kafka Connect is running in distributed mode by entering the following command:

      $ jcmd | grep ConnectDistributed

      If the process is running, the command returns the process ID, for example:

      18514 org.apache.kafka.connect.cli.ConnectDistributed /opt/kafka/config/connect-distributed.properties

   2. Stop the process by entering the kill command with the process ID, for example,

      $ kill 18514

7. Edit the connect-distributed.properties file in /opt/kafka/config/ and set the value of plugin.path to the location of the parent directory for the Debezium connector plug-ins:

   plugin.path=/opt/kafka/connector-plugins

8. Start Kafka Connect in distributed mode.

   $ /opt/kafka/bin/connect-distributed.sh /opt/kafka/config/connect-distributed.properties

9. After Kafka Connect is running, use the Kafka Connect API to register the connector.
   Enter a curl command to submit a POST request that sends the connector configuration JSON that you specified in Section 2.2, “Planning the Debezium connector configuration” to the Kafka Connect REST API endpoint at localhost:8083/connectors.
   For example:

   curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors/ -d '{"name": "inventory-connector", "config":
   {"connector.class": "io.debezium.connector.mysql.MySqlConnector",
   "tasks.max": "1",
   "database.hostname": "mysql",
   "database.port": "3306",
   "database.user": "debezium",
   "database.password": "dbz",
   "database.server.id": "184054",
   "topic.prefix": "dbserver1",
   "table.include.list": "public.inventory",
   "schema.history.internal.kafka.bootstrap.servers": "kafka:9092", "schema.history.internal.kafka.topic": "dbhistory.inventory" }
   }'

   To register multiple connectors, submit a separate request for each one.

10. Restart Kafka Connect to implement your changes.

    As Kafka Connect starts, it loads the configured Debezium connectors from the connector-plugins directory.

    After you complete the configuration, the deployed connector connects to the source database and produces events for each inserted, updated, or deleted row or document.

11. Repeat Steps 5–10 for each Kafka Connect worker node.

Procedure

1. From a terminal window on the host, enter the following command to request the list of connectors from the Kafka Connect API:

   $ curl -H "Accept:application/json" localhost:8083/connectors/

   The query returns the name of the deployed connector, for example:

   ["inventory-connector"]

2. From a terminal window on the host, enter the following command to view the tasks that the connector is running:

   $ curl -i -X GET -H "Accept:application/json" localhost:8083/connectors/inventory-connector

   The command returns output that is similar to the following example:

   HTTP/1.1 200 OK
   Date: Thu, 06 Feb 2020 22:12:03 GMT
   Content-Type: application/json
   Content-Length: 531
   Server: Jetty(9.4.20.v20190813)
   
   {
     "name": "inventory-connector",
     ...
     "tasks": [
       {
         "connector": "inventory-connector",
         "task": 0
       }
     ]
   }

3. Display a list of topics in the Kafka cluster.
   From a terminal window, navigate to /opt/kafka/bin/ and run the following shell script:

   ./kafka-topics.sh --bootstrap-server=localhost:9092 --list

   The Kafka broker returns a list of topics that the connector creates. The available topics depends on the settings of the connector’s snapshot.mode,snapshot.include.collection.list, and table.include.list configuration properties. By default, the connector creates a topic for each non-system table in the database.

4. View the contents of a topic.
   From a terminal window, navigate to /opt/kafka/bin/, and run the kafka-console-consumer.sh shell script to display the contents of one of the topics returned by the preceding command:
   

   For example:

   ./kafka-console-consumer.sh \
   >     --bootstrap-server localhost:9092 \
   >     --from-beginning \
   >     --property print.key=true \
   >     --topic=dbserver1.inventory.products_on_hand

   For each event in the topic, the command returns information that is similar to the following output:

   Example 2.1. Content of a Debezium change event

   {"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"}],"optional":false,"name":"dbserver1.inventory.products_on_hand.Key"},"payload":{"product_id":101}}	{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"},{"type":"int32","optional":false,"field":"quantity"}],"optional":true,"name":"dbserver1.inventory.products_on_hand.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"},{"type":"int32","optional":false,"field":"quantity"}],"optional":true,"name":"dbserver1.inventory.products_on_hand.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"string","optional":true,"field":"table"},{"type":"int64","optional":false,"field":"server_id"},{"type":"string","optional":true,"field":"gtid"},{"type":"string","optional":false,"field":"file"},{"type":"int64","optional":false,"field":"pos"},{"type":"int32","optional":false,"field":"row"},{"type":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"query"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"field":"transaction"}],"optional":false,"name":"dbserver1.inventory.products_on_hand.Envelope"},"payload":{"before":null,"after":{"product_id":101,"quantity":3},"source":{"version":"2.7.3.Final-redhat-00001","connector":"mysql","name":"inventory_connector_mysql","ts_ms":1638985247805,"snapshot":"true","db":"inventory","sequence":null,"table":"products_on_hand","server_id":0,"gtid":null,"file":"mysql-bin.000003","pos":156,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1638985247805,"transaction":null}}

   In the preceding example, the payload value shows that the connector snapshot generated a read ("op" ="r") event from the table inventory.products_on_hand. The "before" state of the product_id record is null, indicating that no previous value exists for the record. The "after" state shows a quantity of 3 for the item with product_id 101.

Procedure

1. Download a copy of the Debezium connector plug-in that you want to replace from the Software Downloads.

2. Extract the contents of the Debezium connector archive to the /opt/kafka/connector-plugins directory.

   $ sudo unzip debezium-connector-mysql-2.7.3.Final.zip -d /opt/kafka/connector-plugins

3. Restart Kafka Connect.