Debezium connector for Vitess

Typically, a MySQL server is not configured to retain the complete history of the database in the binary logs. As a result, the connector is unable to read the entire history of the database from the binary logs. For this reason, the first time that the connector starts, it performs an initial consistent snapshot of the database. You can change this behavior by setting the snapshot.mode connector configuration property to a value other than initial. This snapshot feature is built on VStream Copy introduced in version 7.0.

The Vitess connector spends all its time streaming changes from the VTGate’s VStream gRPC service to which it is subscribed. The client receives changes from VStream as they are committed in the underlying MySQL server’s binlog at certain positions, which are referred to as VGTID.

The VGTID in Vitess is the equivalent of GTID in MySQL, it describes the position in the VStream in which a change event happens. Typically, A VGTID has multiple shard GTIDs, each shard GTID is a tuple of (Keyspace, Shard, GTID), which describes the GTID position of a given shard.

When subscribing to a VStream service, the connector needs to provide a VGTID and a Tablet Type (e.g. MASTER, REPLICA). The VGTID describes the position from which VStream should starts sending change events; the Tablet type describes which underlying MySQL instance (master or replica) in each shard do we read change events from.

The first time the connector connects to a Vitess cluster, it gets the current VGTID from a Vitess component called VTCtld and provides the current VGTID to VStream.

The Debezium Vitess connector acts as a gRPC client of VStream. When the connector receives changes it transforms the events into Debezium create, update, or delete events that include the VGTID of the event. The Vitess connector forwards these change events in records to the Kafka Connect framework, which is running in the same process. The Kafka Connect process asynchronously writes the change event records in the same order in which they were generated to the appropriate Kafka topic.

Periodically, Kafka Connect records the most recent offset in another Kafka topic. The offset indicates source-specific position information that Debezium includes with each event. For the Vitess connector, the VGTID recorded in each change event is the offset.

When Kafka Connect gracefully shuts down, it stops the connectors, flushes all event records to Kafka, and records the last offset received from each connector. When Kafka Connect restarts, it reads the last recorded offset for each connector, and starts each connector at its last recorded offset. When the connector restarts, it sends a request to VStream to send the events starting just after that position.

The Vitess connector writes events for all insert, update, and delete operations on a single table to a single Kafka topic. By default, the Kafka topic name is topicPrefix.keyspaceName.tableName where:

For example, suppose that fulfillment is the logical server name in the configuration for a connector that is capturing changes in a Vitess installation that has an commerce keyspace that contains four tables: products, products_on_hand, customers, and orders. Regardless of how many shards the keyspace has, the connector would stream records to these four Kafka topics:

Debezium can generate events that represent transaction boundaries and that enrich data change event messages.

Debezium generates transaction boundary events for the BEGIN and END delimiters in every transaction. Transaction boundary events contain the following fields:

Unless overridden via the topic.transaction option, the connector emits transaction events to the <topic.prefix>.transaction topic.

When transaction metadata is enabled the data message Envelope is enriched with a new transaction field. This field provides information about every event in the form of a composite of fields:

Following is an example of a message:

You can configure Debezium to include additional metadata in data change event records. Such supplemental metadata can assist downstream consumers in processing messages in the correct order when repartitioning, or some other disruption, might otherwise lead to data being consumed out of order.

To configure the connector to emit enriched data change event records, set the transaction.metadata.factory property. When this property is set to VitessOrderedTransactionMetadataFactory, the connector includes a transaction field in the message Envelope. The transaction field adds metadata that provides information about the order in which transactions occur. The following fields are added to every message:

A third field is also relevant to event ordering:

The following example illustrates how to use these fields to establish event order.

Suppose Debezium emits change event records for two events that occur in the same shard and share the same primary key. If the Kafka topic where these events are sent is repartitioned, then the consumer order of the two events cannot be trusted. If Debezium is configured to provide enriched transaction metadata, applications that consume from the topic can apply the following logic to determine which of the two event to apply (the newer event) and which to discard:

If neither of the two events have a higher total_order value, then the events are part of the same transaction. Because the total_order field represents the order of events within a transaction, the event with the greater value is the most recent event.

