Pulsar
| 功能 | 说明 |
|---|---|
结构迁移 | 如目标端不存在的 Topic , 则自动进行 Topic 创建，并支持设置分区数 |
增量实时同步 | 支持订阅源端 Topic 的消息 |
修改订阅 | 新增、删除、修改订阅 Topic，文档：修改订阅 |
重置位点 | 按 时间戳 回溯位点，重新消费过去一段时间的数据 |
高级功能
| 功能 | 说明 |
|---|---|
Message Format | The following message formats are supported. See Message Format.
|
限制和注意点
| 限制项 | 说明 |
|---|---|
Creating Tables in the Target in Advance | Only support automatic Topic creation for messages. |
前置条件
| 条件 | 说明 |
|---|---|
Port Preparation | Allow the migration and sync node (Worker) to connect to the nodes of Pulsar. |
任务参数
| 参数名称 | 说明 |
|---|---|
schemaFormat | Message format. For more information, see Message Format. |
subscriptionMode | Pulsar subscription mode, including Durable and NonDurable. |
consumeParallel | Degree of consuming Pulsar topics in parallel. |
receiverQueueSize | The size of a Pulsar receiver queue. |
receiverMaxNumBytes | The maximum bytes of a batch received from Pulsar (in bytes). |
receiverTimeout | The timeout period for receiving a batch from Pulsar (in seconds). |
connectionTimeoutMs | The timeout period for connection to Pulsar Client (in milliseconds). |
Tips: 通用参数配置请参考 通用参数及功能
前置条件
| 条件 | 说明 |
|---|---|
Port Preparation | Allow the migration and sync node (Worker) to connect to the nodes of Pulsar. |
任务参数
| 参数名称 | 说明 |
|---|---|
schemaFormat | Message format. For more information, see Message Format. |
batchWriteSize | The maximum data size of a single message. If the size exceeds the limit, the message will be split. |
enableBatching | Enable Pulsar to push messages in batches. |
batchingMaxBytes | The maximum bytes of messages in a batch pushed by Pulsar (in bytes). |
connectionTimeoutMs | The timeout period for connection to Pulsar Client (in milliseconds). |
compressionType | Set compression algorithm. Support LZ4, ZLIB, ZSTD, SNAPPY. |
envelopSchemaInclude | When schemaFormat is set to DEBEZIUM_ENVELOP_JSON_FOR_MQ, it means whether the message body contains schema information. |
Tips: 通用参数配置请参考 通用参数及功能