Debezium connector for Vitess

维塔斯连接器花所有的时间流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 aTablet 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 calledVTCtldand 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 Debeziumcreate,update, ordeleteevents 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 recentoffsetin 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 isserverName.keyspaceName.tableNamewhere:

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

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 acustomerstable defined in thecommercekeyspace and the example of a change event key for that table.

If thedatabase.server.nameconnector configuration property has the valueVitess_server, every change event for thecustomerstable 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 aschemasection and apayloadsection. Theschema部分包含模式描述Envelopestructure of thepayloadsection, 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 forUPDATEandDELETEoeprations 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 thecustomerstable:

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

The value in adeletechange event has the sameschemaportion ascreateandupdateevents for the same table. Thepayloadportion in adeleteevent for the samplecustomerstable looks like this:

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

Vitess connector events are designed to work withKafka 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, thedeleteevent 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 benull. To make this possible, the Vitess connector follows adeleteevent with a specialtombstoneevent that has the same key but anullvalue.

The following table describes how the connector maps basic Vitess data types to aliteral typeand asemantic typein event fields.

You can follow theLocal Install via Dockerguide, or theVitess Operator for Kubernetesguide to install Vitess. No special setup is needed to support Vitess connector.

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 isfullfillment. 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.jsonfile 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 thecomplete list of Vitess connector propertiesthat can be specified in these configurations.

You can send this configuration with aPOSTcommand 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.

当连接器启动时,它开始产生哒ta 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 documentationprovides details for how to expose these metrics by using JMX.

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 arerequiredunless a default value is available.

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

The connector also supportspass-throughconfiguration properties that are used when creating the Kafka producer and consumer.

Be sure to consult theKafka documentationfor all of the configuration properties for Kafka producers and consumers. The Vitess connector does use thenew 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 messageIllegal prefix '@' for column: x, from schema: y, table: z, and your table doesn’t have such a column, it is a Vitess vstreambugthat 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 wasrecordedby 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.