The following example shows a data change event with ordered transaction metadata:

If you enable the connector to provide transaction metadata, it generates significantly more data. Not only does the connector send additional messages to the transaction topics, but messages that it sends to the data change topics are larger, because they include a transaction metadata block. The added volume is due to the following factors:

To enable the Vitess connector to encode transaction metadata without significantly increasing the volume of data produced, Debezium provides several single message transformations (SMTs). The following SMTs are designed to reduce the amount of data in events that the Vitess connector emits:

The following example shows an excerpt from the configuration of a Vitess connector that uses the preceding transformations:

For a given table, the change event’s key has a structure that contains a field for each column in the primary key of the table at the time the event was created.

Consider a customers table defined in the commerce keyspace and the example of a change event key for that table.

If the topic.prefix connector configuration property has the value Vitess_server, every change event for the customers table while it has this definition has the same key structure, which in JSON looks like this:

The value in a change event is a bit more complicated than the key. Like the key, the value has a schema section and a payload section. The schema section contains the schema that describes the Envelope structure of the payload section, including its nested fields. Change events for operations that create, update or delete data all have a value payload with an envelope structure.

Consider the same sample table that was used to show an example of a change event key:

The emitted events for UPDATE and DELETE operations contain the previous values of all columns in the table.

The following example shows the value portion of a change event that the connector generates for an operation that creates data in the customers table:

The value of a change event for an update in the sample customers table has the same schema as a create event for that table. Likewise, the event value’s payload has the same structure. However, the event value payload contains different values in an update event. Here is an example of a change event value in an event that the connector generates for an update in the customers table:

An UPDATE operation that changes a row’s primary key field(s) is known as a primary key change. For a primary key change, in place of an UPDATE event record, the connector emits a DELETE event record for the old key and a CREATE event record for the new (updated) key. These events have the usual structure and content, and in addition, each one has a message header related to the primary key change:

The value in a delete change event has the same schema portion as create and update events for the same table. The payload portion in a delete event for the sample customers table looks like this:

A delete change event record provides a consumer with the information it needs to process the removal of this row.

Vitess connector events are designed to work with Kafka log compaction. Log compaction enables removal of some older messages as long as at least the most recent message for every key is kept. This lets Kafka reclaim storage space while ensuring that the topic contains a complete data set and can be used for reloading key-based state.

When a row is deleted, the delete event value still works with log compaction, because Kafka can remove all earlier messages that have that same key. However, for Kafka to remove all messages that have that same key, the message value must be null. To make this possible, the Vitess connector follows a delete event with a special tombstone event that has the same key but a null value.

The following table describes how the connector maps basic Vitess data types to a literal type and a semantic type in event fields.

Excluding the TIMESTAMP data type, Vitess temporal types depend on the value of the time.precision.mode connector configuration property.

The DATETIME type represents a local date and time such as "2018-01-13 09:48:27". As you can see in the preceding example, this type does not include time zone information. Columns of this type are converted into epoch milliseconds or microseconds based on the column’s precision by using UTC. The TIMESTAMP type represents a timestamp without time zone information. When writing data, MySQL converts the TIMESTAMP type from the time zone of the server or session into UTC format. When it reads values, the database converts from UTC format to the current time zone of the server or session. For example:

Such columns are converted into an equivalent io.debezium.time.ZonedTimestamp in UTC, based on the time zone of the server or session. By default, Debezium queries the server for the time zone. If this fails, you must explicitly specify the timezone by setting the connectionTimeZone option in the JDBC connection string. For example, if the database’s time zone (either globally, or as configured for the connector by means of the connectionTimeZone option) is "America/Los_Angeles", the TIMESTAMP value "2018-06-20 06:37:03" is represented by a ZonedTimestamp with the value "2018-06-20T13:37:03Z".

The time zone of the JVM that runs Kafka Connect and Debezium does not affect these conversions.

For more information about properties that affect temporal values, see the connector configuration properties.

Because Vitess connector reads change events from the VTGate VStream gRPC server, it does not need to connect directly to MySQL instances. Therefore, no special database user and permissions are needed. At the moment, Vitess connector only supports unauthenticated access to the VTGate gRPC server.

