Debezium notifications

Overview

Debezium notifications provide a mechanism to obtain status information about the connector. Notifications can be sent to the following channels:

SinkNotificationChannel

Sends notifications through the Connect API to a configured topic.

LogNotificationChannel

Notifications are appended to the log.

JmxNotificationChannel

Notifications are exposed as an attribute in a JMX bean.

Custom

Notifications are sent to a custom channel that you implement.

Debezium notification format

Notification messages contain the following information:

Available notifications

Debezium notifications deliver information about the progress of initial snapshots or incremental snapshots.

Debezium notifications about the status of an initial snapshot

The following example shows a typical notification that provides the status of an initial snapshot:

{
    "id": "5563ae14-49f8-4579-9641-c1bbc2d76f99",
    "aggregate_type": "Initial Snapshot",
    "type": "COMPLETED", (1)
    "additional_data" : {
        "connector_name": "myConnector"
    },
    "timestamp": "1695817046353"
}

The following table shows examples of the different payloads that might be present in notifications that report the status of initial snapshots:

  {
      “id”:”ff81ba59-15ea-42ae-b5d0-4d74f1f4038f”,
      “aggregate_type”:”Initial Snapshot”,
      “type”:”STARTED”,
      “additional_data”:{
         “connector_name”:”my-connector”
      },
      “timestamp”: “1695817046353”
}

{
   “id”:”6d82a3ec-ba86-4b36-9168-7423b0dd5c1d”,
   “aggregate_type”:”Initial Snapshot”,
   “type”:”IN_PROGRESS”,
   “additional_data”:{
      “connector_name”:”my-connector”,
      “data_collections”:”table1, table2”,
      “current_collection_in_progress”:”table1”
   },
   “timestamp”: “1695817046353”
}

{
   “id”:”6d82a3ec-ba86-4b36-9168-7423b0dd5c1d”,
   “aggregate_type”:”Initial Snapshot”,
   “type”:”TABLE_SCAN_COMPLETED”,
   “additional_data”:{
      “connector_name”:”my-connector”,
      “data_collection”:”table1, table2”,
      “scanned_collection”:”table1”,
      “total_rows_scanned”:”100”,
      “status”:”SUCCEEDED”
   },
   “timestamp”: “1695817046353”
}

  {
      “id”:”ff81ba59-15ea-42ae-b5d0-4d74f1f4038f”,
      “aggregate_type”:”Initial Snapshot”,
      “type”:”COMPLETED”,
      “additional_data”:{
         “connector_name”:”my-connector”
      },
      “timestamp”: “1695817046353”
}

  {
      “id”:”ff81ba59-15ea-42ae-b5d0-4d74f1f4038f”,
      “aggregate_type”:”Initial Snapshot”,
      “type”:”ABORTED”,
      “additional_data”:{
         “connector_name”:”my-connector”
      },
      “timestamp”: “1695817046353”
}

  {
      “id”:”ff81ba59-15ea-42ae-b5d0-4d74f1f4038f”,
      “aggregate_type”:”Initial Snapshot”,
      “type”:”SKIPPED”,
      “additional_data”:{
         “connector_name”:”my-connector”
      },
      “timestamp”: “1695817046353”
}

Debezium notifications about the progress of incremental snapshots

The following table shows examples of the different payloads that might be present in notifications that report the status of incremental snapshots:

  {
      “id”:”ff81ba59-15ea-42ae-b5d0-4d74f1f4038f”,
      “aggregate_type”:”Incremental Snapshot”,
      “type”:”STARTED”,
      “additional_data”:{
         “connector_name”:”my-connector”,
         “data_collections”:”table1, table2”
      },
      “timestamp”: “1695817046353”
}

{
      “id”:”068d07a5-d16b-4c4a-b95f-8ad061a69d51”,
      “aggregate_type”:”Incremental Snapshot”,
      “type”:”PAUSED”,
      “additional_data”:{
         “connector_name”:”my-connector”,
         “data_collections”:”table1, table2”
      },
      “timestamp”: “1695817046353”
}

 {
   “id”:”a9468204-769d-430f-96d2-b0933d4839f3”,
   “aggregate_type”:”Incremental Snapshot”,
   “type”:”RESUMED”,
   “additional_data”:{
      “connector_name”:”my-connector”,
      “data_collections”:”table1, table2”
   },
   “timestamp”: “1695817046353”
}

