<commit_before>== Cluster Controller

The cluster controller is in charge of deploying a Kafka cluster alongside a Zookeeper ensemble. As part of the Kafka cluster,
it can also deploy the topic controller which provides operator-style topic management via ConfigMaps.
The cluster controller is also able to deploy a Kafka Connect cluster which connects to an existing Kafka cluster.
On OpenShift such a cluster can be deployed using the Source2Image feature, providing an easy way of including more connectors.

image::cluster_controller.png[Cluster Controller]

When the cluster controller is up, it starts to "watch" for ConfigMaps containing the Kafka or Kafka Connect
cluster configuration. Such ConfigMaps need to have a specific label which is, by default, `strimzi.io/kind=cluster`
(as described <<Format of the cluster ConfigMap,later>>) that can be changed through a corresponding environment variable.

When a new ConfigMap is created in the Kubernetes/OpenShift cluster, the controller gets the cluster configuration from
its `data` section and starts creating a new Kafka or Kafka Connect cluster by creating the necessary Kubernetes/OpenShift
resources, such as StatefulSets, ConfigMaps, Services etc.

Every time the ConfigMap is updated by the user with some changes in the `data` section, the controller performs corresponding
updates on the Kubernetes/OpenShift resources which make up the Kafka or Kafka Connect cluster. Resources are either patched
or deleted and then re-created in order to make the Kafka or Kafka Connect cluster reflect the state of the cluster ConfigMap.
This might cause a rolling update which might lead to service disruption.

Finally, when the ConfigMap is deleted, the controller starts to un-deploy the cluster deleting all the related Kubernetes/OpenShift
resources.

=== Reconciliation

Although the controller reacts to all notifications about the cluster ConfigMaps received from the Kubernetes/OpenShift cluster,
if the controller is not running, or if a notification is not received for any reason, the ConfigMaps will get out of sync
with the state of the running Kubernetes/OpenShift cluster.

In order to handle failovers properly, a periodic reconciliation process is executed by the cluster controller so
that it can compare the state of the ConfigMaps with the current cluster deployment in order to have
a consistent state across all of them.

[[config_map_details]]
=== Format of the cluster ConfigMap

The controller watches for ConfigMaps having the label `strimzi.io/kind=cluster` in order to find and get
configuration for a Kafka or Kafka Connect cluster to deploy.

In order to distinguish which "type" of cluster to deploy, Kafka or Kafka Connect, the controller checks the
`strimzi.io/type` label which can have one of the the following values :

* `kafka`: the ConfigMap provides configuration for a Kafka cluster (with Zookeeper ensemble) deployment
* `kafka-connect`: the ConfigMap provides configuration for a Kafka Connect cluster deployment
* `kafka-connect-s2i`: the ConfigMap provides configuration for a Kafka Connect cluster deployment using Build and Source2Image
features (works only with OpenShift)

Whatever other labels are applied to the ConfigMap will also be applied to the Kubernetes/OpenShift resources making
up the Kafka or Kafka Connect cluster. This provides a convenient mechanism for those resource to be labelled
in whatever way the user requires.

The `data` section of such ConfigMaps contains different keys depending on the "type" of deployment as described in the 
following sections.

[[kafka_config_map_details]]
==== Kafka

In order to configure a Kafka cluster deployment, it's possible to specify the following fields in the `data` section of 
the related ConfigMap :

