Configures a Kafka cluster.

[id='property-kafka-listeners-{context}']
=== `listeners`

Use the `listeners` property to configure listeners to provide access to Kafka brokers.

.Example configuration of a plain (unencrypted) listener without authentication

[source,yaml,subs=attributes+]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
spec:
  kafka:
    # ...
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
    # ...
  zookeeper:
    # ...
----

[id='property-kafka-config-{context}']
=== `config`

Use the `config` properties to configure Kafka broker options as keys.

Standard Apache Kafka configuration may be provided, restricted to those properties not managed directly by Strimzi.

Configuration options that cannot be configured relate to:

* Security (Encryption, Authentication, and Authorization)
* Listener configuration
* Broker ID configuration
* Configuration of log data directories
* Inter-broker communication
* ZooKeeper connectivity

The values can be one of the following JSON types:

* String
* Number
* Boolean

You can specify and configure the options listed in the {ApacheKafkaBrokerConfig} with the exception of those options that are managed directly by Strimzi.
Specifically, all configuration options with keys equal to or starting with one of the following strings are forbidden:

* `listeners`
* `advertised.`
* `broker.`
* `listener.`
* `host.name`
* `port`
* `inter.broker.listener.name`
* `sasl.`
* `ssl.`
* `security.`
* `password.`
* `principal.builder.class`
* `log.dir`
* `zookeeper.connect`
* `zookeeper.set.acl`
* `authorizer.`
* `super.user`

When a forbidden option is present in the `config` property, it is ignored and a warning message is printed to the Cluster Operator log file.
All other supported options are passed to Kafka.

There are exceptions to the forbidden options.
For client connection using a specific _cipher suite_ for a TLS version, you can xref:con-common-configuration-ssl-reference[configure allowed `ssl` properties].
You can also configure the `zookeeper.connection.timeout.ms` property to set the maximum time allowed for establishing a ZooKeeper connection.

.Example Kafka broker configuration
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    config:
      num.partitions: 1
      num.recovery.threads.per.data.dir: 1
      default.replication.factor: 3
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 1
      log.retention.hours: 168
      log.segment.bytes: 1073741824
      log.retention.check.interval.ms: 300000
      num.network.threads: 3
      num.io.threads: 8
      socket.send.buffer.bytes: 102400
      socket.receive.buffer.bytes: 102400
      socket.request.max.bytes: 104857600
      group.initial.rebalance.delay.ms: 0
      ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384"
      ssl.enabled.protocols: "TLSv1.2"
      ssl.protocol: "TLSv1.2"
      zookeeper.connection.timeout.ms: 6000
    # ...
----

[id='property-kafka-brokerRackInitImage-{context}']
=== `brokerRackInitImage`

When rack awareness is enabled, Kafka broker pods use init container to collect the labels from the Kubernetes cluster nodes.
The container image used for this container can be configured using the `brokerRackInitImage` property.
When the `brokerRackInitImage` field is missing, the following images are used in order of priority:

. Container image specified in `STRIMZI_DEFAULT_KAFKA_INIT_IMAGE` environment variable in the Cluster Operator configuration.
. `{DockerKafkaInit}` container image.

.Example `brokerRackInitImage` configuration
[source,yaml,subs=attributes+]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    rack:
      topologyKey: topology.kubernetes.io/zone
    brokerRackInitImage: my-org/my-image:latest
    # ...
----

NOTE: Overriding container images is recommended only in special situations, where you need to use a different container registry.
For example, because your network does not allow access to the container registry used by Strimzi. In this case, you should either copy the Strimzi images or build them from the source.
If the configured image is not compatible with Strimzi images, it might not work properly.

[id='property-kafka-logging-{context}']
=== `logging`

Kafka has its own configurable loggers:

* `log4j.logger.org.I0Itec.zkclient.ZkClient`
* `log4j.logger.org.apache.zookeeper`
* `log4j.logger.kafka`
* `log4j.logger.org.apache.kafka`
* `log4j.logger.kafka.request.logger`
* `log4j.logger.kafka.network.Processor`
* `log4j.logger.kafka.server.KafkaApis`
* `log4j.logger.kafka.network.RequestChannel$`
* `log4j.logger.kafka.controller`
* `log4j.logger.kafka.log.LogCleaner`
* `log4j.logger.state.change.logger`
* `log4j.logger.kafka.authorizer.logger`

Kafka uses the Apache `log4j` logger implementation.

Use the `logging` property to configure loggers and logger levels.

You can set the log levels by specifying the logger and level directly (inline) or use a custom (external) ConfigMap.
If a ConfigMap is used, you set `logging.valueFrom.configMapKeyRef.name` property to the name of the ConfigMap containing the external logging configuration. Inside the ConfigMap, the logging configuration is described using `log4j.properties`. Both `logging.valueFrom.configMapKeyRef.name` and `logging.valueFrom.configMapKeyRef.key` properties are mandatory. A ConfigMap using the exact logging configuration specified is created with the custom resource when the Cluster Operator is running, then recreated after each reconciliation. If you do not specify a custom ConfigMap, default logging settings are used. If a specific logger value is not set, upper-level logger settings are inherited for that logger.
For more information about log levels, see {ApacheLoggers}.

Here we see examples of `inline` and `external` logging.

.Inline logging
[source,yaml,subs="+quotes,attributes"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
spec:
  # ...
  kafka:
    # ...
    logging:
      type: inline
      loggers:
        kafka.root.logger.level: "INFO"
  # ...
----

.External logging
[source,yaml,subs="+quotes,attributes"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
spec:
  # ...
  logging:
    type: external
    valueFrom:
      configMapKeyRef:
        name: customConfigMap
        key: kafka-log4j.properties
  # ...
----

Any available loggers that are not configured have their level set to `OFF`.

If Kafka was deployed using the Cluster Operator,
changes to Kafka logging levels are applied dynamically.

If you use external logging, a rolling update is triggered when logging appenders are changed.

.Garbage collector (GC)

Garbage collector logging can also be enabled (or disabled) using the xref:con-common-configuration-garbage-collection-reference[`jvmOptions` property].
