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

[id='proc-kafka-connect-config-{context}']
= Configuring Kafka Connect

Use Kafka Connect to set up external data connections to your Kafka cluster.
Use the properties of the `KafkaConnect` resource to configure your Kafka Connect deployment.

.KafkaConnector configuration
`KafkaConnector` resources allow you to create and manage connector instances for Kafka Connect in a Kubernetes-native way.

In your Kafka Connect configuration, you enable KafkaConnectors for a Kafka Connect cluster by adding the `strimzi.io/use-connector-resources` annotation.
You can also add a `build` configuration so that Strimzi automatically builds a container image with the connector plugins you require for your data connections.
External configuration for Kafka Connect connectors is specified through the `externalConfiguration` property.

To manage connectors, you can use use `KafkaConnector` custom resources or the Kafka Connect REST API.
`KafkaConnector` resources must be deployed to the same namespace as the Kafka Connect cluster they link to.
For more information on using these methods to create, reconfigure, or delete connectors, see link:{BookURLDeploying}#using-kafka-connect-with-plug-ins-{context}[Adding connectors^].

Connector configuration is passed to Kafka Connect as part of an HTTP request and stored within Kafka itself.
ConfigMaps and Secrets are standard Kubernetes resources used for storing configurations and confidential data.
You can use ConfigMaps and Secrets to configure certain elements of a connector.
You can then reference the configuration values in HTTP REST commands, which keeps the configuration separate and more secure, if needed.
This method applies especially to confidential data, such as usernames, passwords, or certificates.

.Handling high volumes of messages
You can tune the configuration to handle high volumes of messages.
For more information, see link:{BookURLDeploying}#con-high-volume-config-properties-{context}[Handling high volumes of messages^].

.Prerequisites

* A Kubernetes cluster
* A running Cluster Operator

See the _Deploying and Upgrading Strimzi_ guide for instructions on running a:

* link:{BookURLDeploying}#cluster-operator-str[Cluster Operator^]
* link:{BookURLDeploying}#deploying-kafka-cluster-str[Kafka cluster^]

.Procedure

. Edit the `spec` properties of the `KafkaConnect` resource.
+
The properties you can configure are shown in this example configuration:
+
[source,yaml,subs=attributes+,options="nowrap"]
----
apiVersion: {KafkaConnectApiVersion}
kind: KafkaConnect <1>
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true" <2>
spec:
  replicas: 3 <3>
  authentication: <4>
    type: tls
    certificateAndKey:
      certificate: source.crt
      key: source.key
      secretName: my-user-source
  bootstrapServers: my-cluster-kafka-bootstrap:9092 <5>
  tls: <6>
    trustedCertificates:
      - secretName: my-cluster-cluster-cert
        certificate: ca.crt
      - secretName: my-cluster-cluster-cert
        certificate: ca2.crt
  config: <7>
    group.id: my-connect-cluster
    offset.storage.topic: my-connect-cluster-offsets
    config.storage.topic: my-connect-cluster-configs
    status.storage.topic: my-connect-cluster-status
    key.converter: org.apache.kafka.connect.json.JsonConverter
    value.converter: org.apache.kafka.connect.json.JsonConverter
    key.converter.schemas.enable: true
    value.converter.schemas.enable: true
    config.storage.replication.factor: 3
    offset.storage.replication.factor: 3
    status.storage.replication.factor: 3
  build: <8>
    output: <9>
      type: docker
      image: my-registry.io/my-org/my-connect-cluster:latest
      pushSecret: my-registry-credentials
    plugins: <10>
      - name: debezium-postgres-connector
        artifacts:
          - type: tgz
            url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/2.1.1.Final/debezium-connector-postgres-2.1.1.Final-plugin.tar.gz
            sha512sum: 962a12151bdf9a5a30627eebac739955a4fd95a08d373b86bdcea2b4d0c27dd6e1edd5cb548045e115e33a9e69b1b2a352bee24df035a0447cb820077af00c03
      - name: camel-telegram
        artifacts:
          - type: tgz
            url: https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-telegram-kafka-connector/0.9.0/camel-telegram-kafka-connector-0.9.0-package.tar.gz
            sha512sum: a9b1ac63e3284bea7836d7d24d84208c49cdf5600070e6bd1535de654f6920b74ad950d51733e8020bf4187870699819f54ef5859c7846ee4081507f48873479
  externalConfiguration: <11>
    env:
      - name: AWS_ACCESS_KEY_ID
        valueFrom:
          secretKeyRef:
            name: aws-creds
            key: awsAccessKey
      - name: AWS_SECRET_ACCESS_KEY
        valueFrom:
          secretKeyRef:
            name: aws-creds
            key: awsSecretAccessKey
  resources: <12>
    requests:
      cpu: "1"
      memory: 2Gi
    limits:
      cpu: "2"
      memory: 2Gi
  logging: <13>
    type: inline
    loggers:
      log4j.rootLogger: "INFO"
  readinessProbe: <14>
    initialDelaySeconds: 15
    timeoutSeconds: 5
  livenessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  metricsConfig: <15>
    type: jmxPrometheusExporter
    valueFrom:
      configMapKeyRef:
        name: my-config-map
        key: my-key
  jvmOptions: <16>
    "-Xmx": "1g"
    "-Xms": "1g"
  image: my-org/my-image:latest <17>
  rack:
    topologyKey: topology.kubernetes.io/zone <18>
  template: <19>
    pod:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            - labelSelector:
                matchExpressions:
                  - key: application
                    operator: In
                    values:
                      - postgresql
                      - mongodb
              topologyKey: "kubernetes.io/hostname"
    connectContainer: <20>
      env:
        - name: JAEGER_SERVICE_NAME
          value: my-jaeger-service
        - name: JAEGER_AGENT_HOST
          value: jaeger-agent-name
        - name: JAEGER_AGENT_PORT
          value: "6831"