* `kafka-nodes`: number of Kafka broker nodes. Default is 3
* `kafka-image`: the Docker image to use for the Kafka brokers. Default is `strimzi/kafka:latest`
* `kafka-healthcheck-delay`: the initial delay for the liveness and readiness probes for each Kafka broker node. Default is 15
* `kafka-healthcheck-timeout`: the timeout on the liveness and readiness probes for each Kafka broker node. Default is 5
* `zookeeper-nodes`: number of Zookeeper nodes
* `zookeeper-image`: the Docker image to use for the Zookeeper nodes. Default is `strimzi/zookeeper:latest`
* `zookeeper-healthcheck-delay`: the initial delay for the liveness and readiness probes for each Zookeeper node. Default is 15
* `zookeeper-healthcheck-timeout`: the timeout on the liveness and readiness probes for each Zookeeper node. Default is 5
* `KAFKA_DEFAULT_REPLICATION_FACTOR`: the default replication factors for automatically created topics. It sets the 
`default.replication.factor` property in the properties configuration file used by Kafka broker nodes on startup. Default is 3
* `KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR`: the replication factor for the offsets topic. It sets the  
`offsets.topic.replication.factor` property in the properties configuration file used by Kafka broker nodes on startup. Default is 3
* `KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR`: the replication factor for the transaction topic. It sets the 
`transaction.state.log.replication.factor` property in the properties configuration file used by Kafka broker nodes on startup. Default is 3
* `kafka-storage`: a JSON string representing the storage configuration for the Kafka broker nodes. See related section
* `zookeeper-storage`: a JSON string representing the storage configuration for the Zookeeper nodes. See related section
* `kafka-metrics-config`: a JSON string representing the JMX exporter configuration for exposing metrics from Kafka broker nodes.
 Removing this field means having no metrics exposed.
* `zookeeper-metrics-config`: a JSON string representing the JMX exporter configuration for exposing metrics from Zookeeper nodes.
 Removing this field means having no metrics exposed.
* `topic-controller-config`: a JSON string representing the topic controller configuration. See the <<topic_controller_json_config>>
documentation for further details. More info about the topic controller in the related <<Topic Controller>> documentation page.
 
The following is an example of a ConfigMap for a Kafka cluster.

.Example Kafka cluster ConfigMap
[source,yaml,options="nowrap"]
----
apiVersion: v1
kind: ConfigMap
metadata:
  name: my-cluster
  labels:
    strimzi.io/kind: cluster
    strimzi.io/type: kafka
data:
  kafka-nodes: "3"
  kafka-image: "strimzi/kafka:latest"
  kafka-healthcheck-delay: "15"
  kafka-healthcheck-timeout: "5"
  zookeeper-nodes: "1"
  zookeeper-image: "strimzi/zookeeper:latest"
  zookeeper-healthcheck-delay: "15"
  zookeeper-healthcheck-timeout: "5"
  KAFKA_DEFAULT_REPLICATION_FACTOR: "3"
  KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: "3"
  KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: "3"
  kafka-storage: |-
    { "type": "ephemeral" }
  zookeeper-storage: |-
    { "type": "ephemeral" }
  kafka-metrics-config: |-
    {
      "lowercaseOutputName": true,
      "rules": [
          {
            "pattern": "kafka.server<type=(.+), name=(.+)PerSec\\w*><>Count",
            "name": "kafka_server_$1_$2_total"
          },
          {
            "pattern": "kafka.server<type=(.+), name=(.+)PerSec\\w*, topic=(.+)><>Count",
            "name": "kafka_server_$1_$2_total",
            "labels":
            {
              "topic": "$3"
            }
          }
      ]
    }
  zookeeper-metrics-config: |-
    {
      "lowercaseOutputName": true
    }
----

The resources created by the cluster controller into the Kubernetes/OpenShift cluster will be the following :

* `[cluster-name]-zookeeper` StatefulSet which is in charge to create the Zookeeper node pods
* `[cluster-name]-kafka` StatefulSet which is in charge to create the Kafka broker pods
* `[cluster-name]-zookeeper-headless` Service needed to have DNS resolve the Zookeeper pods IP addresses directly
* `[cluster-name]-kafka-headless` Service needed to have DNS resolve the Kafka broker pods IP addresses directly
* `[cluster-name]-zookeeper` Service used by Kafka brokers to connect to Zookeeper nodes as clients
* `[cluster-name]-kafka` Service can be used as bootstrap servers for Kafka clients
* `[cluster-name]-zookeeper-metrics-config` ConfigMap which contains the Zookeeper metrics configuration and mounted as
a volume by the Zookeeper node pods
* `[cluster-name]-kafka-metrics-config` ConfigMap which contains the Kafka metrics configuration and mounted as
a volume by the Kafka broker pods

===== Storage

Both Kafka and Zookeeper save data to files.
<commit_msg>Added warning about running cluster with "ephemeral" storage<commit_after>== Cluster Controller