Following is an example of the configuration for a Vitess connector that connects to a Vitess (VTGate’s VStream) server on port 15991 at 192.168.99.100, whose logical name is fullfillment. It also connects to a VTCtld server on port 15999 at 192.168.99.101 to get the initial VGTID. Typically, you configure the Debezium Vitess connector in a .json file using the configuration properties available for the connector.

You can choose to produce events for a subset of the schemas and tables. Optionally, ignore, mask, or truncate columns that are sensitive, too large, or not needed.

See the complete list of Vitess connector properties that can be specified in these configurations.

You can send this configuration with a POST command to a running Kafka Connect service. The service records the configuration and starts the connector task that connects to the Vitess database and streams change event records to Kafka topics.

When you have a big Vitess installation which requires more than one connector task to process the change logs, you can use offset-storage-per-task feature to launch multiple connector tasks and have each task work with a subset of vitess shards. Each task will persist its offsets (the vgtids for the shards it’s tracking) in Kafka’s offset topic in its own partition space.

Following is the same example for a Vitess connector that connects to a Vitess (VTGate’s VStream) server but with three additional parameteres to invoke the offset-storage-per-task mode.

The task to vitess shards distribution is based on a simple round robin algorithm. In this example of launching two connector tasks and assume we have 4 vitess shards (-40,40-80,80-c0,c0-), task0 will be working on shards (-40,80-c0) and task1 will be working on shards (40-80,c0-).

The reason that we need three config params is to make sure the offsets saved by each connector task don’t collide with each other and to handle the migration of offsets by the previous task paralleism automatically. In order to make sure that we don’t collide on the partition keys in Kafka offset topic, we are using this partition name scheme for each connector task: taskId_numTasks_gen. So for the current example of launching two tasks with generation number 1, task0 will be writing its offsets in Kafka’s offset topic in partition key: task0_2_1 and task1 will be using partition key: (task1_2_1). The gen config param is used to distinguish the partition keys generated from different generations (generation corresponds to each change of task parallelism)

When the task paralleism changes (e.g. you want to launch 4 connector tasks instead of 2 to handle the bigger volume of traffic from vitess), you will specify tasks.max=4, vitess.offset.storage.task.key.gen=2, vitess.prev.num.tasks=2, the offset partition for this task paralleism generation will be: task0_4_2, task1_4_2, task2_4_2, task3_4_2. Once the connector restarts, the connector will detect there is no previous offsets saved for the current 4 partition keys and it will invoke an automatic offset migration from the offsets saved in the previous generation keys: task0_2_1 and task1_2_1. For the current example of 4 vitess shards (-40,40-80,80-c0,c0-), task0 will be working on shard:(-40), task1:(40-80), task2:(80-c0), task3:(c0-). The offsets for those 4 shards from the previous generation of parallelism (using 2 tasks with each task working with 2 shards) will be auto-migrated to this generation of using 4 tasks (one task working with one shard each).

Note that the task parallelism gen number is defaulted to be 0 for the offsets saved in Kafka offset topic before offset-storage-per-task feature is enabled, there is a special offset lookup during offset migration. So if you have the vitess connector running for a while without the offset-storage-per-task feature on and now you want to turn on this feature, please specify vitess.offset.storage.task.key.gen=1, vitess.prev.num.tasks=1 to help the offset auto migration.

Note that vitess.prev.num.tasks needs to match the actual number of tasks launched in the previous task parallelism generation. The number of connector tasks is usually the same as the tasks.max config params you specified, but in the rare case that tasks.max > number of vitess shards, the connector will only launch the_number_of_tasks = the_number_of_vitess_shards. This rare case is probably a mis-configuration in the first place.

See the complete list of Vitess connector properties that can be specified in these configurations.

You can send this configuration with a POST command to a running Kafka Connect service. The service records the configuration and starts the connector task that connects to the Vitess database and streams change event records to Kafka topics.

To start running a Vitess connector, create a connector configuration and add the configuration to your Kafka Connect cluster.

When the connector starts, it starts generating data change events for row-level operations and streaming change event records to Kafka topics.

