KnowForgeYashanDB
qianmoQ· 更新于 2026-09-23· 阅读 339 分钟· 0 次阅读Debezium Connector for YashanDB
Table of Contents
Overview
YashanDB is a next-generation database management system independently developed by the Shenzhen Institute of Computing Sciences. Based on classical database theory, it integrates original computing paradigms including bounded computation, approximate computation, parallel scalability, and cross-modal fusion computing. This enables YashanDB to meet the demanding requirements of core industries such as finance, government, and energy for high performance, high concurrency, and high security.
The Debezium YashanDB connector captures and records row-level changes that occur in databases on a YashanDB server, including tables that are added while the connector is running. You can configure the connector to emit change events for specific subsets of schemas and tables, synchronizing the change events to Kafka.
For information about the YashanDB versions that are compatible with this connector, see the Debezium release overview.
Debezium can ingest change events from YashanDB by using the native YStream database package. For more information about YashanDB and YStream, see the YashanDB website.
How the Debezium YashanDB connector works
To optimally configure and run a Debezium YashanDB connector, it is helpful to understand how the connector works.
YStream Mechanism
The Debezium YashanDB connector uses the YashanDB YStream interface to capture changes from the database transaction logs. YStream is the native Change Data Capture (CDC) engine in YashanDB.
The YashanDB connector works as follows:
- The connector connects to the YashanDB database via JDBC.
- The connector establishes a connection to the specified YStream service and uses the YStream client API to obtain committed transaction data in real time.
- The connector first performs an initial snapshot, capturing the current state of the specified tables.
- After the snapshot completes, the connector switches to streaming mode and continuously captures incremental data changes through YStream.
Snapshots
The redo logs on a YashanDB server are typically configured to not retain the complete history of the database. As a result, the Debezium YashanDB connector cannot retrieve the entire history of the database from the logs. To enable the connector to establish a baseline for the current state of the database, the first time that the connector starts, it performs an initial consistent snapshot of the database.
Table 1. Settings for snapshot.mode connector configuration property
Setting Description
always
Perform a snapshot on each connector start. After the snapshot completes, the connector begins to stream event records for subsequent database changes.
initial
The connector performs a database snapshot. After the snapshot completes, the connector begins to stream event records for subsequent database changes.
initial_only
The connector performs a database snapshot and stops before streaming any change event records. The connector does not capture any change events that occur after the snapshot.
no_data
The connector captures the structure of all relevant tables, performing all of the steps described in the default snapshot workflow, except that it does not create READ events to represent the data set at the point of the connector’s start-up.
recovery
Set this option to restore a database schema history topic that is lost or corrupted. After a restart, the connector runs a snapshot that rebuilds the topic from the source tables.
when_needed
After the connector starts, it performs a snapshot only if it detects one of the following circumstances:
- It cannot detect any topic offsets.
- A previously recorded offset specifies a log position that is not available on the server.
Custom snapshotter SPI
To customize snapshot behavior beyond what is available in the standard snapshot modes, you can implement one or more of the Debezium snapshotter SPI interfaces. These interfaces can control whether a snapshot runs, how data is queried, and whether tables are locked.
io.debezium.snapshot.spi.Snapshotter
Controls whether the connector takes a snapshot.
io.debezium.snapshot.spi.SnapshotQuery
Controls how data is queried during a snapshot.
io.debezium.snapshot.spi.SnapshotLock
Controls whether the connector locks tables when taking a snapshot.
io.debezium.snapshot.spi.Snapshotter interface. All built-in snapshot modes implement this interface.
/**
* {@link Snapshotter} is used to determine the following details about the snapshot process:
* <p>
* - Whether a snapshot occurs. <br>
* - Whether streaming continues during the snapshot. <br>
* - Whether the snapshot includes schema (if supported). <br>
* - Whether to snapshot data or schema following an error.
* <p>
* Although Debezium provides many default snapshot modes,
* to provide more advanced functionality, such as partial snapshots,
* you can customize implementation of the interface.
* For more information, see the documentation.
*
*
*
*/
@Incubating
public interface Snapshotter extends Configurable {
/**
* @return the name of the snapshotter.
*
*
*/
String name();
/**
* @param offsetExists is {@code true} when the connector has an offset context (i.e. restarted)
* @param snapshotInProgress is {@code true} when the connector is started, but a snapshot is already in progress
*
* @return {@code true} if the snapshotter should take a data snapshot
*/
boolean shouldSnapshotData(boolean offsetExists, boolean snapshotInProgress);
/**
* @param offsetExists is {@code true} when the connector has an offset context (i.e. restarted)
* @param snapshotInProgress is {@code true} when the connector is started, but a snapshot is already in progress
*
* @return {@code true} if the snapshotter should take a schema snapshot
*/
boolean shouldSnapshotSchema(boolean offsetExists, boolean snapshotInProgress);
/**
* @return {@code true} if the snapshotter should stream after taking a snapshot
*/
boolean shouldStream();
/**
* @return {@code true} whether the schema can be recovered if database schema history is corrupted.
*/
boolean shouldSnapshotOnSchemaError();
/**
* @return {@code true} whether the snapshot should be re-executed when there is a gap in data stream.
*/
boolean shouldSnapshotOnDataError();
/**
*
* @return {@code true} if streaming should resume from the start of the snapshot
* transaction, or {@code false} for when a connector resumes and takes a snapshot,
* streaming should resume from where streaming previously left off.
*/
default boolean shouldStreamEventsStartingFromSnapshot() {
return true;
}
/**
* Lifecycle hook called after the snapshot phase is successful.
*/
default void snapshotCompleted() {
// no operation
}
/**
* Lifecycle hook called after the snapshot phase is aborted.
*/
default void snapshotAborted() {
// no operation
}
}io.debezium.snapshot.spi.SnapshotQuery interface. All built-in snapshot query modes implement this interface.
/**
* {@link SnapshotQuery} is used to determine the query used during a data snapshot
*
*
*/
public interface SnapshotQuery extends Configurable, Service {
/**
* @return the name of the snapshot lock.
*
*
*/
String name();
/**
* Generate a valid query string for the specified table, or an empty {@link Optional}
* to skip snapshotting this table (but that table will still be streamed from)
*
* @param tableId the table to generate a query for
* @param snapshotSelectColumns the columns to be used in the snapshot select based on the column
* include/exclude filters
* @return a valid query string, or none to skip snapshotting this table
*/
Optional<String> snapshotQuery(String tableId, List<String> snapshotSelectColumns);
}io.debezium.snapshot.spi.SnapshotLock interface. All built-in snapshot lock modes implement this interface.
/**
* {@link SnapshotLock} is used to determine the table lock mode used during schema snapshot
*
*
*/
public interface SnapshotLock extends Configurable, Service {
/**
* @return the name of the snapshot lock.
*
*
*/
String name();
/**
* Returns a SQL statement for locking the given table during snapshotting, if required by the specific snapshotter
* implementation.
*/
Optional<String> tableLockingStatement(Duration lockTimeout, String tableId);
}For more information, see snapshot.mode in the table of connector configuration properties.
Topic names
By default, the YashanDB connector writes change events for all INSERT, UPDATE, and DELETE operations that occur in a table to a single Apache Kafka topic that is specific to that table. The connector uses the following convention to name change event topics:
topicPrefix.schemaName.tableName
The following list provides definitions for the components of the default name:
topicPrefix
The topic prefix as specified by the topic.prefix connector configuration property.
schemaName
The name of the schema in which the operation occurred.
tableName
The name of the table in which the operation occurred.
For example, if fulfillment is the server name, inventory is the schema name, and the database contains tables with the names orders, customers, and products, the Debezium YashanDB connector emits events to the following Kafka topics, one for each table in the database:
fulfillment.inventory.orders
fulfillment.inventory.customers
fulfillment.inventory.productsThe connector applies similar naming conventions to label its internal database schema history topics, schema change topics, and transaction metadata topics.
If the default topic name does not meet your requirements, you can configure custom topic names. To configure custom topic names, you specify regular expressions in the logical topic routing SMT. For more information about using the logical topic routing SMT to customize topic naming, see Topic routing.
Schema history topic
When a database client queries a database, the client uses the database’s current schema. However, the database schema can change at any time, which means that the connector must be able to identify what the schema was at the time each insert, update, or delete operation was recorded. Also, a connector cannot necessarily apply the current schema to every event. If an event is relatively old, it’s possible that it was recorded before the current schema was applied.
To ensure correct processing of events that occur after a schema change, YashanDB includes in the log not only the row-level changes that affect the data, but also the DDL statements that are applied to the database. As the connector encounters these DDL statements in the log, it parses them and updates an in-memory representation of each table’s schema. The connector uses this schema representation to identify the structure of the tables at the time of each insert, update, or delete operation and to produce the appropriate change event. In a separate database schema history Kafka topic, the connector records all DDL statements along with the position in the log where each DDL statement appeared.
When the connector restarts after either a crash or a graceful stop, it starts reading the log from a specific position, that is, from a specific point in time. The connector rebuilds the table structures that existed at this point in time by reading the database schema history Kafka topic and parsing all DDL statements up to the point in the log where the connector is starting.
This database schema history topic is for internal connector use only. Optionally, the connector can also emit schema change events to a different topic that is intended for consumer applications.
Schema change topic
You can configure a Debezium YashanDB connector to produce schema change events that describe structural changes that are applied to tables in the database. The connector writes schema change events to a Kafka topic named <serverName>, where serverName is the namespace that is specified in the topic.prefix configuration property.
Debezium emits a new message to the schema change topic whenever it streams data from a new table, or when the structure of the table is altered.
Messages that the connector sends to the schema change topic contain a payload, and, optionally, also contain the schema of the change event message.
The schema for the schema change event has the following elements:
name
The name of the schema change event message.
type
The type of the change event message.
version
The version of the schema. The version is an integer that is incremented each time the schema is changed.
fields
The fields that are included in the change event message.
Example: Schema of the YashanDB connector schema change topic
The following example shows a typical schema in JSON format.
{
"schema": {
"type": "struct",
"fields": [
{
"type": "string",
"optional": false,
"field": "databaseName"
}
],
"optional": false,
"name": "io.debezium.connector.yashandb.SchemaChangeKey",
"version": 1
},
"payload": {
"databaseName": "inventory"
}
}The payload of a schema change event message includes the following elements:
ddl
Provides the SQL CREATE, ALTER, or DROP statement that results in the schema change.
databaseName
The name of the database to which the statements are applied. The value of databaseName serves as the message key.
schemaName
The name of the schema to which the statements are applied.
tableChanges
A structured representation of the entire table schema after the schema change. The tableChanges field contains an array that includes entries for each column of the table. Because the structured representation presents data in JSON or Avro format, consumers can easily read messages without first processing them through a DDL parser.
| When the connector is configured to capture a table, it stores the history of the table’s schema changes not only in the schema change topic, but also in an internal database schema history topic. The internal database schema history topic is for connector use only and it is not intended for direct use by consuming applications. Ensure that applications that require notifications about schema changes consume that information only from the schema change topic. |
|---|
Never partition the database schema history topic. For the database schema history topic to function correctly, it must maintain a consistent, global order of the event records that the connector emits to it.
To ensure that the topic is not split among partitions, set the partition count for the topic by using one of the following methods:
- If you create the database schema history topic manually, specify a partition count of
1. - If you use the Apache Kafka broker to create the database schema history topic automatically, the topic is created, set the value of Kafka
num.partitionsconfiguration option to1.
Example: Message emitted to the YashanDB connector schema change topic
The following example shows a typical schema change message in JSON format. The message contains a logical representation of the table schema.
{
"schema": {
...
},
"payload": {
"source": {
"version": "3.6.3.Final",
"connector": "yashandb",
"name": "server1",
"ts_ms": 1780017300692,
"snapshot": "false",
"db": "",
"sequence": null,
"ts_us": 1780017300692853,
"ts_ns": 1780017300692853000,
"schema": "DEBEZIUM",
"table": "CUSTOMERS",
"txId": "131072047",
"scn": "828249295637925888",
"batch_row_id": 0,
"position_scn": 828249295637925888,
"group_lsn": 3692924,
"group_offset": 220,
"instance_id": "0"
},
"ts_ms": 1780045807728, (1)
"databaseName": "inventory", (2)
"schemaName": "DEBEZIUM", (3)
"ddl": "CREATE TABLE \"DEBEZIUM\".\"CUSTOMERS\" \n ( \"ID\" NUMBER(9,0) NOT NULL ENABLE, \n \"NAME\" VARCHAR2(255) \n )", (4)
"tableChanges": [ (5)
{
"type": "CREATE", (6)
"id": "\"DEBEZIUM\".\"CUSTOMERS\"", (7)
"table": { (8)
"defaultCharsetName": null,
"primaryKeyColumnNames": [ (9)
"ID"
],
"columns": [ (10)
{
"name": "ID",
"jdbcType": 2,
"nativeType": null,
"typeName": "NUMBER",
"typeExpression": "NUMBER",
"charsetName": null,
"length": 9,
"scale": 0,
"position": 1,
"optional": false,
"autoIncremented": false,
"generated": false
},
{
"name": "NAME",
"jdbcType": 12,
"nativeType": null,
"typeName": "VARCHAR2",
"typeExpression": "VARCHAR2",
"charsetName": null,
"length": 255,
"scale": null,
"position": 2,
"optional": true,
"autoIncremented": false,
"generated": false
}
],
"attributes": [ (11)
{
"customAttribute": "attributeValue"
}
]
}
}
]
}
}Table 2. Descriptions of fields in messages emitted to the schema change topic
Item Field name Description
1
ts_ms
Optional field that displays the time at which the connector processed the event. The time is based on the system clock in the JVM running the Kafka Connect task.
2
databaseName
Identifies the database that contains the change.
3
schemaName
Identifies the schema that contains the change.
4
ddl
This field contains the DDL that is responsible for the schema change.
5
tableChanges
An array of one or more items that contain the schema changes generated by a DDL command.
6
type
Describes the kind of change. The type can be set to one of the following values:
CREATE
Table created.
ALTER
Table modified.
DROP
Table deleted.
7
id
Full identifier of the table that was created, altered, or dropped. In the case of a table rename, this identifier is a concatenation of <old>,<new> table names.
8
table
Represents table metadata after the applied change.
9
primaryKeyColumnNames
List of columns that compose the table’s primary key.
10
columns
Metadata for each column in the changed table.
11
attributes
Custom attribute metadata for each table change.
In messages that the connector sends to the schema change topic, the message key is the name of the database that contains the schema change. In the following example, the payload field contains the databaseName key:
{
"schema": {
"type": "struct",
"fields": [
{
"type": "string",
"optional": false,
"field": "databaseName"
}
],
"optional": false,
"name": "io.debezium.connector.yashandb.SchemaChangeKey",
"version": 1
},
"payload": {
"databaseName": "inventory"
}
}Transaction metadata
Debezium can generate events that represent transaction metadata boundaries and that enrich data change event messages.
Limits on when Debezium receives transaction metadata
Debezium registers and receives metadata only for transactions that occur after you deploy the connector. Metadata for transactions that occur before you deploy the connector is not available.
Database transactions are represented by a statement block that is enclosed between the BEGIN and END keywords. Debezium generates transaction boundary events for the BEGIN and END delimiters in every transaction. Transaction boundary events contain the following fields:
status
BEGIN or END.
id
String representation of the unique transaction identifier.
ts_ms
The time of a transaction boundary event (BEGIN or END event) at the data source.
event_count (for END events)
Total number of events emitted by the transaction.
data_collections (for END events)
An array of pairs of data_collection and event_count elements that indicates the number of events that the connector emits for changes that originate from a data collection.
The following example shows a typical transaction boundary message:
Example: YashanDB connector transaction boundary event
{
"status": "BEGIN",
"id": "5.6.641",
"ts_ms": 1486500577125,
"event_count": null,
"data_collections": null
}
{
"status": "END",
"id": "5.6.641",
"ts_ms": 1486500577691,
"event_count": 2,
"data_collections": [
{
"data_collection": "inventory.DEBEZIUM.CUSTOMERS",
"event_count": 1
},
{
"data_collection": "inventory.DEBEZIUM.ORDERS",
"event_count": 1
}
]
}Data change events
Every data change event that the YashanDB connector emits has a key and a value. The structures of the key and value depend on the table from which the change events originate. For information about how Debezium constructs topic names, see Topic names.
The Debezium YashanDB connector ensures that all Kafka Connect schema names are valid Avro schema names. To qualify as a valid Avro schema name, the logical server name must start with an alphabetic character or an underscore ([a-z,A-Z,_]). The remaining characters in the logical server name, and all characters in the schema and table names, must be alphanumeric characters or underscores ([a-z,A-Z,0-9,\_]). The connector automatically replaces invalid characters with an underscore character.
Unexpected naming conflicts can result when the only distinguishing characters between multiple logical server names, schema names, or table names are not valid characters, and those characters are replaced with underscores.
Debezium and Kafka Connect are designed around continuous streams of event messages. However, the structure of these events might change over time, which can be difficult for topic consumers to handle. To facilitate the processing of mutable event structures, each event in Kafka Connect is self-contained. Every message key and value has two parts: a schema and payload. The schema describes the structure of the payload, while the payload contains the actual data.
Change event keys
For each changed table, the change event key is structured such that a field exists for each column in the primary key (or unique key constraint) of the table at the time when the event is created.
For example, consider the following SQL for a customers table that is defined in the inventory database schema:
CREATE TABLE customers (
ID INT NOT NULL PRIMARY KEY,
NAME VARCHAR(255)
);If the value of the <topic.prefix>.transaction configuration property is set to server1, the JSON representation for every change event that occurs in the customers table in the database features the following key structure:
{
"schema": {
"type": "struct",
"fields": [
{
"type": "int32",
"optional": false,
"field": "ID"
}
],
"optional": false,
"name": "server1.inventory.customers.Key"
},
"payload": {
"ID": 1001
}
}The schema portion of the key contains a Kafka Connect schema that describes the content of the key portion. In the preceding example, the payload value is not optional, the structure is defined by a schema named server1.inventory.customers.Key, and there is one required field named ID of type int32. The value of the key’s payload field indicates that it is indeed a structure (which in JSON is just an object) with a single ID field, whose value is 1001.
Therefore, you can interpret this key as describing the row in the inventory.customers table (output from the connector named server1) whose ID primary key column had a value of 1001.
Change event values
The structure of a value in a change event message mirrors the structure of the message key in the change event in the message, and contains both a schema section and a payload section.
Payload of a change event value
An envelope structure in the payload sections of a change event value contains the following fields:
op
A mandatory field that contains a string value describing the type of operation. The op field in the payload of a YashanDB connector change event value contains one of the following values: c (create or insert), u (update), d (delete), or r (read, which indicates a snapshot).
before
An optional field that, if present, describes the state of the row before the event occurred. The structure is described by the server1.inventory.customers.Value Kafka Connect schema, which the server1 connector uses for all rows in the inventory.customers table.
after
An optional field that, if present, contains the state of a row after a change occurs. The structure is described by the same server1.inventory.customers.Value Kafka Connect schema that is used for the before field.
source
A mandatory field that contains a structure that describes the source metadata for the event. In the case of the YashanDB connector, the structure includes the following fields:
- The Debezium version.
- The connector type and name.
- Timestamps (
ts_ms,ts_us,ts_ns) that represent when the connector processed the event, based on the system clock of the JVM running the Kafka Connect task. For snapshots, the timestamp indicates when the snapshot occurred. - Whether the event is part of an ongoing snapshot or not.
- Database and schema names.
- Table name.
- The transaction ID (
txId), not included for snapshots. - Row sequence number within a batch operation such as batch insert (
batch_row_id). - Commit SCN of the transaction to which the logical log entry belongs (
position_scn). - SCN of the log group to which the logical log entry belongs (
group_lsn). - Physical offset of the logical log entry within its log group (
group_offset). - Instance identifier of the instance to which the transaction belongs (
instance_id).
ts_ms
Provides the timestamp in milliseconds.
ts_us
Provides the timestamp in microseconds.
ts_ns
Provides the timestamp in nanoseconds.
Schema of a change event value
The schema portion of the event message’s value contains a schema that describes the envelope structure of the payload and the nested fields within it.
create events
The following example shows the value of a create event value from the customers table that is described in the change event keys example:
{
"schema": {
"type": "struct",
"fields": [
{
"type": "struct",
"fields": [
{
"type": "int32",
"optional": false,
"field": "ID"
},
{
"type": "string",
"optional": true,
"field": "NAME"
}
],
"optional": true,
"name": "server1.inventory.customers.Value",
"field": "before"
},
{
"type": "struct",
"fields": [
{
"type": "int32",
"optional": false,
"field": "ID"
},
{
"type": "string",
"optional": true,
"field": "NAME"
}
],
"optional": true,
"name": "server1.inventory.customers.Value",
"field": "after"
},
{
"type": "struct",
"fields": [
{
"type": "string",
"optional": true,
"field": "version"
},
{
"type": "string",
"optional": false,
"field": "connector"
},
{
"type": "string",
"optional": false,
"field": "name"
},
{
"type": "int64",
"optional": true,
"field": "ts_ms"
},
{
"type": "string",
"optional": true,
"field": "snapshot"
},
{
"type": "string",
"optional": true,
"field": "db"
},
{
"type": "string",
"optional": true,
"field": "sequence"
},
{
"type": "int64",
"optional": true,
"field": "ts_us"
},
{
"type": "int64",
"optional": true,
"field": "ts_ns"
},
{
"type": "string",
"optional": true,
"field": "schema"
},
{
"type": "string",
"optional": true,
"field": "table"
},
{
"type": "string",
"optional": true,
"field": "txId"
},
{
"type": "int64",
"optional": true,
"field": "batch_row_id"
},
{
"type": "int64",
"optional": true,
"field": "position_scn"
},
{
"type": "int64",
"optional": true,
"field": "group_lsn"
},
{
"type": "int64",
"optional": true,
"field": "group_offset"
},
{
"type": "string",
"optional": true,
"field": "instance_id"
}
],
"optional": false,
"name": "io.debezium.connector.yashandb.Source",
"field": "source"
},
{
"type": "string",
"optional": false,
"field": "op"
},
{
"type": "int64",
"optional": true,
"field": "ts_ms"
},
{
"type": "int64",
"optional": true,
"field": "ts_us"
},
{
"type": "int64",
"optional": true,
"field": "ts_ns"
}
],
"optional": false,
"name": "server1.inventory.customers.Envelope"
},
"payload": {
"before": null,
"after": {
"ID": 1001,
"NAME": "Test Record"
},
"source": {
"version": "3.6.3.Final",
"connector": "yashandb",
"name": "server1",
"ts_ms": 1780017300692,
"snapshot": "false",
"db": "",
"sequence": null,
"ts_us": 1780017300692853,
"ts_ns": 1780017300692853000,
"schema": "DEBEZIUM",
"table": "CUSTOMERS",
"txId": "131072047",
"batch_row_id": 0,
"position_scn": 828249295637925888,
"group_lsn": 3692924,
"group_offset": 220,
"instance_id": "0"
},
"op": "c",
"ts_ms": 1688252618953,
"ts_us": 1688252618953000,
"ts_ns": 1688252618953000000
}
}The following list describes select fields in the value portion of the preceding create event message:
schema
Specifies the schema of the event value. The schema describes the structure of the value’s payload. Every change event that Debezium emits for a table uses the same value schema, as long as the table schema remains unchanged.
name
The schema section can contain multiple name fields. Each name field specifies the schema for a field in the payload of the event value.
server1.inventory.customers.Value is the schema for the payload’s before and after fields. This schema is specific to the customers table.
The schema names of the before and after fields are of the form logicalName.schemaName.tableName.Value. This format ensures that schema names are unique within the database. In environments that use the Avro converter, having unique schema names ensures that the Avro schema for each table in each logical source has its own evolution and history.
"name": "io.debezium.connector.yashandb.Source"
io.debezium.connector.yashandb.Source is the schema for the payload’s source field. This schema is specific to the YashanDB connector. The connector uses it for all events that it generates.
"name": "server1.inventory.customers.Envelope"
Specifies the name of the schema for the overall structure of the payload. The schema name is made up of the following elements:
server1
Specifies the name of the connector that generated this event.
inventory
Specifies the database that contains the table that was changed.
customers
Specifies the table that was changed.
payload
Specifies the actual data for the row that was changed.
Because the JSON representation of an event includes both the schema and the payload portions of the message, it is often larger than the row that it describes. To reduce the size of the messages that the connector streams to Kafka topics, you can use the Avro converter.
op
Specifies the type of operation that caused the connector to generate the event. In this example, c indicates that a create operation was performed resulting in a new row. This field can contain one of the following values:
c | Create a row. |
|---|---|
u | Update a row. |
d | Delete a row. |
r | Read a row (applies to only snapshots). |
ts_ms, ts_us, ts_ns
Displays timestamps in milliseconds, microseconds, and nanoseconds that indicate when the connector processed the event. The time is based on the system clock in the JVM that runs the Kafka Connect task.
before
An optional field that specifies the state of the row before the event occurred. Because the op field in the preceding example is c (create), this change event describes new data, so the value of the before field is null.
after
An optional field that specifies the state of the row after the event occurred. In this example, the after field contains the values of the new row’s ID and NAME columns.
source
Mandatory field that describes the source metadata for the event. This field contains information that you can use to compare this event to other events, with regard to the origin of the events, the order in which the events occurred, and whether events were part of the same transaction. The source metadata provides the following information:
version
Debezium version.
connector
The type of connector.
name
Connector name.
ts_ms, ts_us, ts_ns
Displays timestamps in milliseconds, microseconds, and nanoseconds that indicate when the change was made in the database.
snapshot
Specifies whether the event resulted from a snapshot operation.
db
Name of the database that contains the new row.
schema
Name of the schema that contains the new row.
table
Name of the table that contains the new row.
txId
Transaction identifier.
batch_row_id
Row sequence number within a batch operation such as batch insert.
position_scn
Commit SCN of the transaction to which the logical log entry belongs.
group_lsn
SCN of the log group to which the logical log entry belongs.
group_offset
Physical offset of the logical log entry within its log group.
instance_id
Instance identifier of the instance to which the transaction belongs.
update events
The following example shows an update change event that the connector captures from the same table as the preceding create event.
{
"schema": { ... },
"payload": {
"before": {
"ID": 1001,
"NAME": "Test Record"
},
"after": {
"ID": 1001,
"NAME": "Updated Record"
},
"source": {
"version": "3.6.3.Final",
"connector": "yashandb",
"name": "server1",
"ts_ms": 1780017300692,
"snapshot": "false",
"db": "",
"sequence": null,
"ts_us": 1780017300692853,
"ts_ns": 1780017300692853000,
"schema": "DEBEZIUM",
"table": "CUSTOMERS",
"txId": "131072047",
"batch_row_id": 0,
"position_scn": 828249295637925888,
"group_lsn": 3692924,
"group_offset": 220,
"instance_id": "0"
},
"op": "u",
"ts_ms": 1688252619000,
"ts_us": 1688252619000000,
"ts_ns": 1688252619000000000
}
}The payload has the same structure as the payload of a create (insert) event, but the following values are different:
- The value of the
opfield isu, signifying that this row changed because of an update. - The
beforefield shows the former state of the row with the values that were present before the update database commit. - The
afterfield shows the updated state of the row, with theNAMEvalue now set toUpdated Record. - The structure of the
sourcefield includes the same fields as before, but the values are different, because the connector captured the event from a different position in the log. - The
ts_msfield shows the timestamp that indicates when Debezium processed the event.
The payload section reveals several other useful pieces of information. For example, by comparing the before and after structures, we can determine how a row changed as the result of a commit. The source structure provides information about YashanDB’s record of this change, providing traceability. It also gives us insight into when this event occurred in relation to other events in this topic and in other topics. Did it occur before, after, or as part of the same commit as another event?
delete events
The following example shows a delete event for the table that is shown in the preceding create and update event examples. The schema portion of the delete event is identical to the schema portion for those events.
{
"schema": { ... },
"payload": {
"before": {
"ID": 1001,
"NAME": "Updated Record"
},
"after": null,
"source": {
"version": "3.6.3.Final",
"connector": "yashandb",
"name": "server1",
"ts_ms": 1780017300692,
"snapshot": "false",
"db": "",
"sequence": null,
"ts_us": 1780017300692853,
"ts_ns": 1780017300692853000,
"schema": "DEBEZIUM",
"table": "CUSTOMERS",
"txId": "131072047",
"batch_row_id": 0,
"position_scn": 828249295637925888,
"group_lsn": 3692924,
"group_offset": 220,
"instance_id": "0"
},
"op": "d",
"ts_ms": 1688252620000,
"ts_us": 1688252620000000,
"ts_ns": 1688252620000000000
}
}The payload shows a row with the same key as the preceding create and update events, but contains different values:
- The
opfield isd, signifying that this row was deleted. - The
beforefield contains the values that were present in the row before it was deleted with the database commit. - The
afterfield isnull, indicating that the row no longer exists. - The
sourcefield has the same structure as thesourcefield in create events, but some field values are different. - The
ts_msshows a timestamp that indicates when Debezium processed this event.
The delete event provides consumers with the information that they require to process the removal of this row.
The YashanDB connector’s events are designed to work with Kafka log compaction, which allows for the removal of some older messages as long as at least the most recent message for every key is kept. This allows Kafka to reclaim storage space while ensuring the topic contains a complete dataset and can be used for reloading key-based state.
When a row is deleted, the delete event value shown in the preceding example still works with log compaction, because Kafka is able to remove all earlier messages that use the same key. The message value must be set to null to instruct Kafka to remove all messages that share the same key. To make this possible, by default, the Debezium YashanDB connector always follows a delete event with a special tombstone event that has the same key but a null value.
truncate events
A truncate change event signals that a table has been truncated. The message key is null in this case, the message value looks like this:
{
"schema": { ... },
"payload": {
"before": null,
"after": null,
"source": { (1)
"version": "3.6.3.Final",
"connector": "yashandb",
"name": "my_topic",
"ts_ms": 1780017300692,
"snapshot": "false",
"db": "",
"sequence": null,
"ts_us": 1780017300692853,
"ts_ns": 1780017300692853000,
"schema": "DEBEZIUM",
"table": "CUSTOMERS",
"txId": "131072047",
"batch_row_id": 0,
"position_scn": 828249295637925888,
"group_lsn": 3692924,
"group_offset": 220,
"instance_id": "0"
},
"op": "t", (2)
"ts_ms": 1688252630000, (3)
"ts_us": 1688252630000000, (3)
"ts_ns": 1688252630000000000 (3)
}
}Table 3. Descriptions of truncate event value fields
Item Field name Description
1
source
Mandatory field that describes the source metadata for the event. In a truncate event value, the source field structure is the same as for create, update, and delete events for the same table, provides this metadata:
- Debezium version
- Connector type and name
- Timestamps (
ts_ms,ts_us,ts_ns) for when the change was made in the database - Whether the event is part of a snapshot (always
falsefor truncate events) - Database and schema that contains the table
- Table name
- Transaction identifier (
txId) - Row sequence number within a batch operation such as batch insert (
batch_row_id) - Commit SCN of the transaction to which the logical log entry belongs (
position_scn) - SCN of the log group to which the logical log entry belongs (
group_lsn) - Physical offset of the logical log entry within its log group (
group_offset) - Instance identifier of the instance to which the transaction belongs (
instance_id)
2
op
Mandatory string that describes the type of operation. The op field value is t, signifying that this table was truncated.
3
ts_ms, ts_us, ts_ns
Optional field that displays the time at which the connector processed the event. The time is based on the system clock in the JVM running the Kafka Connect task.
In the source object, ts_ms indicates the time that the change was made in the database. By comparing the value for payload.source.ts_ms with the value for payload.ts_ms, you can determine the lag between the source database update and Debezium.
If a single TRUNCATE operation affects multiple tables, the connector emits one truncate change event record for each truncated table.
| A truncate event represents a change that is made to an entire table and it has no message key. As a result, for topics with multiple partitions, there is no ordering guarantee for the change events (create, update, and so forth), or truncate events that pertain to a table. For example, if a consumer reads events for a table from multiple partitions, it might receive an update event for a table from one partition after it receives a truncate event that deletes all of the data in the table from another partition. Ordering is guaranteed only for topics that use a single partition. |
|---|
If you do not want the connector to capture truncate events, set the skipped.operations option to filter them out.
Data type mapping
When the Debezium YashanDB connector detects that a value in a table row has changed, it emits a change event that represents that change. Each change event record has the same structure as the original table, and the event record contains a field for each column value. The data type of the table column determines how the connector represents the column’s value in the change event field.
For each column in the table, Debezium maps the source data type to a literal type.
Literal types describe how Debezium represents values literally, using one of the following Kafka Connect schema types: INT8, INT16, INT32, INT64, FLOAT32, FLOAT64, BOOLEAN, STRING, BYTES, ARRAY, MAP, STRUCT.
| If a column is not mapped, the connector ignores changes to the column. The connector maps all other columns in the table and generates change events for those columns. If a column is not mapped, it is not included in the change event. |
|---|
Character types
The following table describes how the connector maps character data types.
| YashanDB data type | Literal type (schema type) | Semantic type (schema name) and Notes |
|---|---|---|
CHAR[(M)] | STRING | n/a |
NCHAR[(M)] | STRING | n/a |
VARCHAR[(M)] | STRING | n/a |
NVARCHAR[(M)] | STRING | n/a |
Table 4. Mappings for YashanDB character types
Binary and Character LOB types
The following table describes how the connector maps binary and character large object (LOB) data types.
| YashanDB data type | Literal type (schema type) | Semantic type (schema name) and Notes |
|---|---|---|
BLOB | BYTES | Depending on the setting of the lob.enabled property in the connector configuration, the connector maps LOB values of this type to raw bytes or a base64-encoded string. |
CLOB | STRING | n/a |
NCLOB | STRING | n/a |
RAW | BYTES | Depending on the setting of the lob.enabled property in the connector configuration, the connector maps LOB values of this type to raw bytes or a base64-encoded string. |
Table 5. Mappings for YashanDB binary and character LOB types
Numeric types
The following table describes how the connector maps numeric data types.
Table 6. Mappings for YashanDB numeric types
YashanDB data type Literal type (schema type) Semantic type (schema name) and Notes
TINYINT
INT8
n/a
SMALLINT
INT16
n/a
INT
INT32
n/a
BIGINT
INT64
n/a
FLOAT
FLOAT32
n/a
DOUBLE
FLOAT64
n/a
NUMBER
BYTES / INT8 / INT16 / INT32 / INT64
org.apache.kafka.connect.data.Decimal
Depending on the value of the decimal.handling.mode property, the connector maps NUMBER to a BYTES representation, or one of the INT types. For NUMBER with a negative scale, use decimal.handling.mode=string to avoid serialization issues.
BIT(1)
BOOLEAN
n/a
BIT(n)
BYTES
n/a
BOOLEAN
BOOLEAN
n/a
Temporal types
The following table describes how the connector maps temporal data types. The way that the connector converts temporal types depends on the time.precision.mode configuration property.
Table 7. Mappings for YashanDB temporal types when time.precision.mode is connect
YashanDB data type Literal type (schema type) Semantic type (schema name) and Notes
DATE
INT64
io.debezium.time.Timestamp
Represents the number of milliseconds since the UNIX epoch.
TIME
INT64
io.debezium.time.MicroTime
Represents the number of microseconds past midnight.
TIMESTAMP
INT64
io.debezium.time.MicroTimestamp
Represents the number of microseconds since the UNIX epoch.
INTERVAL YEAR TO MONTH
FLOAT64
io.debezium.time.MicroDuration
Represents the number of months in the interval, expressed as years with decimal precision.
INTERVAL DAY TO SECOND
FLOAT64
io.debezium.time.MicroDuration
Represents the number of microseconds in the interval.
Other types
The following table describes how the connector maps other data types.
| YashanDB data type | Literal type (schema type) | Semantic type (schema name) and Notes |
|---|---|---|
ROWID | STRING | n/a |
Table 8. Mappings for YashanDB other types
LOB handling
YashanDB only supplies column values for CLOB, NCLOB, and BLOB data types, as well as VARCHAR, NVARCHAR, and RAW data types with a size exceeding 32000, if they are explicitly set or changed in a SQL statement. For LOB columns that are not changed, the connector does not include the values of these columns in the change event.
To capture LOB values and serialize them in change events, set the lob.enabled option to true. When LOB handling is enabled, the connector incurs some performance overhead when emitting LOB data.
Decimal handling
You can modify the way that the connector maps NUMBER data types by changing the value of the connector’s decimal.handling.mode configuration property.
When the property is set to its default value of precise, the connector maps these data types to the Kafka Connect org.apache.kafka.connect.data.Decimal logical type.
When the property value is set to double or string, the connector uses alternate mappings.
| YashanDB data type | precise | double | string |
|---|---|---|---|
NUMBER | org.apache.kafka.connect.data.Decimal | FLOAT64 | STRING |
Table 9. Mappings for numeric data types
DATE, TIME, and TIMESTAMP types are mapped to INT64 (timestamp form) by default. If you want to map them to a fixed format string such as yyyy-MM-dd HH:mm:ss.SSSSSS, refer to the Data type conversion section.
Custom converters
By default, the Debezium YashanDB connector provides several CustomConverter implementations specific to YashanDB data types. These custom converters provide alternative mappings for specific data types based on the connector configuration. To add a CustomConverter to the connector, follow the instructions in the Custom Converters documentation.
The Debezium YashanDB connector provides the following custom converters:
TimestampToStringConverter
The TimestampToStringConverter converts TIMESTAMP type data to a string in a customized format.
| Property | Description |
|---|---|
yashandb_timestamp_formatter.type | io.debezium.connector.yashandb.converters.TimestampToStringConverter |
yashandb_timestamp_formatter.format.datetime | Date format, for example: yyyy-MM-dd HH:mm:ss.SSSSSS |
Table 10. Converter configuration
DateToStringConverter
The DateToStringConverter converts DATE type data to a string in a customized format.
| Property | Description |
|---|---|
yashandb_date_formatter.type | io.debezium.connector.yashandb.converters.DateToStringConverter |
yashandb_date_formatter.format.date | Date format, for example: yyyy-MM-dd |
Table 11. Converter configuration
TimeToStringConverter
The TimeToStringConverter converts TIME type data to a string in a customized format.
| Property | Description |
|---|---|
yashandb_time_formatter.type | io.debezium.connector.yashandb.converters.TimeToStringConverter |
yashandb_time_formatter.format.time | Time format, for example: HH:mm:ss.SSSSSS |
Table 12. Converter configuration
Usage example
Specify the following in the configuration:
# Name two converters: yashandb_timestamp_formatter for TIMESTAMP, yashandb_date_formatter for DATE
"converters": "yashandb_timestamp_formatter,yashandb_date_formatter"
# Bind yashandb_timestamp_formatter to TimestampToStringConverter class
"yashandb_timestamp_formatter.type": "io.debezium.connector.yashandb.converters.TimestampToStringConverter"
# Format TIMESTAMP data as yyyy-MM-dd HH:mm:ss.SSSSSS
"yashandb_timestamp_formatter.format.datetime": "yyyy-MM-dd HH:mm:ss.SSSSSS"
# Bind yashandb_date_formatter to DateToStringConverter class
"yashandb_date_formatter.type": "io.debezium.connector.yashandb.converters.DateToStringConverter"
# Format DATE data as yyyy-MM-dd
"yashandb_date_formatter.format.date": "yyyy-MM-dd"Setting up YashanDB
Before deploying and running the Debezium YashanDB connector, adjust the YashanDB database configuration to ensure compatibility.
Schemas excluded from capture
When the Debezium YashanDB connector captures tables, it automatically excludes tables from the following schemas:
SYSMDSYSXA_SYS
To enable the connector to capture changes from a table, the table must use a schema that is not named in the preceding list.
Configure the YStream memory pool
Incremental data depends on YStream to obtain committed data from YashanDB in real time. When you use YashanDB as the source for tasks that include incremental synchronization, you must first allocate a memory pool for YStream on YashanDB:
ALTER SYSTEM SET STREAM_POOL_SIZE = '512M';| Failure to configure this parameter can cause the task to fail. This parameter is a global parameter. For more information, see the YashanDB documentation. |
|---|
Enable supplemental logging
Reading incremental data changes requires supplemental logging to be enabled.
Database-level supplemental logging
To permit the connector to monitor all objects in the database, including new objects, enable database-level supplemental logging in YashanDB:
ALTER DATABASE ADD SUPPLEMENTAL LOG TABLE TYPE (HEAP);
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;Table-level supplemental logging
If you want the connector to monitor only specific tables, enable table-level supplemental logging in YashanDB:
ALTER TABLE tablename ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;| It is important to note that failing to enable supplemental logging, or enabling it incorrectly, can result in data loss or task failure. |
|---|
Creating a user account for the connector
For the Debezium YashanDB connector to capture change events, it must run as a YashanDB user that has specific permissions.
The YashanDB connector user requires the following permissions for normal operation:
-- Create user
CREATE USER username IDENTIFIED BY password;
-- Session permissions
GRANT CREATE SESSION TO username;
GRANT ALTER SESSION TO username;
-- Debezium specific permissions
GRANT SELECT ANY TABLE TO username;
GRANT LOCK ANY TABLE TO username;
GRANT FLASHBACK ANY TABLE TO username;
GRANT SELECT ON V_$DATABASE TO username;
GRANT SELECT ON V_$TRANSACTION TO username;
GRANT SELECT ON V_$YSTREAM_SERVER TO username;
-- YStream permissions
GRANT YSTREAM_CAPTURE TO username;Configure the YStream service
The YashanDB connector requires a YStream service to obtain incremental data. The following are the steps to configure the YStream service.
Create a YStream service
Use the DBMS_YSTREAM_ADM.CREATE procedure to create a YStream service:
DBMS_YSTREAM_ADM.CREATE(
server_name IN VARCHAR(64),
connect_user IN VARCHAR(64) DEFAULT NULL,
start_scn IN BIGINT DEFAULT NULL);The start_scn is obtained by querying SELECT CURRENT_SCN FROM V$DATABASE.
EXEC DBMS_YSTREAM_ADM.CREATE('serverName', 'connect_user', start_scn);Add tables to the YStream service
Use the DBMS_YSTREAM_ADM.ADD_TABLES procedure to add tables to be captured to the YStream service:
DBMS_YSTREAM_ADM.ADD_TABLES(
server_name IN VARCHAR(64),
table_names IN VARCHAR(4096),
schemas IN VARCHAR(4096));| The table names and schemas that you add to the service must be the same as those that the connector is configured to capture. |
|---|
Set YStream service parameters
Prerequisites
- The YStream service is available to configure. Query the
V$YSTREAM_SERVERview to obtain the service status.
Use the DBMS_YSTREAM_ADM.SET_PARAMETER procedure to set parameters for an existing service:
DBMS_YSTREAM_ADM.SET_PARAMETER(
server_name IN VARCHAR(64),
parameter IN VARCHAR(64),
value IN VARCHAR(64));Start the YStream service
Use the DBMS_YSTREAM_ADM.START procedure to start the YStream service:
DBMS_YSTREAM_ADM.START(server_name IN VARCHAR(64));| If you need to connect to YStream using a standby database, you must start the YStream service on the standby database node. |
|---|
Deploying the YashanDB connector
To deploy a Debezium YashanDB connector, you install the Debezium YashanDB connector archive, configure the connector, and start the connector by adding its configuration to Kafka Connect.
Prerequisites
- Apache Kafka and Kafka Connect are installed.
- YashanDB is installed, and is configured to work with the Debezium connector.
Procedure
- Download the Debezium YashanDB connector plug-in archive.
- Extract all files into your Kafka Connect environment.
- Add the directory with the JAR files to Kafka Connect’s
plugin.path. - Restart your Kafka Connect process to pick up the new JAR files.
Next steps
Debezium YashanDB connector configuration
Typically, you register a Debezium YashanDB connector by submitting a JSON request that specifies the configuration properties for the connector. The following example shows a JSON request for registering an instance of the Debezium YashanDB connector with logical name server1 at port 1688:
Example: Debezium YashanDB connector configuration
{
"name": "inventory-connector", (1)
"config": {
"connector.class" : "io.debezium.connector.yashandb.YashanDbConnector", (2)
"database.hostname" : "<YASHANDB_IP_ADDRESS>", (3)
"database.port" : "1688", (4)
"database.user" : "username", (5)
"database.password" : "password", (6)
"database.dbname" : "dbname", (7)
"topic.prefix" : "server1", (8)
"tasks.max" : "1", (9)
"database.ystream.server.name" : "server1", (10)
"schema.history.internal.kafka.bootstrap.servers" : "kafka:9092", (11)
"schema.history.internal.kafka.topic": "schema-changes.inventory" (12)
}
}| 1 | The name that is assigned to the connector when you register it with a Kafka Connect service. |
|---|---|
| 2 | The name of this YashanDB connector class. |
| 3 | The address of the YashanDB instance. |
| 4 | The port number of the YashanDB instance. |
| 5 | The name of the YashanDB user, as specified in Creating a user account for the connector. |
| 6 | The password for the YashanDB user, as specified in Creating a user account for the connector. |
| 7 | The name of the database to capture changes from. |
| 8 | Topic prefix that identifies and provides a namespace for the YashanDB database server from which the connector captures changes. |
| 9 | The maximum number of tasks to create for this connector. |
| 10 | The name of the YStream service that the connector uses to capture changes. |
| 11 | The list of Kafka brokers that this connector uses to write and recover DDL statements to the database schema history topic. |
| 12 | The name of the database schema history topic where the connector writes and recovers DDL statements. This topic is for internal use only and should not be used by consumers. |
For the complete list of the configuration properties that you can set for the Debezium YashanDB connector, see Connector configuration properties.
You can send this configuration with a POST command to a running Kafka Connect service. The service records the configuration and starts a connector task that performs the following operations:
- Connects to the YashanDB database.
- Reads the redo logs of the database.
- Emits change events for every operation that occurs in the tables that you specify.
- Streams change event records to Kafka topics.
Adding connector configuration
To start running a Debezium YashanDB connector, create a connector configuration, and add the configuration to your Kafka Connect cluster.
Prerequisites
- YashanDB is configured for use with Debezium.
- The Debezium YashanDB connector is installed.
Procedure
- Create a configuration for the YashanDB connector.
- Use the Kafka Connect REST API to add that connector configuration to your Kafka Connect cluster.
Results
After the connector starts, it performs a consistent snapshot of the YashanDB databases that the connector is configured for. The connector then starts generating data change events for row-level operations and streaming the change event records to Kafka topics.
Connector properties
The Debezium YashanDB connector has numerous configuration properties that you can use to achieve the right connector behavior for your application. Many properties have default values. Information about the properties is organized as follows:
Required Debezium YashanDB connector configuration properties
Debezium YashanDB connector database schema history configuration properties
- Pass-through YashanDB connector configuration properties
- Pass-through properties for configuring how producer and consumer clients interact with schema history topics
- Pass-through properties for configuring how the YashanDB connector interacts with the Kafka signaling topic
- Pass-through properties for configuring the Kafka consumer client for the signaling channel
- Pass-through properties for configuring the YashanDB connector sink notification channel
- Debezium connector pass-through database driver configuration properties
Required Debezium YashanDB connector configuration properties
The Debezium YashanDB connector uses numerous configuration properties to create the connector instance. Descriptions of these properties are organized as follows:
Required configuration properties
Property
Default
Description
No default
Unique name for the connector. You can use the specified name to register a connector only once. Registration fails if you attempt to reuse a connector name. (This property is required by all Kafka Connect connectors.)
No default
The name of the Java class for the connector. Always use the following value for the YashanDB connector:
io.debezium.connector.yashandb.YashanDbConnector
1
The maximum number of tasks to create for this connector. The YashanDB connector always uses a single task and therefore does not use this value, so the default is always acceptable.
No default
IP address or hostname of the YashanDB database server.
No default
Integer port number of the YashanDB database server.
No default
Name of the YashanDB user account that the connector uses to connect to the YashanDB database server.
No default
Password to use when connecting to the YashanDB database server.
No default
The name of the YashanDB database.
No default
The JDBC URL for the YashanDB database. Format: jdbc:yasdb://<host>:1688/<dbname>.
No default
The name of the YStream service on the YashanDB database. Specify the YStream service name created in the "Configure YStream service" step.
No default
Topic prefix that provides a namespace for the YashanDB database server from which the connector captures changes. The value that you set is used as the prefix for all Kafka topic names that the connector emits. Specify a topic prefix that is unique across all connectors in your Debezium environment. Valid characters include alphanumeric characters, hyphens, dots, and underscores.
schema.history.internal.kafka.bootstrap.servers
No default
A list of Kafka brokers that the connector uses to write and recover DDL statements to the database schema history topic.
schema.history.internal.kafka.topic
No default
The name of the database schema history topic where the connector writes and recovers DDL statements. This topic is for internal use only and should not be used by consumers.
Optional configuration properties
Property
Default
Description
128
The size of the built-in blocking queue in the YStream client. Incremental logical logs are obtained directly from this queue.
10
The timeout in seconds for obtaining the next result from the blocking queue.
ystream.client.response.timeout
60
The maximum time in seconds that the YStream server waits for the YStream client to respond.
No default
An optional comma-separated list of regular expressions that match the names of schemas for which you want to capture changes. The connector captures changes for any schema whose name is not included in schema.exclude.list , except for system schemas. By default, changes for all non-system schemas are captured. To match schema names, Debezium applies the regular expressions you specify as anchored regular expressions.
No default
An optional comma-separated list of regular expressions that match the names of schemas for which you do not want to capture changes. The connector captures changes for any schema whose name is not included in schema.exclude.list , except for system schemas.
No default
An optional comma-separated list of regular expressions that match fully-qualified table identifiers for tables to be captured. When this property is set, the connector captures changes only from the specified tables. Each table identifier uses the following format: <schema_name>.<table_name>. By default, the connector monitors every non-system table in each captured database schema.
To match the name of a table, Debezium applies the regular expression that you specify as an anchored regular expression.
No default
An optional comma-separated list of regular expressions that match fully-qualified table identifiers for tables to be excluded from monitoring. The connector captures change events from any table that is not specified in the exclude list. Specify each table identifier using the following format: <schema_name>.<table_name>.
To match the name of a table, Debezium applies the regular expression that you specify as an anchored regular expression.
2048
A positive integer value that specifies the maximum size of each batch of events to be processed during each iteration of this connector.
8192
A positive integer value that specifies the maximum number of records that the blocking queue can hold. When Debezium reads the stream of events from the database, it places them in a blocking queue before writing the events to Kafka. The blocking queue prevents data loss in situations where the connector receives messages faster than it can write them to Kafka, or when Kafka is unavailable.
0 (disabled)
A long integer value that specifies the maximum capacity of the blocking queue in bytes. By default, no volume limit is specified for the blocking queue. To specify the number of bytes the queue can consume, set this property to a positive long value. If max.queue.size is also set, writes to the queue are blocked when the queue size reaches the limit specified by either property.
500 (0.5 seconds)
A positive integer value that specifies the number of milliseconds the connector should wait for new change events to appear during each iteration.
t
A comma-separated list of operation types that you want the connector to skip during streaming. You can configure the connector to skip the following types of operations: c (insert/create), u (update), d (delete), t (truncate). By default, only truncate operations are skipped.
initial
Specifies the mode that the connector uses to perform a snapshot of the captured tables. You can set the following values:
always
Perform a snapshot every time the connector starts.
initial
Perform an initial snapshot when the connector starts.
initial_only
Perform only the initial snapshot, without streaming subsequent changes.
no_data
Capture only table structures, not data.
recovery
Restore a lost or corrupted schema history topic.
when_needed
Perform a snapshot only when needed.
10000
Specifies the maximum number of rows to read from each table at a time during a snapshot. The connector reads table contents in multiple batches of the specified size.
1
Specifies the number of threads the connector uses when performing the initial snapshot. To enable parallel initial snapshots, set the property to a value greater than 1. In a parallel initial snapshot, the connector processes multiple tables simultaneously.
This feature is incubating and is subject to change.
false
Specifies whether the parallel initial snapshot uses the legacy table-per-thread algorithm. When set to false (the default), the connector will chunk the contents of the source table across all parallel initial snapshot threads for maximum performance. When set to true, the connector will process each table per thread.
No default
Specifies whether the connector uses locking mode to prevent DDL changes before synchronizing snapshot data. Set one of the following values:
none | No locks are acquired. |
|---|---|
shared | Shared locks are acquired. |
false
Specifies whether large object (CLOB or BLOB, and so on.) column values are emitted in change events. By default, change events have large object columns, but these columns do not contain values. There is some overhead when processing and managing large object column types and payloads. To capture large object values and serialize them in change events, set this option to true.
precise
Specifies how the connector should handle floating-point values for NUMBER columns. You can set one of the following options:
precise
(default) Use java.math.BigDecimal to represent values precisely. Represented in binary form in change events.
double
Use FLOAT64 (double-precision floating-point) to represent values.
string
Use STRING to represent values.
__debezium_unavailable_value
Specifies the constant provided by the connector to indicate that the original value was not changed and is not provided by the database. For example, if a LOB fails to be retrieved, this placeholder is used instead.
No default
The fully qualified name of the data collection used to send signals to the connector. Specify the collection name using the following format: <databaseName>.<schemaName>.<tableName>.
source
The list of signal channel names enabled for the connector. By default, the following channels are available: source, kafka, file, jmx.
No default
The list of notification channel names enabled for the connector. By default, the following channels are available: sink, log, jmx.
incremental.snapshot.chunk.size
1024
The maximum number of rows that the connector fetches and reads into memory during an incremental snapshot chunk. Increasing the chunk size can provide higher efficiency because the snapshot runs fewer snapshot queries, but the queries are larger. However, larger chunk sizes also require more memory to buffer snapshot data. Tune the chunk size to a value that provides optimal performance in your environment.
io.debezium.schema.SchemaTopicNamingStrategy
The name of the class used to determine topic names for data changes, schema changes, transactions, heartbeat events, and so on. Defaults to SchemaTopicNamingStrategy.
.
Specifies the delimiter for topic names. Defaults to a dot (.).
No default
Enumerates a comma-separated list of the symbolic names of the custom converter instances that the connector can use. This property is required to enable the connector to use a custom converter.
For each converter that you configure for a connector, you must also add a .type property, which specifies the fully-qualified name of the class that implements the converter interface.
No default
Configures the class name of a custom converter for Debezium.
No default
Configuration for custom converters, set according to how the converter is used.
10000
The fetch size for JDBC queries.
ddl.parse.fail.retry.read.table
false
After incremental DDL parsing fails, fully read the source table structure for analysis when processing DML events. This property takes effect when both schema.history.internal.skip.unparseable.ddl and ddl.parse.fail.retry.read.table are set to true.
No default
The name of the class that handles schema changes for the connector. For YashanDB, use io.debezium.relational.history.KafkaSchemaHistory.
schema.history.internal.store.only.captured.tables.ddl
false
Specifies whether the connector records the DDL statements for all tables in the database, or only for tables that are captured. Set to true to store DDL only for captured tables.
Debezium YashanDB connector database schema history configuration properties
Debezium provides a set of schema.history.internal.* properties that control how the connector interacts with the schema history topic.
The following table describes the schema.history.internal properties for configuring the Debezium connector.
Table 13. Connector database schema history configuration properties
Property Default Description
schema.history.internal.kafka.topic
No default
The full name of the Kafka topic where the connector stores the database schema history.
schema.history.internal.kafka.bootstrap.servers
No default
A list of host/port pairs that the connector uses for establishing an initial connection to the Kafka cluster. This connection is used for retrieving the database schema history previously stored by the connector, and for writing each DDL statement read from the source database. Each pair should point to the same Kafka cluster used by the Kafka Connect process.
schema.history.internal.kafka.recovery.poll.interval.ms
100
An integer value that specifies the maximum number of milliseconds the connector should wait during startup/recovery while polling for persisted data. The default is 100ms.
schema.history.internal.kafka.query.timeout.ms
3000
An integer value that specifies the maximum number of milliseconds the connector should wait while fetching cluster information using Kafka admin client.
schema.history.internal.kafka.create.timeout.ms
30000
An integer value that specifies the maximum number of milliseconds the connector should wait while create kafka history topic using Kafka admin client.
schema.history.internal.kafka.recovery.attempts
100
The maximum number of times that the connector should try to read persisted history data before the connector recovery fails with an error. The maximum amount of time to wait after receiving no data is recovery.attempts × recovery.poll.interval.ms.
schema.history.internal.skip.unparseable.ddl
false
A Boolean value that specifies whether the connector should ignore malformed or unknown database statements or stop processing so a human can fix the issue. The safe default is false. Skipping should be used only with care as it can lead to data loss or mangling when the binlog is being processed.
schema.history.internal.store.only.captured.tables.ddl
false
A Boolean value that specifies whether the connector records schema structures from all tables in a schema or database, or only from tables that are designated for capture. Specify one of the following values:
false (default)
During a database snapshot, the connector records the schema data for all non-system tables in the database, including tables that are not designated for capture. It’s best to retain the default setting. If you later decide to capture changes from tables that you did not originally designate for capture, the connector can easily begin to capture data from those tables, because their schema structure is already stored in the schema history topic. Debezium requires the schema history of a table so that it can identify the structure that was present at the time that a change event occurred.
true
During a database snapshot, the connector records the table schemas only for the tables from which Debezium captures change events. If you change the default value, and you later configure the connector to capture data from other tables in the database, the connector lacks the schema information that it requires to capture change events from the tables.
schema.history.internal.store.only.captured.databases.ddl
false
A Boolean value that specifies whether the connector records schema structures from all logical databases in the database instance. Specify one of the following values:
true
The connector records schema structures only for tables in the logical database and schema from which Debezium captures change events.
false
The connector records schema structures for all logical databases.
schema.history.internal.memory.optimization
off
Controls how Debezium deduplicates identical schema objects (tables, columns, attributes) in memory using an interner. Specify one of the following values:
off
No deduplication is performed (the default).
on
Each connector uses its own isolated intern pool. Reduces heap usage within a single connector without interfering with other connectors.
shared
All connectors configured with shared share a single global intern pool. Maximises deduplication when many connectors track tables with similar structures.
Pass-through YashanDB connector configuration properties
The connector supports pass-through properties that enable Debezium to specify custom configuration options for fine-tuning the behavior of the Apache Kafka producer and consumer. For information about the full range of configuration properties for Kafka producers and consumers, see the Kafka documentation.
Pass-through properties for configuring how producer and consumer clients interact with schema history topics
Debezium relies on an Apache Kafka producer to write schema changes to database schema history topics. Similarly, it relies on a Kafka consumer to read from database schema history topics when a connector starts. You define the configuration for the Kafka producer and consumer clients by assigning values to a set of pass-through configuration properties that begin with the schema.history.internal.producer.* and schema.history.internal.consumer.* prefixes. The pass-through producer and consumer database schema history properties control a range of behaviors, such as how these clients secure connections with the Kafka broker, as shown in the following example:
schema.history.internal.producer.security.protocol=SSL
schema.history.internal.producer.ssl.keystore.location=/var/private/ssl/kafka.server.keystore.jks
schema.history.internal.producer.ssl.keystore.password=test1234
schema.history.internal.producer.ssl.truststore.location=/var/private/ssl/kafka.server.truststore.jks
schema.history.internal.producer.ssl.truststore.password=test1234
schema.history.internal.producer.ssl.key.password=test1234
schema.history.internal.consumer.security.protocol=SSL
schema.history.internal.consumer.ssl.keystore.location=/var/private/ssl/kafka.server.keystore.jks
schema.history.internal.consumer.ssl.keystore.password=test1234
schema.history.internal.consumer.ssl.truststore.location=/var/private/ssl/kafka.server.truststore.jks
schema.history.internal.consumer.ssl.truststore.password=test1234
schema.history.internal.consumer.ssl.key.password=test1234Debezium strips the prefix from the property name before it passes the property to the Kafka client.
For more information about Kafka producer configuration properties and Kafka consumer configuration properties, see the Apache Kafka documentation .
Pass-through properties for configuring how the YashanDB connector interacts with the Kafka signaling topic
Debezium provides a set of signal.* properties that control how the connector interacts with the Kafka signals topic.
The following table describes the Kafka signal properties.
Table 14. Kafka signals configuration properties
Property Default Description
<topic.prefix>-signal
The name of the Kafka topic that the connector monitors for ad hoc signals.
| If automatic topic creation is disabled, you must manually create the required signaling topic. A signaling topic is required to preserve signal ordering. The signaling topic must have a single partition. |
|---|
kafka-signal
The name of the group ID that is used by Kafka consumers.
signal.kafka.bootstrap.servers
No default
A list of the host and port pairs that the connector uses to establish its initial connection to the Kafka cluster. Each pair references the Kafka cluster that is used by the Debezium Kafka Connect process.
100
An integer value that specifies the maximum number of milliseconds that the connector waits when polling signals.
Pass-through properties for configuring the Kafka consumer client for the signaling channel
The Debezium connector provides for pass-through configuration of the signals Kafka consumer. Pass-through signals properties begin with the prefix signal.consumer.*. For example, the connector passes properties such as signal.consumer.security.protocol=SSL to the Kafka consumer.
Debezium strips the prefixes from the properties before it passes the properties to the Kafka signals consumer.
Pass-through properties for configuring the YashanDB connector sink notification channel
The following table describes properties that you can use to configure the Debezium sink notification channel.
| Property | Default | Description |
|---|---|---|
notification.sink.topic.name | No default | The name of the topic that receives notifications from Debezium. This property is required when you configure the notification.enabled.channels property to include sink as one of the enabled notification channels. |
Table 15. Sink notification configuration properties
Debezium connector pass-through database driver configuration properties
The Debezium connector provides for pass-through configuration of the database driver. Pass-through database properties begin with the prefix driver.*. For example, the connector passes properties such as driver.foobar=false to the JDBC URL.
Debezium strips the prefixes from the properties before it passes the properties to the database driver.
Monitoring
The Debezium YashanDB connector provides three metric types in addition to the built-in support for JMX metrics that Apache Kafka and Kafka Connect have.
- Snapshot metrics for monitoring the connector when performing snapshots
- Streaming metrics for monitoring the connector when processing change events
- Schema history metrics for monitoring the status of the connector’s schema history
For details about how to expose use JMX to expose metrics, see the Debezium monitoring documentation .
Customized MBean names
Debezium connectors expose metrics via the MBean name for the connector. These metrics, which are specific to each connector instance, provide data about the behavior of the connector’s snapshot, streaming, and schema history processes.
By default, when you deploy a correctly configured connector, Debezium generates a unique MBean name for each of the different connector metrics. To view the metrics for a connector process, you configure your observability stack to monitor its MBean. But these default MBean names depend on the connector configuration; configuration changes can result in changes to the MBean names. A change to the MBean name breaks the linkage between the connector instance and the MBean, disrupting monitoring activity. In this scenario, you must reconfigure the observability stack to use the new MBean name if you want to resume monitoring.
To prevent monitoring disruptions that result from MBean name changes, you can configure custom metrics tags. You configure custom metrics by adding the custom.metric.tags property to the connector configuration. The property accepts key-value pairs in which each key represents a tag for the MBean object name, and the corresponding value represents the value of that tag. For example: k1=v1,k2=v2. Debezium appends the specified tags to the MBean name of the connector.
After you configure the custom.metric.tags property for a connector, you can configure the observability stack to retrieve metrics associated with the specified tags. The observability stack then uses the specified tags, rather than the mutable MBean names to uniquely identify connectors. Later, if Debezium redefines how it constructs MBean names, or if the topic.prefix in the connector configuration changes, metrics collection is uninterrupted, because the metrics scrape task uses the specified tag patterns to identify the connector.
A further benefit of using custom tags, is that you can use tags that reflect the architecture of your data pipeline, so that metrics are organized in a way that suits you operational needs. For example, you might specify tags with values that declare the type of connector activity, the application context, or the data source, for example, db1-streaming-for-application-abc. If you specify multiple key-value pairs, all of the specified pairs are appended to the connector’s MBean name.
The following example illustrates how tags modify the default MBean name.
Example 1. How custom tags modify the connector MBean name
By default, the YashanDB connector uses the following MBean name for streaming metrics:
debezium.yashandb:type=connector-metrics,context=streaming,server=<topic.prefix>If you set the value of custom.metric.tags to database=salesdb-streaming,table=inventory, Debezium generates the following custom MBean name:
debezium.yashandb:type=connector-metrics,context=streaming,server=<topic.prefix>,database=salesdb-streaming,table=inventorySnapshot Metrics
The MBean is debezium.yashandb:type=connector-metrics,context=snapshot,server=<topic.prefix>.
The following table lists the JMX metrics that are available for monitoring Debezium snapshot operations, including row counts, table progress, duration, and queue capacity. Snapshot metrics are not exposed unless a snapshot operation is active, or a snapshot has occurred since the last connector start.
| Attributes | Type | Description |
|---|---|---|
LastEvent | string | The last snapshot event that the connector has read. |
MilliSecondsSinceLastEvent | long | The number of milliseconds since the connector has read and processed the most recent event. |
NumberOfErroneousEvents | long | Records the number of change events that the connector identifies as erroneous during a snapshot operation. The connector increments this metric each time that it encounters an event that it cannot process during an initial, incremental, or ad hoc snapshot. Events might fail processing if they are malformed, are incompatible with the schema, or if they encounter failures during transformation. The metric value persists for the lifetime of the connector task. If the snapshot is interrupted, and the connector task restarts, the metric count resets to 0. |
TotalNumberOfEventsSeen | long | The total number of events that this connector has seen since last started or reset. |
NumberOfEventsFiltered | long | The number of events that have been filtered by include/exclude list filtering rules configured on the connector. |
CapturedTables | string[] | The list of tables that are captured by the connector. |
QueueTotalCapacity | int | The length the queue used to pass events between the snapshotter and the main Kafka Connect loop. |
QueueRemainingCapacity | int | The free capacity of the queue used to pass events between the snapshotter and the main Kafka Connect loop. |
TotalTableCount | int | The total number of tables that are being included in the snapshot. |
RemainingTableCount | int | The number of tables that the snapshot has yet to copy. |
SnapshotRunning | boolean | Whether the snapshot was started. |
SnapshotPaused | boolean | Whether the snapshot was paused. |
SnapshotAborted | boolean | Whether the snapshot was aborted. |
SnapshotCompleted | boolean | Whether the snapshot completed. |
SnapshotSkipped | boolean | Whether the snapshot was skipped. |
SnapshotDurationInSeconds | long | The total number of seconds that the snapshot has taken so far, even if not complete. Includes also time when snapshot was paused. |
SnapshotPausedDurationInSeconds | long | The total number of seconds that the snapshot was paused. If the snapshot was paused several times, the paused time adds up. |
RowsScanned | Map<String, Long> | Map containing the number of rows scanned for each table in the snapshot. Tables are incrementally added to the Map during processing. Updates every 10,000 rows scanned and upon completing a table. |
TableChunkCounts | Map<String, Long> | Map containing the number of chunks for each table in the snapshot when using chunk-based multithreaded snapshots. |
TableChunksCompletedCounts | Map<String, Long> | Map containing the number of chunks that have completed for each table in the snapshot when using chunk-based multithreaded snapshots. |
MaxQueueSizeInBytes | long | The maximum buffer of the queue in bytes. This metric is available if max.queue.size.in.bytes is set to a positive long value. |
CurrentQueueSizeInBytes | long | The current volume, in bytes, of records in the queue. |
The following table lists the additional JMX metrics that are available when a connector runs an incremental snapshot, including chunk and table boundary identifiers that you can use to track snapshot progress.
| Attributes | Type | Description |
|---|---|---|
ChunkId | string | The identifier of the current snapshot chunk. |
ChunkFrom | string | The lower bound of the primary key set defining the current chunk. |
ChunkTo | string | The upper bound of the primary key set defining the current chunk. |
TableFrom | string | The lower bound of the primary key set of the currently snapshotted table. |
TableTo | string | The upper bound of the primary key set of the currently snapshotted table. |
Streaming Metrics
The MBean is debezium.yashandb:type=connector-metrics,context=streaming,server=<topic.prefix>.
Common streaming metrics
The following table lists the JMX metrics that are available for monitoring Debezium streaming operations, including event counts by type, lag behind the source, queue capacity, and connection status.
| Attributes | Type | Description |
|---|---|---|
LastEvent | string | The last streaming event that the connector has read. |
MilliSecondsSinceLastEvent | long | The number of milliseconds since the connector has read and processed the most recent event. |
NumberOfErroneousEvents | long | Records the number of change events that the connector identifies as erroneous during streaming. The connector increments this metric each time that it encounters an event that it cannot process during the lifetime of the streaming session. Events might fail processing if they are malformed, are incompatible with the schema, or if they encounter failures during transformation. The metric value persists for the lifetime of the connector task. After a connector restart, the metric count resets to 0. |
TotalNumberOfEventsSeen | long | The total number of data change events reported by the source database since the last connector start, or since a metrics reset. Represents the data change workload for Debezium to process. |
TotalNumberOfCreateEventsSeen | long | The total number of create events processed by the connector since its last start or metrics reset. |
TotalNumberOfUpdateEventsSeen | long | The total number of update events processed by the connector since its last start or metrics reset. |
TotalNumberOfDeleteEventsSeen | long | The total number of delete events processed by the connector since its last start or metrics reset. |
NumberOfEventsFiltered | long | The number of events that have been filtered by include/exclude list filtering rules configured on the connector. |
NumberOfUnchangedEventsSkipped | long | Number of update events skipped since the last connector start or metrics reset because no monitored columns changed. Defaults to -1 if skip.messages.without.change is false. May remain 0 for YashanDB and other connectors that do not support skipping unchanged events, even if the property is set to true. |
CapturedTables | string[] | The list of tables that are captured by the connector. |
QueueTotalCapacity | int | The length the queue used to pass events between the streamer and the main Kafka Connect loop. |
QueueRemainingCapacity | int | The free capacity of the queue used to pass events between the streamer and the main Kafka Connect loop. |
Connected | boolean | Flag that denotes whether the connector is currently connected to the database server. |
MilliSecondsBehindSource | long | The number of milliseconds between the last change event’s timestamp and the connector processing it. The values will incorporate any differences between the clocks on the machines where the database server and the connector are running. |
MilliSecondsBehindSourceMinValue | long | The minimum lag in milliseconds behind the source that is observed during the connector’s runtime. |
MilliSecondsBehindSourceMaxValue | long | The maximum lag in milliseconds behind the source that is observed during the connector’s runtime. |
MilliSecondsBehindSourceAverageValue | double | The average lag in milliseconds behind the source calculated across all observations during the connector’s runtime. |
MilliSecondsBehindSourceP50 | double | The 50th percentile (median) of the lag in milliseconds behind the source. This metric provides a more robust measure of typical lag than the average, as it is less affected by outlier values. Available when statistics.metrics.enabled is set to true (default). |
MilliSecondsBehindSourceP95 | double | The 95th percentile of the lag in milliseconds behind the source. This metric indicates that 95% of lag measurements are below this value, useful for identifying tail latencies and setting SLA thresholds. Available when statistics.metrics.enabled is set to true (default). |
MilliSecondsBehindSourceP99 | double | The 99th percentile of the lag in milliseconds behind the source. This metric indicates that 99% of lag measurements are below this value, useful for understanding worst-case performance scenarios. Available when statistics.metrics.enabled is set to true (default). |
NumberOfCommittedTransactions | long | The number of processed transactions that were committed. |
SourceEventPosition | Map<String, String> | The coordinates of the last received event. |
LastTransactionId | string | Transaction identifier of the last processed transaction. |
MaxQueueSizeInBytes | long | The maximum buffer of the queue in bytes. This metric is available if max.queue.size.in.bytes is set to a positive long value. |
CurrentQueueSizeInBytes | long | The current volume, in bytes, of records in the queue. |
YStream streaming metrics
The Debezium YashanDB connector provides the following additional streaming metrics that are specific to the YStream adapter:
| Attributes | Type | Description |
|---|---|---|
ErrorCount | long | The number of errors detected by the connector during the streaming phase. |
WarningCount | long | The number of warnings detected by the connector during the streaming phase. |
Table 16. Descriptions of YStream-specific streaming metrics
Schema History Metrics
The MBean is debezium.yashandb:type=connector-metrics,context=schema-history,server=<topic.prefix>.
The following table lists the JMX metrics that are available for monitoring the connector’s schema history process, including recovery status, the number of schema changes applied, and timestamps for the most recent changes.
| Attributes | Type | Description |
|---|---|---|
Status | string | One of STOPPED, RECOVERING (recovering history from the storage), RUNNING describing the state of the database schema history. |
RecoveryStartTime | long | The time in epoch seconds at what recovery has started. |
ChangesRecovered | long | The number of changes that were read during recovery phase. |
ChangesApplied | long | the total number of schema changes applied during recovery and runtime. |
MilliSecondsSinceLastRecoveredChange | long | The number of milliseconds that elapsed since the last change was recovered from the history store. |
MilliSecondsSinceLastAppliedChange | long | The number of milliseconds that elapsed since the last change was applied. |
LastRecoveredChange | string | The string representation of the last change recovered from the history store. |
LastAppliedChange | string | The string representation of the last applied change. |
Frequently Asked Questions
Error: YashanDB does not yet have the YStream server 'serverxx' or check option 'database.ystream.server.name' if the parameters are filled in correctly.
The YStream server corresponding to the database.ystream.server.name parameter does not exist in the YashanDB database. Create the relevant YStream server as described in Configure the YStream service.
Error: YashanDB YStream server status is xxx. Please execute 'DBMS_YSTREAM_ADM.START(…)' start YStream server.
The YStream server corresponding to the database.ystream.server.name parameter is not in a running state. Please execute the following command in the database to start the YStream service:
EXEC DBMS_YSTREAM_ADM.START('server1');After Decimal values are synchronized to Kafka, why is the serialized data incorrect?
Debezium performs special handling for Decimals with negative scale. To work around this issue, use the parameter decimal.handling.mode=string.
After DATE/TIME/TIMESTAMP values are synchronized to Kafka, why are they in timestamp form rather than 'yyyy-MM-dd HHss.SSSSSS' form?
By default, Debezium maps temporal types to INT64. For information about how to map temporal types to a fixed format string, configure custom converters, as described in the Data type conversion section.
Does the YashanDB Connector support resuming from a checkpoint? After a task stops or fails, can it capture incremental data from the last committed position?
Yes, the YashanDB Connector supports resuming from a checkpoint. Based on Kafka’s two-phase commit, after a task stops or fails, the last successfully committed log position is recorded in Kafka’s metadata. When the task is resumed, the connector obtains the last committed log position and starts capturing data from that position, ensuring exactly-once data synchronization to Kafka topics.
The YashanDB Connector captures the metadata structure of all tables in the database in the task logs. Is there a way to make the connector only capture the metadata structure for configured tables (schema.include.list and table.include.list)?
By default, Debezium captures the structure of all tables in the database. You can set schema.history.internal.store.only.captured.tables.ddl=true in the task configuration to capture only the structure of configured tables.
After deleting and recreating a task, the table metadata structure is not captured upon restart (for example, "Capturing structure of table" does not appear in the logs). What causes this?
The connector stores table metadata in the Kafka topic specified by the schema.history.internal.kafka.topic property. When you delete and recreate a task with the same name, the connector reads the existing metadata from the schema history topic rather than capturing a new snapshot of the table structures, provided that the schema history topic has not been deleted. To capture the table metadata again, either delete the schema history topic or change the task name.
The documentation states that custom data types, XMLTYPE, and JSON data types are not supported, yet XMLTYPE and JSON data can still be synchronized to Kafka topics. Why?
The type support of the YashanDB Connector depends on the support range of YashanDB YStream. XMLTYPE and JSON data can be synchronized normally, but the correctness of the data cannot be guaranteed and depends on YStream’s support.
A DDL on the YashanDB source caused the task to fail when parsing this DDL. How can I skip this failed DDL?
Debezium provides a feature to skip failed DDL parsing. You can enable this feature by modifying the task configuration to set schema.history.internal.skip.unparseable.ddl=true.
The YashanDB source executed a DDL like "CREATE TABLE … AS SELECT", and subsequent DML data parsing for this table fails. How should this be handled?
This type of DDL cannot be parsed to obtain the table’s metadata, causing subsequent DML data parsing to fail. The Debezium Oracle Connector has the same issue. If you encouter this problem, restart a new synchronization task.
Incremental DDL created using stored procedure packages (such as DBMS_STATS.CREATE_STAT_TABLE) cannot be identified for specific DDL statements. What causes this and how can it be handled?
The connector cannot identify DDL statements created through stored procedure packages as specific DDL statements. As a result metadata mismatches might occur during data synchronization, which can lead to synchronization errors. To avoid this issue, execute DDL statements directly rather than through stored procedure packages.
评论
登录后参与评论