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

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

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

As well as configuring Kafka, you can add configuration for ZooKeeper and the Strimzi Operators.
Common configuration properties, such as logging and healthchecks, are configured independently for each component.

Configuration options that are particularly important include the following:

* Resource requests (CPU / Memory)
* JVM options for maximum and minimum memory allocation
* Listeners for connecting clients to Kafka brokers (and authentication of clients)
* Authentication
* Storage
* Rack awareness
* Metrics
* Cruise Control for cluster rebalancing

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

.Kafka versions

The `inter.broker.protocol.version` property for the Kafka `config` must be the version supported by the specified Kafka version (`spec.kafka.version`).
The property represents the version of Kafka protocol used in a Kafka cluster.

From Kafka 3.0.0, when the `inter.broker.protocol.version` is set to `3.0` or higher, the `log.message.format.version` option is ignored and doesn't need to be set.

An update to the `inter.broker.protocol.version` is required when upgrading your Kafka version.
For more information, see xref:assembly-upgrading-kafka-versions-str[Upgrading Kafka].

.Managing TLS certificates
When deploying Kafka, the Cluster Operator automatically sets up and renews TLS certificates to enable encryption and authentication within your cluster.
If required, you can manually renew the cluster and clients CA certificates before their renewal period starts.
You can also replace the keys used by the cluster and clients CA certificates.
For more information, see xref:proc-renewing-ca-certs-manually-{context}[Renewing CA certificates manually] and xref:proc-replacing-private-keys-{context}[Replacing private keys].

