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

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

Use the properties of the `Kafka` 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.

This procedure shows only some of the possible configuration options, but those that are particularly important include:

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

.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 link:{BookURLDeploying}#assembly-upgrading-kafka-versions-str[Upgrading Kafka].

.Prerequisites

* A Kubernetes cluster
* A running Cluster Operator

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

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

.Procedure

. Edit the `spec` properties for the `Kafka` resource.
+
The properties you can configure are shown in this example 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: {LogMsgVersHigher}
      ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" <19>
      ssl.enabled.protocols: "TLSv1.2"
      ssl.protocol: "TLSv1.2"
    storage: <20>
      type: persistent-claim <21>
      size: 10000Gi <22>
    rack: <23>
      topologyKey: topology.kubernetes.io/zone
    metricsConfig: <24>
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef: <25>
          name: my-config-map
          key: my-key
    # ...
  zookeeper: <26>
    replicas: 3 <27>
    logging: <28>
      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: <29>
    tlsSidecar: <30>
      resources:
        requests:
          cpu: 200m
          memory: 64Mi
        limits:
          cpu: 500m
          memory: 128Mi
    topicOperator:
      watchedNamespace: my-topic-namespace
      reconciliationIntervalSeconds: 60
      logging: <31>
        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: <32>
        type: inline
        loggers:
          rootLogger.level: INFO
      resources:
        requests:
          memory: 512Mi
          cpu: "1"
        limits:
          memory: 512Mi
          cpu: "1"
  kafkaExporter: <33>
    # ...
  cruiseControl: <34>
    # ...
    tlsSidecar: <35>
    # ...
----
<1> xref:con-common-configuration-replicas-reference[The number of replica nodes]. If your cluster already has topics defined, you can
xref:scaling-clusters-{context}[scale clusters].
<2> Kafka version, which can be changed to a supported version by following link:{BookURLDeploying}#assembly-upgrade-str[the upgrade procedure].
<3> Specified xref:property-kafka-logging-reference[Kafka loggers and log levels] added directly (`inline`) or indirectly (`external`) through a ConfigMap. A custom ConfigMap must be placed under the `log4j.properties` key. 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 xref:con-common-configuration-resources-reference[supported resources], currently `cpu` and `memory`, and limits to specify the maximum resources that can be consumed.
<5> xref:con-common-configuration-healthchecks-reference[Healthchecks] to know when to restart a container (liveness) and when a container can accept traffic (readiness).
<6> xref:con-common-configuration-jvm-reference[JVM configuration options] to optimize performance for the Virtual Machine (VM) running Kafka.
<7> ADVANCED OPTION: xref:con-common-configuration-images-reference[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 xref:assembly-securing-kafka-brokers-str[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 for external listeners, as `route`, `loadbalancer`, `nodeport` or `ingress`.
<12> Enables TLS encryption for each listener. Default is `false`. TLS encryption is not required for `route` listeners.
<13> Defines whether the fully-qualified DNS names including the cluster service suffix (usually `.cluster.local`) are assigned.
<14> Listener authentication mechanism xref:assembly-securing-kafka-brokers-str[specified as mutual TLS, SCRAM-SHA-512 or token-based OAuth 2.0].
<15> External listener configuration specifies xref:assembly-accessing-kafka-outside-cluster-str[how the Kafka cluster is exposed outside Kubernetes, such as through a `route`, `loadbalancer` or `nodeport`].
<16> Optional configuration for a xref:kafka-listener-certificates-str[Kafka listener certificate] managed by an external 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 xref:con-securing-kafka-authorization-str[enables simple, OAUTH 2.0, or OPA authorization on the Kafka broker.] Simple authorization uses the `AclAuthorizer` Kafka plugin.
<18> The `config` specifies the broker configuration. xref:property-kafka-config-reference[Standard Apache Kafka configuration may be provided, restricted to those properties not managed directly by Strimzi].
<19> xref:con-common-configuration-ssl-reference[SSL properties for listeners with TLS encryption enabled to enable a specific _cipher suite_ or TLS version].
<20> xref:assembly-storage-{context}[Storage] is configured as `ephemeral`, `persistent-claim` or `jbod`.
<21> Storage size for xref:proc-resizing-persistent-volumes-{context}[persistent volumes may be increased] and additional xref:proc-adding-volumes-to-jbod-storage-{context}[volumes may be added to JBOD storage].
<22> Persistent storage has xref:ref-persistent-storage-{context}[additional configuration options], such as a storage `id` and `class` for dynamic volume provisioning.
<23> xref:type-Rack-reference[Rack awareness] is configured to spread replicas across different racks. A `topologykey` must match the label of a cluster node.
<24> xref:con-common-configuration-prometheus-reference[Prometheus metrics] enabled. In this example, metrics are configured for the Prometheus JMX Exporter (the default metrics exporter).
<25> Prometheus rules for exporting metrics 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`.
<26> ZooKeeper-specific configuration, which contains properties similar to the Kafka configuration.
<27> xref:con-common-configuration-replicas-reference[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.
<28> Specified xref:property-zookeeper-logging-reference[ZooKeeper loggers and log levels].
<29> Entity Operator configuration, which xref:assembly-kafka-entity-operator-{context}[specifies the configuration for the Topic Operator and User Operator].
<30> Entity Operator xref:type-TlsSidecar-reference[TLS sidecar configuration]. Entity Operator uses the TLS sidecar for secure communication with ZooKeeper.
<31> Specified xref:property-topic-operator-logging-reference[Topic Operator loggers and log levels]. This example uses `inline` logging.
<32> Specified xref:property-user-operator-logging-reference[User Operator loggers and log levels].
<33> Kafka Exporter configuration. link:{BookURLDeploying}#assembly-metrics-kafka-exporter-str[Kafka Exporter] is an optional component for extracting metrics data from Kafka brokers, in particular consumer lag data.
<34> Optional configuration for Cruise Control, which is used to xref:cruise-control-concepts-str[rebalance the Kafka cluster].
<35> Cruise Control xref:type-TlsSidecar-reference[TLS sidecar configuration]. Cruise Control uses the TLS sidecar for secure communication with ZooKeeper.

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