----
<1> Use `KafkaConnect`.
<2> Enables KafkaConnectors for the Kafka Connect cluster.
<3> xref:con-common-configuration-replicas-reference[The number of replica nodes] for the workers that run tasks.
<4> Authentication for the Kafka Connect cluster, specified as xref:type-KafkaClientAuthenticationTls-reference[mTLS], xref:type-KafkaClientAuthenticationOAuth-reference[token-based OAuth], SASL-based xref:type-KafkaClientAuthenticationScramSha256-reference[SCRAM-SHA-256]/xref:type-KafkaClientAuthenticationScramSha512-reference[SCRAM-SHA-512], or xref:type-KafkaClientAuthenticationPlain-reference[PLAIN].
By default, Kafka Connect connects to Kafka brokers using a plain text connection.
<5> xref:con-common-configuration-bootstrap-reference[Bootstrap server] for connection to the Kafka Connect cluster.
<6> xref:con-common-configuration-trusted-certificates-reference[TLS encryption] with key names under which TLS certificates are stored in X.509 format for the cluster. If certificates are stored in the same secret, it can be listed multiple times.
<7> xref:property-kafka-connect-config-reference[Kafka Connect configuration] of workers (not connectors).
Standard Apache Kafka configuration may be provided, restricted to those properties not managed directly by Strimzi.
<8> xref:type-Build-reference[Build configuration properties] for building a container image with connector plugins automatically.
<9> (Required) Configuration of the container registry where new images are pushed.
<10> (Required) List of connector plugins and their artifacts to add to the new container image. Each plugin must be configured with at least one `artifact`.
<11> xref:type-ExternalConfiguration-reference[External configuration for Kafka connectors] using environment variables, as shown here, or volumes.
You can also use _configuration provider plugins_ to xref:assembly-loading-config-with-providers-str[load configuration values from external sources].
<12> Requests for reservation of xref:con-common-configuration-resources-reference[supported resources], currently `cpu` and `memory`, and limits to specify the maximum resources that can be consumed.
<13> Specified xref:property-kafka-connect-logging-reference[Kafka Connect loggers and log levels] added directly (`inline`) or indirectly (`external`) through a ConfigMap. A custom ConfigMap must be placed under the `log4j.properties` or `log4j2.properties` key. For the Kafka Connect `log4j.rootLogger` logger, you can set the log level to INFO, ERROR, WARN, TRACE, DEBUG, FATAL or OFF.
<14> xref:con-common-configuration-healthchecks-reference[Healthchecks] to know when to restart a container (liveness) and when a container can accept traffic (readiness).
<15> xref:con-common-configuration-prometheus-reference[Prometheus metrics], which are enabled by referencing a ConfigMap containing configuration for the Prometheus JMX exporter in this example. You can enable metrics without further configuration using a reference to a ConfigMap containing an empty file under `metricsConfig.valueFrom.configMapKeyRef.key`.
<16> xref:con-common-configuration-jvm-reference[JVM configuration options] to optimize performance for the Virtual Machine (VM) running Kafka Connect.
<17> ADVANCED OPTION: xref:con-common-configuration-images-reference[Container image configuration], which is recommended only in special situations.
<18> SPECIALIZED OPTION: xref:type-Rack-reference[Rack awareness] configuration for the deployment. This is a specialized option intended for a deployment within the same location, not across regions. Use this option if you want connectors to consume from the closest replica rather than the leader replica. In certain cases, consuming from the closest replica can improve network utilization or reduce costs . The `topologyKey` must match a node label containing the rack ID. The example used in this configuration specifies a zone using the standard `{K8sZoneLabel}` label. To consume from the closest replica, enable the `RackAwareReplicaSelector`  in the Kafka broker configuration.
<19> xref:assembly-customizing-kubernetes-resources-str[Template customization]. Here a pod is scheduled with anti-affinity, so the pod is not scheduled on nodes with the same hostname.
<20> Environment variables are set for distributed tracing.

. Create or update the resource:
+
[source,shell,subs=+quotes]
kubectl apply -f _KAFKA-CONNECT-CONFIG-FILE_

. If authorization is enabled for Kafka Connect, xref:proc-configuring-kafka-connect-user-authorization-{context}[configure Kafka Connect users to enable access to the Kafka Connect consumer group and topics].

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

* link:{BookURLDeploying}#assembly-distributed-tracing-str[Introducing distributed tracing^]