// 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.

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

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

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

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

.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"]
----
# Basic configuration (required)
apiVersion: {KafkaConnectApiVersion}
kind: KafkaConnect # <1>
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true" # <2>
# Deployment specifications
spec:
  # Replicas (required)
  replicas: 3 # <3>
  # Bootstrap servers (required)
  bootstrapServers: my-cluster-kafka-bootstrap:9092 # <4>
  # Kafka Connect configuration (recommended)
  config: # <5>
    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
  # Resources requests and limits (recommended)
  resources: # <6>
    requests:
      cpu: "1"
      memory: 2Gi
    limits:
      cpu: "2"
      memory: 2Gi
  # Authentication (optional)
  authentication: # <7>
    type: tls
    certificateAndKey:
      certificate: source.crt
      key: source.key
      secretName: my-user-source
  # TLS configuration (optional)
  tls: # <8>
    trustedCertificates:
      - secretName: my-cluster-cluster-cert
        pattern: "*.crt"
      - secretName: my-cluster-cluster-cert
        pattern: "*.crt"
  # Build configuration (optional)
  build: # <9>
    output: # <10>
      type: docker
      image: my-registry.io/my-org/my-connect-cluster:latest
      pushSecret: my-registry-credentials
    plugins: # <11>
      - name: connector-1
        artifacts:
          - type: tgz
            url: <url_to_download_connector_1_artifact>
            sha512sum: <SHA-512_checksum_of_connector_1_artifact>
      - name: connector-2
        artifacts:
          - type: jar
            url: <url_to_download_connector_2_artifact>
            sha512sum: <SHA-512_checksum_of_connector_2_artifact>
  # Logging configuration (optional)
  logging: # <12>
    type: inline
    loggers:
      log4j.rootLogger: INFO
  # Readiness probe (optional)
  readinessProbe: # <13>
    initialDelaySeconds: 15
    timeoutSeconds: 5
  # Liveness probe (optional)
  livenessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  # Metrics configuration (optional)
  metricsConfig: # <14>
    type: jmxPrometheusExporter
    valueFrom:
      configMapKeyRef:
        name: my-config-map
        key: my-key
  # JVM options (optional)
  jvmOptions: # <15>
    "-Xmx": "1g"
    "-Xms": "1g"
  # Custom image (optional)
  image: my-org/my-image:latest # <16>
  # Rack awareness (optional)
  rack:
    topologyKey: topology.kubernetes.io/zone # <17>
  # Pod and container template (optional)
  template: # <18>
    pod:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            - labelSelector:
                matchExpressions:
                  - key: application
                    operator: In
                    values:
                      - postgresql
                      - mongodb
              topologyKey: "kubernetes.io/hostname"
    connectContainer: # <19>
      env:
        - name: OTEL_SERVICE_NAME
          value: my-otel-service
        - name: OTEL_EXPORTER_OTLP_ENDPOINT
          value: "http://otlp-host:4317"
        - name: AWS_ACCESS_KEY_ID
          valueFrom:
            secretKeyRef:
              name: aws-creds
              key: awsAccessKey
        - name: AWS_SECRET_ACCESS_KEY
          valueFrom:
            secretKeyRef:
              name: aws-creds
              key: awsSecretAccessKey
  # Tracing configuration (optional)
  tracing:
    type: opentelemetry # <20>
----
<1> Use `KafkaConnect`.
<2> Enables the use of `KafkaConnector` resources to start, stop, and manage connector instances.
<3> The number of replica nodes for the workers that run tasks.
<4> Bootstrap address for connection to the Kafka cluster. The address takes the format `<cluster_name>-kafka-bootstrap:<port_number>`. The Kafka cluster doesn't need to be managed by Strimzi or deployed to a Kubernetes cluster.
<5> Kafka Connect configuration of workers (not connectors) that run connectors and their tasks.
Standard Apache Kafka configuration may be provided, restricted to those properties not managed directly by Strimzi.
In this example, JSON convertors are specified.
A replication factor of 3 is set for the internal topics used by Kafka Connect (minimum requirement for production environment).
Changing the replication factor after the topics have been created has no effect.
<6> Requests for reservation of supported resources, currently `cpu` and `memory`, and limits to specify the maximum resources that can be consumed.
<7> 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.
<8> TLS configuration for encrypted connections to the Kafka cluster, with trusted certificates stored in X.509 format within the specified secrets.
<9> Build configuration properties for building a container image with connector plugins automatically.
<10> (Required) Configuration of the container registry where new images are pushed.
<11> (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`.
<12> 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.
<13> Healthchecks to know when to restart a container (liveness) and when a container can accept traffic (readiness).
<14> 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`.
<15> JVM configuration options to optimize performance for the Virtual Machine (VM) running Kafka Connect.
<16> ADVANCED OPTION: Container image configuration, which is recommended only in special situations.
<17> 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.
<18> Template customization. Here a pod is scheduled with anti-affinity, so the pod is not scheduled on nodes with the same hostname.
<19> Environment variables are set for distributed tracing and to pass credentials to connectors.
<20> Distributed tracing is enabled by using OpenTelemetry.