tencent cloud

Tencent Cloud DataBuddy

DocumentaçãoTencent Cloud DataBuddyData IntegrationList of Data SourcesKafka Data Sources

Kafka Data Sources

Baixar
Modo Foco
Tamanho da Fonte
Última atualização: 2026-10-09 16:09:22
Traduzido por IA

Overview

Kafka is a distributed stream processing platform. DataBuddy data ingestion supports reading messages from Kafka (including Tencent Cloud CKafka) as a source, covering both offline reading and real-time whole-database scenarios. This document describes how to use it in the order of "Create Data Source → Configure Sync Task → Troubleshoot Common Issues".

Supported Versions

Type
Supported Version
Self-Built Kafka
2.x / 3.x
CKafka
2.4.1 / 2.8.1 / 3.2.3

Read Capability

Capability
Offline read
Real-time full database read
Supported
✓
✓
For a complete comparison of data source capabilities, see Supported Data Sources and Read Capabilities.

Use Limits

Serialization Format Support

Scenario
Read Format
Full database synchronization
canal-JSON / debezium / ogg-json / csv /json /raw
For whole-database synchronization, DDL changes support only adding columns, dropping columns, and creating tables.

Creating a Data Source

Operation Steps

1. Log in to the DataBuddy console.
2. Switch to the target region and Workspace at the top.
3. Go to Data Integration > Task Management > Data Source Management (or Platform Management > Data Source Management).
4. Click Add Data Source and select Kafka.
5. Enter the connection configuration and authentication information. For details, see Parameter Description.
6. Click Test Connection (see Data Source Connectivity Test).
7. After the test passes, click Save or Save & Create Task.

Parameter Description

Parameter
Description
Required
Data Source Name
Unique identifier of the data source within the workspace
Yes
Kafka Type
Self-built open source Kafka / Tencent Cloud CKafka
Yes
Bootstrap Servers
host1:port1,host2:port2 (for example, 10.0.0.1:9092,10.0.0.2:9092)
Yes
Authentication Method
No
Username / Password
Required when SASL authentication is selected (passwords support SSM credential hosting).
Depends on Authentication

Authentication Method

Method
Description
Required field
No Authentication (PLAINTEXT)
Authentication Disabled
-
SASL_PLAINTEXT
Plaintext SASL Authentication
Username and password
SASL_GSSAPI(Kerberos)
Kerberos Authentication
keytab file, conf file, principal, and Kafka service name
SASL_SSL
SASL + SSL Dual Authentication
Truststore certificate file, Keystore password, username, and password
SSL
Mutual TLS Authentication
Truststore certificate file, Truststore password, Keystore certificate file, Keystore password, and private key password

Network and Server Preparation

The integration resource group must have bidirectional connectivity to all brokers.
The listeners / advertised.listeners in the server-side server.properties must be set to externally reachable addresses, which is a common root cause of connection failures.

Using in Data Ingestion Tasks

Real-Time Sync: Source

Parameter
Description
Data Source
Created Kafka data source that supports connectivity tests.
Topic
Topic to consume, supporting multi-select from the dropdown list.
Serialization Format
Read Position
Earliest (earliest) / Latest (latest) / Specified Time
Subscription Type
You can select INSERT / UPDATE / DELETE, and at least one must be selected.

Nested JSON

Nested JSON is not supported by default. To enable it, add the following to the advanced parameters:
source.nested.json.enabled=true
Tip
This parameter applies to both the Kafka key and value. If the key is not JSON (such as a plain string or numeric primary key), append source.key.nested.json.enabled=false. Otherwise, an error will be reported during key parsing.
To reference nested fields, declare them in the parent_field->child_field format. Use multiple -> to connect multiple levels of nesting. When referencing them in a function, enclose the reference in backticks:
TO_TIMESTAMP(`book->published`, 'yyyyMMdd')

Advanced Parameters

Except for the parameters listed in the following table, all native Kafka client parameters can be passed through by adding the properties. prefix, for example, properties.fetch.max.wait.ms=500.
Parameter
Description
Default Value
source.nested.json.enabled
Enable nested JSON (applies to both keys and values).
false
source.key.nested.json.enabled
Whether to enable nested JSON for keys
Follows source.nested.json.enabled
source.ignore-parse-errors
Skip error records when parsing fails.
false
scan.topic-partition-discovery.interval
Interval for periodically discovering new partitions / topics; set to 0 to disable.
5min
timestamp.format
Time field parsing format
ISO8601

Data Format Example

canal-json

{
  "data": [{ "id": "1", "name": "Zhang San", "age": "25" }],
  "database": "test_db",
  "table": "user_info",
  "type": "INSERT",
  "es": 1589373560000,
  "ts": 1589373560798,
  "isDdl": false,
  "mysqlType": { "id": "int", "name": "varchar(50)", "age": "int" },
  "sqlType": { "id": 4, "name": 12, "age": 4 },
  "pkNames": ["id"],
  "old": null,
  "sql": ""
}

debezium

{
  "before": null,
  "after": { "id": 1, "name": "Zhang San", "age": 25 },
  "source": {
    "version": "1.5.0.Final",
    "connector": "mysql",
    "name": "mysql_server",
    "ts_ms": 1589373560000,
    "db": "test_db",
    "table": "user_info"
  },
  "op": "c",
  "ts_ms": 1589373560798
}

FAQs

Q: Connection Failure or Test Connection Timeout?

A: Troubleshoot in the following order:
1. Check whether the integration resource group has bidirectional connectivity to the broker network through the VPC, allowlist, and security group.
2. Check whether the address returned by advertised.listeners is reachable from the network where the resource group resides.
3. Check whether the username, password, and SASL mechanism used for SASL authentication are consistent with those on the server.

Q: No Data Consumed?

Possible cause
How to Confirm
The read position is set to latest, but the Topic has no new messages.
Change to earliest or a specified time point and restart.
The Topic actually has no data.
consume once directly from the Kafka console.
The Consumer group already has a persisted offset externally.
Change to a different group ID and restart.
Data is completely filtered out by the filter operation or database/table scope.
Check the corresponding configuration of the sync task.

Q: Serialization Format Parsing Failure or All Fields null?

A: Troubleshoot in the following order:
1. Use a consumer tool to pull a raw message and compare it with the configured serialization format.
2. The message is nested JSON, but is not enabled source.nested.json.enabled=true.
3. The key is not JSON, but nested JSON is enabled. Add source.key.nested.json.enabled=false.

Q: Missing mysqlType / sqlType / pkNames in the Entire Database?

A: For whole-database synchronization with canal-json / debezium, messages must carry these metadata fields. Confirm that the upstream Canal / Debezium output includes them. You can use a tool to pull a message and verify.

Q: SQL Syntax Error When Nested Field Functions Are Processed?

A: - and > are not valid SQL identifiers. When referencing nested fields in a function, enclose them in backticks and confirm that the fields have been declared through "Add Field":
TO_TIMESTAMP(`book->published`, 'yyyyMMdd')

References

Add Data Source Connection
Data Source Connectivity Test
Supported Data Sources and Read Capabilities
Create a Single-Table Sync Task
Real-Time Whole-Database Multi-Table Task Configuration
Anterior: Data Source OverviewPróximo: MongoDB Data Source

Ajuda e Suporte

Esta página foi útil?

comentários

Search
    Enter keywords to search documentation