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

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

[role="_abstract"]
Update the `spec` properties of the `KafkaConnect` custom resource to configure your Kafka Connect deployment.

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.

For a deeper understanding of the Kafka Connect cluster configuration options, refer to the link:{BookURLConfiguring}[Strimzi Custom Resource API Reference^].

.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 xref: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 xref:con-high-volume-config-properties-{context}[Handling high volumes of messages].

.Example `KafkaConnect` custom resource 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.3.Final/debezium-connector-postgres-2.1.3.Final-plugin.tar.gz
            sha512sum: c4ddc97846de561755dc0b021a62aba656098829c70eb3ade3b817ce06d852ca12ae50c0281cc791a5a131cb7fc21fb15f4b8ee76c6cae5dd07f9c11cb7c6e79
      - name: camel-telegram
        artifacts:
          - type: tgz
            url: https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-telegram-kafka-connector/0.11.5/camel-telegram-kafka-connector-0.11.5-package.tar.gz
            sha512sum: d6d9f45e0d1dbfcc9f6d1c7ca2046168c764389c78bc4b867dab32d24f710bb74ccf2a007d7d7a8af2dfca09d9a52ccbc2831fc715c195a3634cca055185bd91
  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: OTEL_SERVICE_NAME
          value: my-otel-service
        - name: OTEL_EXPORTER_OTLP_ENDPOINT
          value: "http://otlp-host:4317"
  tracing:
    type: opentelemetry # <21>
----
<1> Use `KafkaConnect`.
<2> Enables KafkaConnectors for the Kafka Connect cluster.
<3> The number of replica nodes for the workers that run tasks.
<4> Authentication for the Kafka Connect cluster, specified as mTLS, token-based OAuth, SASL-based SCRAM-SHA-256/SCRAM-SHA-512, or PLAIN.
By default, Kafka Connect connects to Kafka brokers using a plain text connection.
<5> Bootstrap server for connection to the Kafka cluster.
<6> 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> Kafka Connect configuration of workers (not connectors).
Standard Apache Kafka configuration may be provided, restricted to those properties not managed directly by Strimzi.
<8> 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> External configuration for connectors using environment variables, as shown here, or volumes.
You can also use configuration provider plugins to load configuration values from external sources.
<12> Requests for reservation of supported resources, currently `cpu` and `memory`, and limits to specify the maximum resources that can be consumed.
<13> Specified Kafka Connect loggers and log levels added directly (`inline`) or indirectly (`external`) through a ConfigMap. A custom Log4j configuration must be placed under the `log4j.properties` or `log4j2.properties` key in the ConfigMap. For the Kafka Connect `log4j.rootLogger` logger, you can set the log level to INFO, ERROR, WARN, TRACE, DEBUG, FATAL or OFF.
<14> Healthchecks to know when to restart a container (liveness) and when a container can accept traffic (readiness).
<15> 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> JVM configuration options to optimize performance for the Virtual Machine (VM) running Kafka Connect.
<17> ADVANCED OPTION: Container image configuration, which is recommended only in special situations.
<18> SPECIALIZED OPTION: 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> 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.
<21> Distributed tracing is enabled by using OpenTelemetry.