- Release Notes and Announcements
- Product Introduction
- Purchase Guide
- Getting Started
- Operation Guide
- Instance Management
- Creating Instance
- Naming with Consecutive Numeric Suffixes or Designated Pattern String
- Viewing Instance
- Upgrading Instance
- Downgrading Instance Configuration
- Terminating/Returning Instances
- Change from Pay-as-You-Go to Monthly Subscription
- Upgrading Instance Version
- Adding Routing Policy
- Public Network Bandwidth Management
- Connecting to Prometheus
- AZ Migration
- Setting Maintenance Time
- Setting Message Size
- Topic Management
- Consumer Group
- Monitoring and Alarms
- Smart Ops
- Permission Management
- Tag Management
- Querying Message
- Event Center
- Migration to Cloud
- Data Compression
- Instance Management
- CKafka Connector
- Practical Tutorial
- Practical Tutorial of CKafka Client
- Connector Practical Tutorial
- Connecting Flink to CKafka
- Connecting Schema Registry to CKafka
- Connecting Spark Streaming to CKafka
- Connecting Flume to CKafka
- Connecting Kafka Connect to CKafka
- Connecting Storm to CKafka
- Connecting Logstash to CKafka
- Connecting Filebeat to CKafka
- Multi-AZ Deployment
- Log Access
- Replacing Supportive Route (Old)
- Troubleshooting
- API Documentation
- History
- Introduction
- API Category
- Making API Requests
- DataHub APIs
- ACL APIs
- Topic APIs
- BatchModifyGroupOffsets
- BatchModifyTopicAttributes
- CreateConsumer
- CreateDatahubTopic
- CreatePartition
- CreateTopic
- CreateTopicIpWhiteList
- DeleteTopic
- DeleteTopicIpWhiteList
- DescribeDatahubTopic
- DescribeTopic
- DescribeTopicAttributes
- DescribeTopicDetail
- DescribeTopicProduceConnection
- DescribeTopicSubscribeGroup
- FetchMessageByOffset
- FetchMessageListByOffset
- ModifyDatahubTopic
- ModifyTopicAttributes
- DescribeTopicSyncReplica
- Instance APIs
- Route APIs
- Other APIs
- Data Types
- Error Codes
- SDK Documentation
- General References
- Conducting Production and Consumption Pressure Testing on CKafka
- Configuration Guide for Common Parameters in CKafka
- Connecting to Legacy Self-Built Kafka
- Suggestions for CKafka Version Selection
- CKafka Data Reliability Description
- Connector
- FAQs
- Service Level Agreement
- Contact Us
- Glossary
- Release Notes and Announcements
- Product Introduction
- Purchase Guide
- Getting Started
- Operation Guide
- Instance Management
- Creating Instance
- Naming with Consecutive Numeric Suffixes or Designated Pattern String
- Viewing Instance
- Upgrading Instance
- Downgrading Instance Configuration
- Terminating/Returning Instances
- Change from Pay-as-You-Go to Monthly Subscription
- Upgrading Instance Version
- Adding Routing Policy
- Public Network Bandwidth Management
- Connecting to Prometheus
- AZ Migration
- Setting Maintenance Time
- Setting Message Size
- Topic Management
- Consumer Group
- Monitoring and Alarms
- Smart Ops
- Permission Management
- Tag Management
- Querying Message
- Event Center
- Migration to Cloud
- Data Compression
- Instance Management
- CKafka Connector
- Practical Tutorial
- Practical Tutorial of CKafka Client
- Connector Practical Tutorial
- Connecting Flink to CKafka
- Connecting Schema Registry to CKafka
- Connecting Spark Streaming to CKafka
- Connecting Flume to CKafka
- Connecting Kafka Connect to CKafka
- Connecting Storm to CKafka
- Connecting Logstash to CKafka
- Connecting Filebeat to CKafka
- Multi-AZ Deployment
- Log Access
- Replacing Supportive Route (Old)
- Troubleshooting
- API Documentation
- History
- Introduction
- API Category
- Making API Requests
- DataHub APIs
- ACL APIs
- Topic APIs
- BatchModifyGroupOffsets
- BatchModifyTopicAttributes
- CreateConsumer
- CreateDatahubTopic
- CreatePartition
- CreateTopic
- CreateTopicIpWhiteList
- DeleteTopic
- DeleteTopicIpWhiteList
- DescribeDatahubTopic
- DescribeTopic
- DescribeTopicAttributes
- DescribeTopicDetail
- DescribeTopicProduceConnection
- DescribeTopicSubscribeGroup
- FetchMessageByOffset
- FetchMessageListByOffset
- ModifyDatahubTopic
- ModifyTopicAttributes
- DescribeTopicSyncReplica
- Instance APIs
- Route APIs
- Other APIs
- Data Types
- Error Codes
- SDK Documentation
- General References
- Conducting Production and Consumption Pressure Testing on CKafka
- Configuration Guide for Common Parameters in CKafka
- Connecting to Legacy Self-Built Kafka
- Suggestions for CKafka Version Selection
- CKafka Data Reliability Description
- Connector
- FAQs
- Service Level Agreement
- Contact Us
- Glossary
Background
The debezium connector initially synchronizes the existing data of the table only when a connection task is created. Subsequent newly added tables cannot trigger the synchronization of existing data. To support the synchronization of existing data for newly added tables, debezium uses a signal mode to notify the connector to trigger the synchronization of existing data for newly added tables.
Principles
You need to create a signaling table in the subscribed database. When you want to trigger the synchronization of existing data for a new table, insert the relevant information into the signaling table. Additionally, the connector needs to subscribe to this signaling table. When the connector receives a message from the signaling table, the synchronization of existing data for newly added tables will be triggered.
Notes
1. Since the subscription to the signaling table (dip_signal_taskId) messages has been added, the target Topic will include messages from the signaling table, which will require the business side to perform appropriate filtering.
2. Ensure that the user configured in Connection Management has the permissions to create, modify, and delete tables in the database. (This is only necessary for operations on the signaling table.)
3. If the signaling table is used to synchronize existing data for a newly added table, there may be instances of duplicate incremental data. To avoid duplicate data, you can pause the insert, update, and delete operations on the new table, edit the connector task's data source to add the table, and then resume the operations on the new table. Alternatively, you can perform idempotent processing downstream.
4. If you did not select the option to synchronize existing data when the task is created, and later select the option when you modified the data source to add a table with no change messages generated during this period, Debezium will synchronize the existing data of both the old and new tables (not related to the signaling table) by default. If there were incremental change messages in the old table before modifying the data source, then when you modify the data source to add a table and select the existing data synchronization, only the existing data of the new table will be synchronized (triggered by the signaling table).
Yes
No
Was this page helpful?