The cluster controller is in charge of deploying a Kafka cluster alongside a Zookeeper ensemble. As part of the Kafka cluster,
it can also deploy the topic controller which provides operator-style topic management via ConfigMaps.
The cluster controller is also able to deploy a Kafka Connect cluster which connects to an existing Kafka cluster.
On OpenShift such a cluster can be deployed using the Source2Image feature, providing an easy way of including more connectors.

image::cluster_controller.png[Cluster Controller]

When the cluster controller is up, it starts to "watch" for ConfigMaps containing the Kafka or Kafka Connect
cluster configuration. Such ConfigMaps need to have a specific label which is, by default, `strimzi.io/kind=cluster`
(as described <<Format of the cluster ConfigMap,later>>) that can be changed through a corresponding environment variable.

When a new ConfigMap is created in the Kubernetes/OpenShift cluster, the controller gets the cluster configuration from
its `data` section and starts creating a new Kafka or Kafka Connect cluster by creating the necessary Kubernetes/OpenShift
resources, such as StatefulSets, ConfigMaps, Services etc.

Every time the ConfigMap is updated by the user with some changes in the `data` section, the controller performs corresponding
updates on the Kubernetes/OpenShift resources which make up the Kafka or Kafka Connect cluster. Resources are either patched
or deleted and then re-created in order to make the Kafka or Kafka Connect cluster reflect the state of the cluster ConfigMap.
This might cause a rolling update which might lead to service disruption.

Finally, when the ConfigMap is deleted, the controller starts to un-deploy the cluster deleting all the related Kubernetes/OpenShift
resources.

=== Reconciliation

Although the controller reacts to all notifications about the cluster ConfigMaps received from the Kubernetes/OpenShift cluster,
if the controller is not running, or if a notification is not received for any reason, the ConfigMaps will get out of sync
with the state of the running Kubernetes/OpenShift cluster.

In order to handle failovers properly, a periodic reconciliation process is executed by the cluster controller so
that it can compare the state of the ConfigMaps with the current cluster deployment in order to have
a consistent state across all of them.

[[config_map_details]]
=== Format of the cluster ConfigMap

The controller watches for ConfigMaps having the label `strimzi.io/kind=cluster` in order to find and get
configuration for a Kafka or Kafka Connect cluster to deploy.

In order to distinguish which "type" of cluster to deploy, Kafka or Kafka Connect, the controller checks the
`strimzi.io/type` label which can have one of the the following values :

* `kafka`: the ConfigMap provides configuration for a Kafka cluster (with Zookeeper ensemble) deployment
* `kafka-connect`: the ConfigMap provides configuration for a Kafka Connect cluster deployment
* `kafka-connect-s2i`: the ConfigMap provides configuration for a Kafka Connect cluster deployment using Build and Source2Image
features (works only with OpenShift)

Whatever other labels are applied to the ConfigMap will also be applied to the Kubernetes/OpenShift resources making
up the Kafka or Kafka Connect cluster. This provides a convenient mechanism for those resource to be labelled
in whatever way the user requires.

The `data` section of such ConfigMaps contains different keys depending on the "type" of deployment as described in the 
following sections.

[[kafka_config_map_details]]
==== Kafka

In order to configure a Kafka cluster deployment, it's possible to specify the following fields in the `data` section of 
the related ConfigMap :

* `kafka-nodes`: number of Kafka broker nodes. Default is 3
* `kafka-image`: the Docker image to use for the Kafka brokers. Default is `strimzi/kafka:latest`
* `kafka-healthcheck-delay`: the initial delay for the liveness and readiness probes for each Kafka broker node. Default is 15
* `kafka-healthcheck-timeout`: the timeout on the liveness and readiness probes for each Kafka broker node. Default is 5
* `zookeeper-nodes`: number of Zookeeper nodes
* `zookeeper-image`: the Docker image to use for the Zookeeper nodes. Default is `strimzi/zookeeper:latest`
* `zookeeper-healthcheck-delay`: the initial delay for the liveness and readiness probes for each Zookeeper node. Default is 15
* `zookeeper-healthcheck-timeout`: the timeout on the liveness and readiness probes for each Zookeeper node. Default is 5
* `KAFKA_DEFAULT_REPLICATION_FACTOR`: the default replication factors for automatically created topics. It sets the 
`default.replication.factor` property in the properties configuration file used by Kafka broker nodes on startup. Default is 3
* `KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR`: the replication factor for the offsets topic. It sets the  
`offsets.topic.replication.factor` property in the properties configuration file used by Kafka broker nodes on startup. Default is 3
* `KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR`: the replication factor for the transaction topic. It sets the 
`transaction.state.log.replication.factor` property in the properties configuration file used by Kafka broker nodes on startup. Default is 3
* `kafka-storage`: a JSON string representing the storage configuration for the Kafka broker nodes. See related section
* `zookeeper-storage`: a JSON string representing the storage configuration for the Zookeeper nodes. See related section
* `kafka-metrics-config`: a JSON string representing the JMX exporter configuration for exposing metrics from Kafka broker nodes.
 Removing this field means having no metrics exposed.
