// This module is included in:
//
// overview/assembly-configuration-points.adoc

[id="configuration-points-connect_{context}"]
= Kafka Connect configuration

[role="_abstract"]
Use Strimzi’s `KafkaConnect` resource to quickly and easily create new Kafka Connect clusters.

When you deploy Kafka Connect using the `KafkaConnect` resource,
you specify bootstrap server addresses (in `spec.bootstrapServers`) for connecting to a Kafka cluster.
You can specify more than one address in case a server goes down.
You also specify the authentication credentials and TLS encryption certificates to make a secure connection.

NOTE: The Kafka cluster doesn't need to be managed by Strimzi or deployed to a Kubernetes cluster.

You can also use the `KafkaConnect` resource to specify the following:

* Plugin configuration to build a container image that includes the plugins to make connections
* Configuration for the worker pods that belong to the Kafka Connect cluster
* An annotation to enable use of the `KafkaConnector` resource to manage plugins

The Cluster Operator manages Kafka Connect clusters deployed using the `KafkaConnect` resource and connectors created using the `KafkaConnector` resource.

[discrete]
== Plugin configuration

Plugins provide the implementation for creating connector instances.
When a plugin is instantiated, configuration is provided for connection to a specific type of external data system.
Plugins provide a set of one or more JAR files that define a connector and task implementation for connecting to a given kind of data source.
Plugins for many external systems are available for use with Kafka Connect.
You can also create your own plugins.

The configuration describes the source input data and target output data to feed into and out of Kafka Connect.
For a source connector, external source data must reference specific topics that will store the messages.
The plugins might also contain the libraries and files needed to transform the data.

A Kafka Connect deployment can have one or more plugins, but only one version of each plugin.

You can create a custom Kafka Connect image that includes your choice of plugins.
You can create the image in two ways:

* link:{BookURLDeploying}#creating-new-image-using-kafka-connect-build-str[Automatically using Kafka Connect configuration^]
* link:{BookURLDeploying}#creating-new-image-from-base-str[Manually using a Dockerfile and a Kafka container image as a base image^]

To create the container image automatically, you specify the plugins to add to your Kafka Connect cluster using the `build` property of the `KafkaConnect` resource.
Strimzi automatically downloads and adds the plugin artifacts to a new container image.

.Example plugin configuration
[source,yaml,subs="+quotes,attributes"]
----
apiVersion: {KafkaConnectApiVersion}
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  # ...
  build: <1>
    output: <2>
      type: docker
      image: my-registry.io/my-org/my-connect-cluster:latest
      pushSecret: my-registry-credentials
    plugins: <3>
      - name: debezium-postgres-connector
        artifacts:
          - type: tgz
            url: https://_ARTIFACT-ADDRESS_.tgz
            sha512sum: _HASH-NUMBER-TO-VERIFY-ARTIFACT_
      # ...
  # ...
----
<1> link:{BookURLUsing}#type-Build-reference[Build configuration properties^] for building a container image with plugins automatically.
<2> Configuration of the container registry where new images are pushed. The `output` properties describe the type and name of the image, and optionally the name of the secret containing the credentials needed to access the container registry.
<3> List of plugins and their artifacts to add to the new container image. The `plugins` properties describe the type of artifact and the URL from which the artifact is downloaded. Each plugin must be configured with at least one artifact. Additionally, you can specify a SHA-512 checksum to verify the artifact before unpacking it.

If you are using a Dockerfile to build an image, you can use Strimzi’s latest container image as a base image to add your plugin configuration file.

.Example showing manual addition of plugin configuration
[source,subs="+quotes,attributes"]
----
FROM {DockerKafkaConnect}
USER root:root
COPY ./_my-plugins_/ /opt/kafka/plugins/
USER {DockerImageUser}
----

[discrete]
== Kafka Connect cluster configuration for workers

You specify the configuration for workers in the `config` property of the `KafkaConnect` resource.

A distributed Kafka Connect cluster has a group ID and a set of internal configuration topics.

* `group.id`
* `offset.storage.topic`
* `config.storage.topic`
* `status.storage.topic`

Kafka Connect clusters are configured by default with the same values for these properties.
Kafka Connect clusters cannot share the group ID or topic names as it will create errors.
If multiple different Kafka Connect clusters are used, these settings must be unique for the workers of each Kafka Connect cluster created.

The names of the connectors used by each Kafka Connect cluster must also be unique.

In the following example worker configuration, JSON converters are specified.
A replication factor is set for the internal Kafka topics used by Kafka Connect.
This should be at least 3 for a production environment.
Changing the replication factor after the topics have been created will have no effect.

.Example worker configuration
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaConnectApiVersion}
kind: KafkaConnect
# ...
spec:
  config:
    # ...
    group.id: my-connect-cluster <1>
    offset.storage.topic: my-connect-cluster-offsets <2>
    config.storage.topic: my-connect-cluster-configs <3>
    status.storage.topic: my-connect-cluster-status <4>
    key.converter: org.apache.kafka.connect.json.JsonConverter <5>
    value.converter: org.apache.kafka.connect.json.JsonConverter <6>
    key.converter.schemas.enable: true <7>
    value.converter.schemas.enable: true <8>
    config.storage.replication.factor: 3 <9>
    offset.storage.replication.factor: 3 <10>
    status.storage.replication.factor: 3 <11>
  # ...
