anchor.diff:  
   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.
   
  +// Type: concept
  +// ModuleID:debezium-vitess-connector-configuration-example-offset-storage-per-task
  +// Title: {prodname} Vitess connector configuration example-offset-storage-per-task
  +[[vitess-example-configuration-offset-storage-per-task]]
  +=== Connector configuration example for offset-storage-per-task mode
  +
  +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.
  +
  +[source,json]
  +----
  +{
  +  "name": "inventory-connector",
  +  "config": {
  +    "connector.class": "io.debezium.connector.vitess.VitessConnector",
  +    "database.hostname": "192.168.99.100",
  +    "database.port": "15991",
  +    "database.user": "vitess",
  +    "database.password": "vitess_password",
  +    "vitess.keyspace": "commerce",
  +    "vitess.tablet.type": "MASTER",
  +    "vitess.vtctld.host": "192.168.99.101",
  +    "vitess.vtctld.port": "15999",
  +    "vitess.vtctld.user": "vitess",
  +    "vitess.vtctld.password": "vitess_password",
  +    "database.server.name": "fullfillment",
  +    "vitess.offset.storage.per.task": true, // <1>
  +    "vitess.offset.storage.task.key.gen": 1, // <2>
  +    "vitess.prev.num.tasks": 1, // <3>
  +    "tasks.max": 2 // <4>
  +  }
  +}
  +----
  +<1> Specify that we want to turn on offset-storage-per-task feature
  +<2> Specify that the generation number for the current task parallelism is 1
  +<3> Specify that the number of tasks in the previous generation of task parallelism is 1
  +<4> Specify that we want to launch two tasks for the current task parallelism
  +
  +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 xref:vitess-connector-properties[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.
  +
   // Type: procedure
   // ModuleID: adding-debezium-vitess-connector-configuration-to-kafka-connect
   // Title: Adding {prodname} Vitess connector configuration to Kafka Connect
  
positive.code: `false` - only a _delete_ event is emitted. +
   +
  After a source record is deleted, emitting a tombstone event (the default behavior) allows Kafka to completely delete all events that pertain to the key of the deleted row in case {link-kafka-docs}/#compaction[log compaction] is enabled for the topic.
  
  |[[vitess-property-message-key-columns]]<<vitess-property-message-key-columns, `+message.key.columns+`>>
  |_empty string_
  
negative.code: 
  |[[vitess-property-tasks-max]]<<vitess-property-tasks-max, `+tasks.max+`>>
  |`1`
  |The maximum number of tasks that should be created for this connector. The Vitess connector can use more than 1 tasks if you enable offset.storage.per.task mode.
  
  |[[vitess-property-database-hostname]]<<vitess-property-database-hostname, `+database.hostname+`>>
  |
  