The Debezium Vitess connector provides only one type of metrics that are in addition to the built-in support for JMX metrics that Zookeeper, Kafka, and Kafka Connect provide.

Debezium monitoring documentation provides details for how to expose these metrics by using JMX.

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.

The Debezium Vitess connector has many 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:

The following configuration properties are required unless a default value is available.

The following advanced configuration properties have defaults that work in most situations and therefore rarely need to be specified in the connector’s configuration.

The connector also supports pass-through configuration properties that are used when creating the Kafka producer and consumer.

Be sure to consult the Kafka documentation for all of the configuration properties for Kafka producers and consumers. The Vitess connector does use the new consumer configuration properties.

In the following situations, the connector fails when trying to start, reports an error/exception in the log, and stops running:

In these cases, the error message has details about the problem and possibly a suggested workaround. After you correct the configuration or address the Vitess problem, restart the connector.

When the connector is running, the Vitses server (VTGate) that it is connected to could become unavailable for any number of reasons. If this happens, the connector fails with an error and stops. When the server is available again, restart the connector.

The Vitess connector externally stores the last processed offset in the form of a Vitess VGTID. After a connector restarts and connects to a server instance, the connector communicates with the server to continue streaming from that particular offset.

This error happens very rarely. If you receive an error with the message Illegal prefix '@' for column: x, from schema: y, table: z, and your table doesn’t have such a column, it is a Vitess vstream bug that is caused by column renaming or column type change. It is a transient error. You can restart the connector after a small backoff and it should resolve automatically.

Suppose that Kafka Connect is being run in distributed mode and a Kafka Connect process is stopped gracefully. Prior to shutting down that process, Kafka Connect migrates the process’s connector tasks to another Kafka Connect process in that group. The new connector tasks start processing exactly where the prior tasks stopped. There is a short delay in processing while the connector tasks are stopped gracefully and restarted on the new processes.

If the Kafka Connector process stops unexpectedly, any connector tasks it was running terminate without recording their most recently processed offsets. When Kafka Connect is being run in distributed mode, Kafka Connect restarts those connector tasks on other processes. However, Vitess connectors resume from the last offset that was recorded by the earlier processes. This means that the new replacement tasks might generate some of the same change events that were processed just prior to the crash. The number of duplicate events depends on the offset flush period and the volume of data changes just before the crash.

Because there is a chance that some events might be duplicated during a recovery from failure, consumers should always anticipate some duplicate events. Debezium changes are idempotent, so a sequence of events always results in the same state.

In each change event record, Debezium connectors insert source-specific information about the origin of the event, including the Vitess server’s time of the event, the position in the binlog where the transaction changes were written. Consumers can keep track of this information, especially the VGTID, to determine whether an event is a duplicate.

As the connector generates change events, the Kafka Connect framework records those events in Kafka by using the Kafka producer API. Periodically, at a frequency that you specify in the Kafka Connect configuration, Kafka Connect records the latest offset that appears in those change events. If the Kafka brokers become unavailable, the Kafka Connect process that is running the connectors repeatedly tries to reconnect to the Kafka brokers. In other words, the connector tasks pause until a connection can be re-established, at which point the connectors resume exactly where they left off.

If the connector is gracefully stopped, the database can continue to be used. Any changes are recorded in the Vitess binlog. When the connector restarts, it resumes streaming changes where it left off. That is, it generates change event records for all database changes that were made while the connector was stopped.

A properly configured Kafka cluster is able to handle massive throughput. Kafka Connect is written according to Kafka best practices, and given enough resources a Kafka Connect connector can also handle very large numbers of database change events. Because of this, after being stopped for a while, when a Debezium connector restarts, it is very likely to catch up with the database changes that were made while it was stopped. How quickly this happens depends on the capabilities and performance of Kafka and the volume of changes being made to the data in Vitess.

If a snapshot fails to complete, the connector does not automatically reattempt the snapshot. If the connector is restarted with the previous offset, the connector skips the snapshot process and immediately begins to stream change events.

Consequently, to recover from the failure, remove the connector offsets manually and start the connector.