----
<1> The Kafka Connect cluster ID within Kafka. Must be unique for each Kafka Connect cluster.
<2> Kafka topic that stores connector offsets. Must be unique for each Kafka Connect cluster.
<3> Kafka topic that stores connector and task status configurations. Must be unique for each Kafka Connect cluster.
<4> Kafka topic that stores connector and task status updates. Must be unique for each Kafka Connect cluster.
<5> Converter to transform message keys into JSON format for storage in Kafka.
<6> Converter to transform message values into JSON format for storage in Kafka.
<7> Schema enabled for converting message keys into structured JSON format.
<8> Schema enabled for converting message values into structured JSON format.
<9> Replication factor for the Kafka topic that stores connector offsets.
<10> Replication factor for the Kafka topic that stores connector and task status configurations.
<11> Replication factor for the Kafka topic that stores connector and task status updates.

[discrete]
== `KafkaConnector` management of connectors

After plugins have been added to the container image used for the worker pods in a deployment,
you can use Strimzi’s `KafkaConnector` custom resource or the Kafka Connect API to manage connector instances.
You can also create new connector instances using these options.

The `KafkaConnector` resource offers a Kubernetes-native approach to management of connectors by the Cluster Operator.
To manage connectors with `KafkaConnector` resources, you must specify an annotation in your `KafkaConnect` custom resource.

.Annotation to enable KafkaConnectors
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaConnectApiVersion}
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
  # ...
----

Setting `use-connector-resources` to `true`  enables KafkaConnectors to create, delete, and reconfigure connectors.

If `use-connector-resources` is enabled in your `KafkaConnect` configuration, you must use the `KafkaConnector` resource to define and manage connectors.
`KafkaConnector` resources are configured to connect to external systems.
They are deployed to the same Kubernetes cluster as the Kafka Connect cluster and Kafka cluster interacting with the external data system.

.Kafka components are contained in the same Kubernetes cluster
image:overview/kafka-concepts-kafka-connector.png[Kafka and Kafka Connect clusters]

The configuration specifies how connector instances connect to an external data system, including any authentication.
You also need to state what data to watch.
For a source connector, you might provide a database name in the configuration.
You can also specify where the data should sit in Kafka by specifying a target topic name.

Use `tasksMax` to specify the maximum number of tasks.
For example, a source connector with `tasksMax: 2` might split the import of source data into two tasks.

.Example KafkaConnector source connector configuration
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaConnectApiVersion}
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 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> link:{BookURLDeploying}#kafkaconnector-configs[Connector configuration^] as key-value pairs.
<6> Location of the external data file. In this example, we're configuring the `FileStreamSourceConnector` to read from the `/opt/kafka/LICENSE` file.
<7> Kafka topic to publish the source data to.

NOTE: You can link:{BookURLUsing}#proc-loading-config-with-provider-str[load confidential configuration values for a connector^] from Kubernetes Secrets or ConfigMaps.

[discrete]
== Kafka Connect API

Use the Kafka Connect REST API as an alternative to using `KafkaConnector` resources to manage connectors.
The Kafka Connect REST API is available as a service running on `_<connect_cluster_name>_-connect-api:8083`, where _<connect_cluster_name>_ is the name of your Kafka Connect cluster.

You add the connector configuration as a JSON object.

.Example curl request to add connector configuration
[source,curl,subs=attributes+]
----
curl -X POST \
  http://my-connect-cluster-connect-api:8083/connectors \
  -H 'Content-Type: application/json' \
  -d '{ "name": "my-source-connector",
    "config":
    {
      "connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector",
      "file": "/opt/kafka/LICENSE",
      "topic":"my-topic",
      "tasksMax": "4",
      "type": "source"
    }
}'
----

If KafkaConnectors are enabled, manual changes made directly using the Kafka Connect REST API are reverted by the Cluster Operator.

The operations supported by the REST API are described in the http://kafka.apache.org[Apache Kafka documentation^].

NOTE: You can expose the Kafka Connect API service outside Kubernetes.
You do this by creating a service that uses a connection mechanism that provides the access, such as an ingress or route.
Use advisedly as the connection is insecure.

[role="_additional-resources"]
.Additional resources

* link:{BookURLUsing}#assembly-kafka-connect-str[Kafka Connect configuration options^]
* link:{BookURLUsing}#con-kafka-connect-multiple-instances-str[Kafka Connect configuration for multiple instances^]
* link:{BookURLDeploying}#using-kafka-connect-with-plug-ins-str[Extending Kafka Connect with plugins^]
* link:{BookURLDeploying}#creating-new-image-using-kafka-connect-build-str[Creating a new container image automatically using Strimzi^]
* link:{BookURLDeploying}#creating-new-image-from-base-str[Creating a Docker image from the Kafka Connect base image^]
* link:{BookURLUsing}#type-Build-reference[Build schema reference^]
* link:{BookURLDeploying}#kafkaconnector-configs[Source and sink connector configuration options^]
* link:{BookURLUsing}#proc-loading-config-with-provider-str[Loading configuration values from external sources^]