{
   “id”:”83fb3d6c-190b-4e40-96eb-f8f427bf482c”,
   “aggregate_type”:”Incremental Snapshot”,
   “type”:”ABORTED”,
   “additional_data”:{
      “connector_name”:”my-connector”
   },
   “timestamp”: “1695817046353”
}

{
   “id”:”d02047d6-377f-4a21-a4e9-cb6e817cf744”,
   “aggregate_type”:”Incremental Snapshot”,
   “type”:”IN_PROGRESS”,
   “additional_data”:{
      “connector_name”:”my-connector”,
      “data_collections”:”table1, table2”,
      “current_collection_in_progress”:”table1”,
      “maximum_key”:”100”,
      “last_processed_key”:”50”
   },
   “timestamp”: “1695817046353”
}

{
   “id”:”6d82a3ec-ba86-4b36-9168-7423b0dd5c1d”,
   “aggregate_type”:”Incremental Snapshot”,
   “type”:”TABLE_SCAN_COMPLETED”,
   “additional_data”:{
      “connector_name”:”my-connector”,
      “data_collection”:”table1, table2”,
      “scanned_collection”:”table1”,
      “total_rows_scanned”:”100”,
      “status”:”SUCCEEDED”
   },
   “timestamp”: “1695817046353”
}

{
   “id”:”6d82a3ec-ba86-4b36-9168-7423b0dd5c1d”,
   “aggregate_type”:”Incremental Snapshot”,
   “type”:”COMPLETED”,
   “additional_data”:{
      “connector_name”:”my-connector”
   },
   “timestamp”: “1695817046353”
}

Enabling Debezium notifications

To enable Debezium to emit notifications, specify a list of notification channels by setting the notification.enabled.channels configuration property. By default, the following notification channels are available:

• sink

• log

• jmx

Access to Debezium JMX notifications

To enable Debezium to report events that are exposed through JMX beans, complete the following configuration steps:

1. Enable the JMX MBean Server to expose the notification bean.

2. Add jmx to the notification.enabled.channels property in the connector configuration.

3. Connect your preferred JMX client to the MBean Server.

Notifications are exposed through the Notifications attribute of a bean with the name debezium._<connector-type>_.management.notifications._<server>_.

The following image shows a notification that reports the start of an incremental snapshot:

To discard a notification, call the reset operation on the bean.

The notifications are also exposed as a JMX notification with type debezium.notification. To enable an application to listen for the JMX notifications that an MBean emits, subscribe the application to the notifications.

Custom notification channels

The notification mechanism is designed to be extensible. You can implement channels as needed to deliver notifications in a manner that works best in your environment. Adding a notification channel involves several steps:

1. Create a Java project for the channel to implement the channel, and add Debezium Core as a dependency.

2. Deploy the notification channel.

3. Enable connectors to use the custom notification channel by modifying the connector configuration.

Configuring custom notification channels

Custom notification channels are Java classes that implement the io.debezium.pipeline.notification.channels.NotificationChannel service provider interface (SPI). For example:

public interface NotificationChannel {
    String name(); (1)
    void init(CommonConnectorConfig config); (2)
    void send(Notification notification); (3)
    void close(); (4)
}

Debezium core module dependencies

A custom notification channel Java project has compile dependencies on the Debezium core module. You must include these compile dependencies in your project’s pom.xml file, as shown in the following example:

<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-core</artifactId>
    <version>${version.debezium}</version> (1)
</dependency>

Declare your implementation in the META-INF/services/io.debezium.pipeline.notification.channels.NotificationChannel file.

Deploying a custom notification channel

Prerequisites

• You have a custom notification channel Java program.

Procedure

• To use a notification channel with a Debezium connector, export the Java project to a JAR file, and copy the file to the directory that contains the JAR file for each Debezium connector that you want to use it with.

  For example, in a typical deployment, the Debezium connector files are stored in subdirectories of a Kafka Connect directory (/kafka/connect), with each connector JAR in its own subdirectory (/kafka/connect/debezium-connector-db2, /kafka/connect/debezium-connector-mysql, and so forth). To use a signaling channel with a connector, add the converter JAR file to the connector’s subdirectory.

Configuring connectors to use a custom notification channel

In the connector configuration, add the name of the custom notification channel to the notification.enabled.channels property.