// Module included in the following assemblies:
//
// assembly-kafka-connect.adoc

[id='proc-deploying-kafkaconnector-{context}']
= Deploying example KafkaConnector resources

[role="_abstract"]
Use KafkaConnectors with Kafka Connect to stream data to and from a Kafka cluster.

Strimzi provides xref:deploy-examples-{context}[example configuration files].
In this procedure, we use the following example file:

* `examples/connect/source-connector.yaml`.

The file is used to create the following connector instances:

* A `FileStreamSourceConnector` instance that reads each line from the Kafka license file (the source) and writes the data as messages to a single Kafka topic.
* A `FileStreamSinkConnector` instance that reads messages from the Kafka topic and writes the messages to a temporary file (the sink).

Up until Apache Kafka 3.1.0, the example connector plugins were included with Apache Kafka.
Starting from the 3.1.1 and 3.2.0 releases of Apache Kafka, the examples need to be added to the plugin path as any other connector.
See xref:using-kafka-connect-with-plug-ins-str[Extending Kafka Connect with connector plugins] for more details.
You can also use the `examples/connect/kafka-connect-build.yaml` example file.

[NOTE]
====
In a production environment, you prepare container images containing your desired Kafka Connect connectors, as described in xref:using-kafka-connect-with-plug-ins-{context}[].

The `FileStreamSourceConnector` and `FileStreamSinkConnector` are provided as examples. Running these connectors in containers as described here is unlikely to be suitable for production use cases.
====

.Prerequisites

* A Kafka Connect deployment
* link:{BookURLUsing}#proc-kafka-connect-config-str[KafkaConnectors are enabled in the Kafka Connect deployment^]
* The Cluster Operator is running

.Procedure

. Edit the `examples/connect/source-connector.yaml` file:
+
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaConnectorApiVersion}
kind: KafkaConnector
metadata:
  name: my-source-connector <1>
  labels:
    strimzi.io/cluster: my-connect-cluster <2>
spec:
  class: org.apache.kafka.connect.file.FileStreamSourceConnector <3>
  tasksMax: 2 <4>
  config: <5>
    file: "/opt/kafka/LICENSE" <6>
    topic: my-topic <7>
    # ...
----
+
<1> Name of the `KafkaConnector` resource, which is used as the name of the connector. Use any name that is valid for a Kubernetes resource.
<2> Name of the Kafka Connect cluster to create the connector instance in. Connectors must be deployed to the same namespace as the Kafka Connect cluster they link to.
<3> Full name or alias of the connector class. This should be present in the image being used by the Kafka Connect cluster.
<4> Maximum number of Kafka Connect `Tasks` that the connector can create.
<5> xref:kafkaconnector-configs[Connector configuration] as key-value pairs.
<6> This example source connector configuration reads data from the `/opt/kafka/LICENSE` file.
<7> Kafka topic to publish the source data to.

. Create the source `KafkaConnector` in your Kubernetes cluster:
+
[source,shell,subs="+quotes"]
----
kubectl apply -f examples/connect/source-connector.yaml
----

. Create an `examples/connect/sink-connector.yaml` file:
+
[source,shell,subs="+quotes"]
----
touch examples/connect/sink-connector.yaml
----

. Paste the following YAML into the `sink-connector.yaml` file:
+
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaConnectorApiVersion}
kind: KafkaConnector
metadata:
  name: my-sink-connector
  labels:
    strimzi.io/cluster: my-connect
spec:
  class: org.apache.kafka.connect.file.FileStreamSinkConnector <1>
  tasksMax: 2
  config: <2>
    file: "/tmp/my-file" <3>
    topics: my-topic <4>
----
+
<1> Full name or alias of the connector class. This should be present in the image being used by the Kafka Connect cluster.
<2> xref:#kafkaconnector-configs[Connector configuration] as key-value pairs.
<3> Temporary file to publish the source data to.
<4> Kafka topic to read the source data from.

. Create the sink `KafkaConnector` in your Kubernetes cluster:
+
[source,shell,subs="+quotes"]
----
kubectl apply -f examples/connect/sink-connector.yaml
----

. Check that the connector resources were created:
+
[source,shell,subs="+quotes"]
----
kubectl get kctr --selector strimzi.io/cluster=_MY-CONNECT-CLUSTER_ -o name

my-source-connector
my-sink-connector
----
+
Replace _MY-CONNECT-CLUSTER_ with your Kafka Connect cluster.

. In the container, execute `kafka-console-consumer.sh` to read the messages that were written to the topic by the source connector:
+
[source,shell,subs="+quotes"]
----
kubectl exec _MY-CLUSTER_-kafka-0 -i -t -- bin/kafka-console-consumer.sh --bootstrap-server _MY-CLUSTER_-kafka-bootstrap._NAMESPACE_.svc:9092 --topic my-topic --from-beginning
----

[[kafkaconnector-configs]]
[discrete]
== Source and sink connector configuration options

The connector configuration is defined in the `spec.config` property of the `KafkaConnector` resource.

The `FileStreamSourceConnector` and `FileStreamSinkConnector` classes support the same configuration options as the Kafka Connect REST API.
Other connectors support different configuration options.

.Configuration options for the `FileStreamSource` connector class
[cols="4*",options="header",stripes="none",separator=¦]
|===

¦Name
¦Type
¦Default value
¦Description

m¦file
¦String
¦Null
¦Source file to write messages to. If not specified, the standard input is used.

m¦topic
¦List
¦Null
¦The Kafka topic to publish data to.

|===

.Configuration options for `FileStreamSinkConnector` class
[cols="4*",options="header",stripes="none",separator=¦]
|===

¦Name
¦Type
¦Default value
¦Description

m¦file
¦String
¦Null
¦Destination file to write messages to. If not specified, the standard output is used.

m¦topics
¦List
¦Null
¦One or more Kafka topics to read data from.

m¦topics.regex
¦String
¦Null
¦A regular expression matching one or more Kafka topics to read data from.

|===
