// Metadata created by nebel
//

[id="add-mysql-connector-configuration-to-kafka-connect"]
= Adding MySQL connector configuration to Kafka Connect

ifdef::community[]
To start running a MySQL connector, configure a connector and add the configuration to your Kafka Connect cluster. 

.Prerequisites

* {link-prefix}:{link-mysql-connector}#setup-the-mysql-server[MySQL server] is 
set up for a {prodname} connector.

* {prodname} MySQL connector is installed. 

.Procedure

. Create a configuration for the MySQL connector.

. Use the link:{link-kafka-docs}/#connect_rest[Kafka Connect REST API] to add that connector configuration to your Kafka Connect cluster. 

endif::community[]

ifdef::product[]
You can use a provided {prodname} container to deploy a {prodname} MySQL connector. In this procedure, you build a custom Kafka Connect container image for {prodname}, configure the {prodname} connector as needed, and then add your connector configuration to your Kafka Connect environment. 

.Prerequisites

* Podman or Docker is installed and you have sufficient rights to create and manage containers.
* You installed the {prodname} MySQL connector archive. 

.Procedure

. Extract the {prodname} MySQL connector archive to create a directory structure for the connector plug-in, for example: 
+
[subs=+macros]
----
pass:quotes[*tree ./my-plugins/*]
./my-plugins/
├── debezium-connector-mysql
│   ├── ...
----

. Create and publish a custom image for running your {prodname} connector:

.. Create a new `Dockerfile` by using `{DockerKafkaConnect}` as the base image. In the following example, you would replace _my-plugins_ with the name of your plug-ins directory:
+
[subs=+macros,subs="+attributes"]
----
FROM {DockerKafkaConnect}
USER root:root
pass:quotes[COPY _./my-plugins/_ /opt/kafka/plugins/]
USER 1001
----
+
Before Kafka Connect starts running the connector, Kafka Connect loads any third-party plug-ins that are in the `/opt/kafka/plugins` directory.

.. Build the container image. For example, if you saved the `Dockerfile` that you created in the previous step as `debezium-container-for-mysql`, then you would run the following command:
+
`podman build -t debezium-container-for-mysql:latest`

.. Push your custom image to your container registry, for example:
+
`podman push debezium-container-for-mysql:latest`

.. Point to the new container image. Do one of the following:
+
* Edit the `spec.image` property of the `KafkaConnector` custom resource. If set, this property overrides the `STRIMZI_DEFAULT_KAFKA_CONNECT_IMAGE` variable in the Cluster Operator. For example:
+
[source,yaml,subs=attributes+]
----
apiVersion: {KafkaConnectApiVersion}
kind: KafkaConnector
metadata:
  name: my-connect-cluster
spec:
  #...
  image: debezium-container-for-mysql
----
+
* In the `install/cluster-operator/050-Deployment-strimzi-cluster-operator.yaml` file, edit the `STRIMZI_DEFAULT_KAFKA_CONNECT_IMAGE` variable to point to the new container image and reinstall the Cluster Operator. If you edit this file you must apply it to your OpenShift cluster.

. Create a `KafkaConnector` custom resource that defines your {prodname} MySQL connector instance. See {LinkDebeziumUserGuide}#configure-the-mysql-connector_{context}[the connector configuration example].

. Apply the connector instance, for example: 
+
`oc apply -f inventory-connector.yaml`
+
This registers `inventory-connector` and the connector starts to run against the `inventory` database.

. Verify that the connector was created and has started to capture changes in the specified database. You can verify the connector instance by watching the Kafka Connect log output as, for example, `inventory-connector` starts.

.. Display the Kafka Connect log output:
+
[source,shell,options="nowrap"]
----
oc logs $(oc get pods -o name -l strimzi.io/name=my-connect-cluster-connect)
----

.. Review the log output to verify that the initial snapshot has been executed. You should see something like the following lines: 
+
[source,shell,options="nowrap"]
----
... INFO Starting snapshot for ...
... INFO Snapshot is using user 'debezium' ... 
----

endif::product[]

.Results

When the connector starts, it {link-prefix}:{link-mysql-connector}#how-the-mysql-connector-performs-database-snapshots_{context}[performs a consistent snapshot] of the MySQL databases that the connector is configured for. The connector then starts generating data change events for row-level operations and streaming change event records to Kafka topics. 