* `zookeeper-metrics-config`: a JSON string representing the JMX exporter configuration for exposing metrics from Zookeeper nodes.
 Removing this field means having no metrics exposed.
* `topic-controller-config`: a JSON string representing the topic controller configuration. See the <<topic_controller_json_config>>
documentation for further details. More info about the topic controller in the related <<Topic Controller>> documentation page.
 
The following is an example of a ConfigMap for a Kafka cluster.

.Example Kafka cluster ConfigMap
[source,yaml,options="nowrap"]
----
apiVersion: v1
kind: ConfigMap
metadata:
  name: my-cluster
  labels:
    strimzi.io/kind: cluster
    strimzi.io/type: kafka
data:
  kafka-nodes: "3"
  kafka-image: "strimzi/kafka:latest"
  kafka-healthcheck-delay: "15"
  kafka-healthcheck-timeout: "5"
  zookeeper-nodes: "1"
  zookeeper-image: "strimzi/zookeeper:latest"
  zookeeper-healthcheck-delay: "15"
  zookeeper-healthcheck-timeout: "5"
  KAFKA_DEFAULT_REPLICATION_FACTOR: "3"
  KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: "3"
  KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: "3"
  kafka-storage: |-
    { "type": "ephemeral" }
  zookeeper-storage: |-
    { "type": "ephemeral" }
  kafka-metrics-config: |-
    {
      "lowercaseOutputName": true,
      "rules": [
          {
            "pattern": "kafka.server<type=(.+), name=(.+)PerSec\\w*><>Count",
            "name": "kafka_server_$1_$2_total"
          },
          {
            "pattern": "kafka.server<type=(.+), name=(.+)PerSec\\w*, topic=(.+)><>Count",
            "name": "kafka_server_$1_$2_total",
            "labels":
            {
              "topic": "$3"
            }
          }
      ]
    }
  zookeeper-metrics-config: |-
    {
      "lowercaseOutputName": true
    }
----

WARNING: If the cluster is deployed using the "ephemeral" storage, the Kafka brokers could have problems to deal with
Zookeeper nodes restarts which could happen on configuration updates in the cluster ConfigMap.

The resources created by the cluster controller into the Kubernetes/OpenShift cluster will be the following :

* `[cluster-name]-zookeeper` StatefulSet which is in charge to create the Zookeeper node pods
* `[cluster-name]-kafka` StatefulSet which is in charge to create the Kafka broker pods
* `[cluster-name]-zookeeper-headless` Service needed to have DNS resolve the Zookeeper pods IP addresses directly
* `[cluster-name]-kafka-headless` Service needed to have DNS resolve the Kafka broker pods IP addresses directly
* `[cluster-name]-zookeeper` Service used by Kafka brokers to connect to Zookeeper nodes as clients
* `[cluster-name]-kafka` Service can be used as bootstrap servers for Kafka clients
* `[cluster-name]-zookeeper-metrics-config` ConfigMap which contains the Zookeeper metrics configuration and mounted as
a volume by the Zookeeper node pods
* `[cluster-name]-kafka-metrics-config` ConfigMap which contains the Kafka metrics configuration and mounted as
a volume by the Kafka broker pods

===== Storage

Both Kafka and Zookeeper save data to files.

{'fraction_non_alphanumeric': 0.06351111873320757, 'fraction_numerical': 0.003429943405933802, 'mean_word_length': 4.614249037227214, 'pattern_counts': {'":': 22, '<': 23, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4030740', 'n_tokens_mistral': 4731, 'n_tokens_neox': 4469, 'n_words': 2151}