.Example `Kafka` custom resource configuration
[source,yaml,subs="+attributes"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    replicas: 3 # <1>
    version: {DefaultKafkaVersion} # <2>
    logging: # <3>
      type: inline
      loggers:
        kafka.root.logger.level: INFO
    resources: # <4>
      requests:
        memory: 64Gi
        cpu: "8"
      limits:
        memory: 64Gi
        cpu: "12"
    readinessProbe: # <5>
      initialDelaySeconds: 15
      timeoutSeconds: 5
    livenessProbe:
      initialDelaySeconds: 15
      timeoutSeconds: 5
    jvmOptions: # <6>
      -Xms: 8192m
      -Xmx: 8192m
    image: my-org/my-image:latest # <7>
    listeners: # <8>
      - name: plain # <9>
        port: 9092 # <10>
        type: internal # <11>
        tls: false # <12>
        configuration:
          useServiceDnsDomain: true # <13>
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication: # <14>
          type: tls
      - name: external # <15>
        port: 9094
        type: route
        tls: true
        configuration:
          brokerCertChainAndKey: # <16>
            secretName: my-secret
            certificate: my-certificate.crt
            key: my-key.key
    authorization: # <17>
      type: simple
    config: # <18>
      auto.create.topics.enable: "false"
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      default.replication.factor: 3
      min.insync.replicas: 2
      inter.broker.protocol.version: "{DefaultInterBrokerVersion}"
    storage: # <19>
      type: persistent-claim # <20>
      size: 10000Gi
    rack: # <21>
      topologyKey: topology.kubernetes.io/zone
    metricsConfig: # <22>
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef: # <23>
          name: my-config-map
          key: my-key
    # ...
  zookeeper: # <24>
    replicas: 3 # <25>
    logging: # <26>
      type: inline
      loggers:
        zookeeper.root.logger: INFO
    resources:
      requests:
        memory: 8Gi
        cpu: "2"
      limits:
        memory: 8Gi
        cpu: "2"
    jvmOptions:
      -Xms: 4096m
      -Xmx: 4096m
    storage:
      type: persistent-claim
      size: 1000Gi
    metricsConfig:
      # ...
  entityOperator: # <27>
    tlsSidecar: # <28>
      resources:
        requests:
          cpu: 200m
          memory: 64Mi
        limits:
          cpu: 500m
          memory: 128Mi
    topicOperator:
      watchedNamespace: my-topic-namespace
      reconciliationIntervalSeconds: 60
      logging: # <29>
        type: inline
        loggers:
          rootLogger.level: INFO
      resources:
        requests:
          memory: 512Mi
          cpu: "1"
        limits:
          memory: 512Mi
          cpu: "1"
    userOperator:
      watchedNamespace: my-topic-namespace
      reconciliationIntervalSeconds: 60
      logging: # <30>
        type: inline
        loggers:
          rootLogger.level: INFO
      resources:
        requests:
          memory: 512Mi
          cpu: "1"
        limits:
          memory: 512Mi
          cpu: "1"
  kafkaExporter: # <31>
    # ...
  cruiseControl: # <32>
    # ...
----
<1> The number of replica nodes.
<2> Kafka version, which can be changed to a supported version by following the upgrade procedure.
<3> Kafka loggers and log levels added directly (`inline`) or indirectly (`external`) through a ConfigMap. A custom Log4j configuration must be placed under the `log4j.properties` key in the ConfigMap. For the Kafka `kafka.root.logger.level` logger, you can set the log level to INFO, ERROR, WARN, TRACE, DEBUG, FATAL or OFF.
<4> Requests for reservation of supported resources, currently `cpu` and `memory`, and limits to specify the maximum resources that can be consumed.
<5> Healthchecks to know when to restart a container (liveness) and when a container can accept traffic (readiness).
<6> JVM configuration options to optimize performance for the Virtual Machine (VM) running Kafka.
<7> ADVANCED OPTION: Container image configuration, which is recommended only in special situations.
<8> Listeners configure how clients connect to the Kafka cluster via bootstrap addresses. Listeners are configured as _internal_ or _external_ listeners for connection from inside or outside the Kubernetes cluster.
<9> Name to identify the listener. Must be unique within the Kafka cluster.
<10> Port number used by the listener inside Kafka. The port number has to be unique within a given Kafka cluster. Allowed port numbers are 9092 and higher with the exception of ports 9404 and 9999, which are already used for Prometheus and JMX. Depending on the listener type, the port number might not be the same as the port number that connects Kafka clients.
<11> Listener type specified as `internal` or `cluster-ip` (to expose Kafka using per-broker `ClusterIP` services), or for external listeners, as `route` (OpenShift only), `loadbalancer`, `nodeport` or `ingress` (Kubernetes only).
<12> Enables TLS encryption for each listener. Default is `false`. TLS encryption has to be enabled, by setting it to `true`, for `route` and `ingress` type listeners.
<13> Defines whether the fully-qualified DNS names including the cluster service suffix (usually `.cluster.local`) are assigned.
<14> Listener authentication mechanism specified as mTLS, SCRAM-SHA-512, or token-based OAuth 2.0.
<15> External listener configuration specifies how the Kafka cluster is exposed outside Kubernetes, such as through a `route`, `loadbalancer` or `nodeport`.
<16> Optional configuration for a Kafka listener certificate managed by an external CA (certificate authority). The `brokerCertChainAndKey` specifies a `Secret` that contains a server certificate and a private key. You can configure Kafka listener certificates on any listener with enabled TLS encryption.
<17> Authorization enables simple, OAUTH 2.0, or OPA authorization on the Kafka broker. Simple authorization uses the `AclAuthorizer` Kafka plugin.
<18> Broker configuration. Standard Apache Kafka configuration may be provided, restricted to those properties not managed directly by Strimzi.
<19> Storage size for persistent volumes may be increased and additional volumes may be added to JBOD storage.
<20> Persistent storage has additional configuration options, such as a storage `id` and `class` for dynamic volume provisioning.
<21> Rack awareness configuration to spread replicas across different racks, data centers, or availability zones. 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.
<22> Prometheus metrics enabled. In this example, metrics are configured for the Prometheus JMX Exporter (the default metrics exporter).
<23> Rules for exporting metrics in Prometheus format to a Grafana dashboard through the Prometheus JMX Exporter, which are enabled by referencing a ConfigMap containing configuration for the Prometheus JMX exporter. You can enable metrics without further configuration using a reference to a ConfigMap containing an empty file under `metricsConfig.valueFrom.configMapKeyRef.key`.
<24> ZooKeeper-specific configuration, which contains properties similar to the Kafka configuration.
<25> The number of ZooKeeper nodes. ZooKeeper clusters or ensembles usually run with an odd number of nodes, typically three, five, or seven. The majority of nodes must be available in order to maintain an effective quorum.
If the ZooKeeper cluster loses its quorum, it will stop responding to clients and the Kafka brokers will stop working.
Having a stable and highly available ZooKeeper cluster is crucial for Strimzi.
<26> ZooKeeper loggers and log levels.
<27> Entity Operator configuration, which specifies the configuration for the Topic Operator and User Operator.
<28> Entity Operator TLS sidecar configuration. Entity Operator uses the TLS sidecar for secure communication with ZooKeeper.
<29> Specified Topic Operator loggers and log levels. This example uses `inline` logging.
<30> Specified User Operator loggers and log levels.
<31> Kafka Exporter configuration. Kafka Exporter is an optional component for extracting metrics data from Kafka brokers, in particular consumer lag data. For Kafka Exporter to be able to work properly, consumer groups need to be in use. 
<32> Optional configuration for Cruise Control, which is used to rebalance the Kafka cluster.