Chapter 2. Installing Debezium connectors on RHEL

Prerequisites

• A source database is deployed and the Debezium connector can access the database.

• You know the following information, which the connector requires to access the source database:

  • Name or IP address of the database host.
  • Port number for connecting to the database.
  • Name of the account that the connector can use to sign in to the database.
  • Password of the database user account.
  • Name of the database.
  • The names of the tables from which you want the connector to capture information.
  • The name of the Kafka broker to which you want the connector to emit change events.
  • The name of the Kafka topic to which you want the connector to send database history information.

Procedure

• Specify the configuration that you want to apply to the Debezium connector in JSON format.
  

  The following example shows a simple configuration for a Debezium MySQL connector:

  {
    "name": "inventory-connector",  1
    "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector", 2
      "tasks.max": "1",  3
      "database.hostname": "mysql",  4
      "database.port": "3306", 5
      "database.user": "debezium", 6
      "database.password": "dbz", 7
      "database.server.id": "184054",  8
      "topic.prefix": "dbserver1",   9
      "table.include.list": "public.inventory",  10
      "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",  11
      "schema.history.internal.kafka.topic": "dbhistory.inventory"  12
    }
  }

  1

  The name of the connector to register with the Kafka Connect cluster.

  2

  The name of the connector class.

  3

  The number of tasks that can operate concurrently. Only one task should operate at a time.

  4

  The host name or IP address of the host database instance.

  5

  The port number of the database instance.

  6

  The name of the user account through which Debezium connects to the database.

  7

  The password for the database user account.

  8

  A unique numeric ID for the connector.
  This property is used for the MySQL connector only.

  9

  A string that serves as the logical identifier for the database server or cluster of servers from which the connector captures changes. The specified string designates a namespace. Debezium prefixes this name to each Kafka topic that the connector writes to, as well as to the names of Kafka Connect schemas, and the namespaces of the corresponding Avro schema, when the Avro converter is used.

  10

  The list of tables from which the connector captures change events.

  11

  The name of the Kafka broker where the connector sends the database schema history. The specified broker also receives the change events that the connector emits.

  12

  The name of the Kafka topic that stores the schema history.
  After a connector restart, the connector resumes reading the database log from the point at which it stopped, emitting events for any transactions that occurred while it was offline. Before the connector writes change events for an unread transaction to Kafka, it checks the schema history and then applies the schema that was in effect when the original transaction occurred.

Additional information

• For information about the configuration properties that you can set for each type of connector, see the deployment documentation for the connector in the Debezium User Guide.

Prerequisites

• The host environment to which you want to deploy Debezium runs Red Hat Enterprise Linux, AMQ Streams, and Java in a supported configuration.
  

  • For information about how to install AMQ Streams, see Installing AMQ Streams.
  • For information about how to install a basic, non-production AMQ Streams cluster that contains a single ZooKeeper node, and a single Kafka node, see Running a single node AMQ Streams cluster.

• You have administrative privileges (sudo access) on the host.
• Apache ZooKeeper and the Apache Kafka broker are running.
• Kafka Connect is running in distributed mode, and not in standalone mode.
• You know the credentials of the kafka user that was created when AMQ Streams was installed.
• A source database is deployed and the host where you deploy Debezium has access to the database.
• You know how you want to configure the connector.

Additional resources

• Adding connector plugins

Prerequisites

• You deployed a connector on Red Hat Enterprise Linux, based on the instructions in Section 2.3, “Deploying Debezium with AMQ Streams on Red Hat Enterprise Linux”. .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 Integration 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.1.4.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.