<a id="co-storage"></a>

# Configure Storage for Confluent Platform Using Confluent for Kubernetes

## Overview

Confluent for Kubernetes (CFK) requires the use of Kubernetes Storage Classes to provision
persistent storage volumes for most of the Confluent Platform components.

Apart from the above requirements of persistent volume Storage Class, CFK does
not have specific requirements or recommendations for storage solutions or
storage configuration details. You should test and tune the storage solutions
you select before configuring them for CFK.

Connect and Schema Registry do not use persistent storage volumes, and thus do not need
a Storage Class to be specified.

### Kafka

For Kafka persistent volumes, CFK supports block storage. Example block storage
solutions you can use are:

* AWS EBS

  On Kubernetes 1.23 or higher, EBS CSI driver is required to make CFK and Kafka
  work. Enable and set up the EBS-CSI add-on as described in [Managing the
  Amazon EBS CSI driver as an Amazon EKS add-on](https://docs.aws.amazon.com/eks/latest/userguide/managing-ebs-csi.html).
* Azure Disk
* GCE Disk
* Ceph RBD
* Portworx

You need 1 persistent volume on each Kafka host.

Object storage or network file storage solutions are not supported for Kafka.

Network-attached storage (NAS) is not recommended as noted in [Running Kafka in
Production with Confluent Platform](https://docs.confluent.io/platform/current/kafka/deployment.html#disks).

Storage Area Networks (SANs) that provide block storage (as opposed to file storage) do not present the same issues as NAS.

In addition to persistent storage, you can configure Kafka with Tiered Storage.
For details, [Configure Kafka Tiered Storage](#co-tiered-storage-configure).

### ZooKeeper

#### IMPORTANT
Starting with Confluent Platform version 8.0, ZooKeeper is no longer part of Confluent Platform.

ZooKeeper uses the same persistent storage volume solution that Kafka uses.

You need 2 persistent volumes on each ZooKeeper host.

### Other Confluent components

Other than Connect and Schema Registry, each Confluent component requires 1 persistent
storage volume to be configured as the component uses it for maintaining working
state during the lifetime of the service.

Connect and Schema Registry do not use persistent storage volumes.

Additionally, Confluent components utilize Kafka for their durable shared storage
needs. For example, Schema Registry stores schemas in Kafka, and Confluent Control Center stores
operational state  and metrics in Kafka.

## Persistent Volumes

Storage class configuration is one of the most critical steps in the Confluent for Kubernetes
(CFK) configuration process.

When configuring Confluent components to use persistent storage volumes, the
following options are supported:

* Dynamic Provisioning: Use a pre-defined Kubernetes Storage Class
* Dynamic Provisioning: Use the Kubernetes default Storage Class
* Custom Provisioning: Use pre-provisioned persistent storage volumes

By default, CFK manages storage using dynamic storage provisioning that
Kubernetes provides.

### Consideration

CFK does not support an automated change to storage classes on an existing
deployment. To make changes to a storage class on an existing deployment, such
as listed below, contact your account team:

* Migrating from one storage class to another.
* Changing the storage class, for example, enabling encryption on the persistent
  volumes.

<a id="co-predefined-storage"></a>

### Use pre-defined Kubernetes StorageClass for dynamic provisioning

You can provide a storage class to use for the entire Confluent Platform, or you can specify
different storage classes for different components such as ZooKeeper (Confluent Platform 7.9 or
earlier only), Kafka, ksqlDB, and Control Center.

To use a pre-defined storage class:

1. Create or use a pre-defined StorageClass you want to use in your Kubernetes
   cluster.

   The following settings are the best practice recommendations for your storage
   class:
   * `volumeBindingMode: WaitForFirstConsumer`
   * `reclaimPolicy: Retain`

     Especially for production deployments, this setting is required.
   * `allowVolumeExpansion: true`

   You need to have sufficient permissions to create and modify StorageClasses
   in your Kubernetes cluster if you intend to create a new StorageClass to use
   rather than using a pre-existing one.

   For more information and example, see [Kubernetes Storage Classes](https://kubernetes.io/docs/concepts/storage/storage-classes).
2. In your Confluent component custom resource (CR), specify the name of the
   StorageClass to use:
   ```yaml
   spec:
     storageClass:
       name: my-storage-class
   ```

For a sample scenario for creating a storage class for a production environment,
see [Create a Production Storage Class on GKE](https://github.com/confluentinc/confluent-kubernetes-examples/tree/master/storage).

<a id="co-kubernetes-storage"></a>

### Use default Kubernetes StorageClass for dynamic provisioning

The support for default StorageClasses is enabled by default in versions 1.11
and higher of Kubernetes. If you do not provide the `spec.storageClass` in
the CR, CFK will use the default storage class.

#### IMPORTANT
We do not recommend using the default StorageClass in production
environments.

Use the following command to get the name of the current default storage class:

```bash
kubectl get sc
```

To use the Kubernetes default storage class, make sure the following properties
are set on the default StorageClasses:

* `volumeBindingMode: WaitForFirstConsumer`
* `reclaimPolicy: Retain`

  Especially for production deployments, this setting is required.
* `allowVolumeExpansion: true`

### Use statically provisioned persistent volumes

By default, CFK automates disk management by leveraging Kubernetes dynamic
storage provisioning. If your Kubernetes cluster does not support dynamic
provisioning, you can follow the instructions in this section to use
statically-provisioned disks for your Confluent Platform deployments.

Connect and Schema Registry do not use persistent storage volumes, so you do not need to
follow the steps in this section.

To use statically-provisioned persistent volumes for a Confluent Platform component:

1. Create a StorageClass in Kubernetes for local provisioning. For example:
   ```yaml
   apiVersion: storage.k8s.io/v1
   kind: StorageClass
   metadata:
     name: my-storage-class
   provisioner: kubernetes.io/no-provisioner
   volumeBindingMode: WaitForFirstConsumer
   ```
2. Create [PersistentVolumes](https://kubernetes.io/docs/concepts/storage/persistent-volumes/#persistent-volumes)
   with the desired host path and the hostname label for each of the desired
   worker nodes.

   You need the following number of persistent volumes for Confluent Platform components:
   * 2 persistent volumes on each ZooKeeper host (Confluent Platform 7.9 or earlier only)
   * 1 persistent volume on each Kafka, ksqlDB, Control Center host

   For example:
   ```yaml
   apiVersion: v1
   kind: PersistentVolume
   metadata:
     name: pv-1                                                --- [1]
   spec:
     capacity:
       storage: 10Gi                                           --- [2]
     volumeMode: Filesystem
     accessModes:
     - ReadWriteOnce
     persistentVolumeReclaimPolicy: Retain                     --- [3]
     storageClassName: my-storage-class                        --- [4]
     local:
        path: /mnt/data/broker-1-data                          --- [5]
     nodeAffinity:
       required:
         nodeSelectorTerms:
         - matchExpressions:
           - key: kubernetes.io/hostname
             operator: In
             values:
             - gke-myhost-cluster-default-pool-5cc13882-k0gb  --- [6]
   ```

   * [1] Choose a name for the PersistentVolume.
   * [2] Choose a storage size that is greater than or equal to the storage
     you’re requesting for each Kafka broker instance. This corresponds to the
     `spec.dataVolumeCapacity` property of the component CR.
   * [3] Choose `Retain` if you want the data to be retained after you delete
     the PersitentVolumeClaim that CFK will eventually create and which
     Kubernetes will eventually bind to this PersistentVolume.

     Choose `Delete` if you want this data to be garbage-collected when the
     PersistentVolumeClaim is deleted.

     #### WARNING
     With `persistentVolumeReclaimPolicy: Delete`, your data on the volume
     will be deleted when you delete the CFK component custom resource (CR),
     for example, when you delete the Kafka CR with the `kubectl delete kafka
     <kafka_cluster-name>` .
   * [4] The `storageClassName` must match the one created in Step 1.
   * [5] This is the directory path you want to use on the worker node for the
     broker as its persistent data volume. The path must exist on the worker
     node.
   * [6] This is the value of the `kubernetes.io/hostname` label of the worker
     node you want to host this broker instance. To find this hostname, run the
     following command:
     ```bash
     kubectl get nodes \
       -o 'custom-columns=NAME:metadata.name,HOSTNAME:metadata.labels.kubernetes\.io/hostname'

     NAME                                           HOSTNAME
     gke-myhost-cluster-default-pool-5cc13882-k0gb   gke-myhost-cluster-default-pool-5cc13882-k0gb
     gke-myhost-cluster-default-pool-5cc13882-n8vr   gke-myhost-cluster-default-pool-5cc13882-n8vr
     gke-myhost-cluster-default-pool-5cc13882-tbbj   gke-myhost-cluster-default-pool-5cc13882-tbbj
     ```
3. Add the storageClass to the component CR, for example:
   ```yaml
   spec:
     storageClass:
       name: my-storage-class
   ```
4. After deploying the new CR, validate that the PersistentVolumes are bound:
   ```bash
   kubectl get pv

   NAME  CAPACITY ACCESS MODES RECLAIM POLICY STATUS CLAIM                   STORAGECLASS
   pv-1  10Gi     RWO          Retain         Bound  operator/data0-kafka-0  my-storage-class
   pv-2  10Gi     RWO          Retain         Bound  operator/data0-kafka-2  my-storage-class
   pv-3  10Gi     RWO          Retain         Bound  operator/data0-kafka-1  my-storage-class
   ```
5. Validate that the Confluent Platform pods are healthy. For example:
   ```bash
   kubectl get pods -l app=kafka

   NAME      READY   STATUS    RESTARTS   AGE
   kafka-0   1/1     Running   0          40m
   kafka-1   1/1     Running   0          40m
   kafka-2   1/1     Running   0          40m
   ```

## Persistent Volume Claims

<a id="co-set-pvc-accessmode"></a>

### Set access mode for Persistent Volume Claims

In Kubernetes, a [Persistent Volume Claim (PVC) access mode](https://kubernetes.io/docs/concepts/storage/persistent-volumes/#access-modes)
defines how pods can interact with the underlying storage provided by a Persistent Volume, specifying
whether multiple pods can access the volume simultaneously and if they can read
and write to it.

By default, CFK creates Persistent Volume Claim (PVC) with the
`ReadWriteOnce` access mode.

You can override the access mode in a PVC with the
`platform.confluent.io/pvcAccessMode` annotation while creating the CFK
custom resources (CRs).

CFK supports the following access modes:

- `ReadWriteOnce`: The volume can be mounted in read/write mode to exactly one CR pod.
- `ReadWriteMany`: The volume can be mounted in read/write mode to many CR pods.

For example, to set the Kafka PVC access mode to `ReadWriteMany`:

```bash
kubectl annotate kafka kafka platform.confluent.io/pvc-access-mode=ReadWriteMany
```

Once a PVC is created, the access mode cannot be changed.

<!-- Uncomment this when CFK-2170 is fixed.

.. _co-set-customvolumemode:

Set custom VolumeMode for Persistent Volume Claims
--------------------------------------------------

|co-long| (|co|) creates Persistent Volume Claim (PVC) with the filesystem
volume mode.

You can override the volume mode in a PVC with a Kubernetes annotation while
creating the CFK resources.

For example, to set the |ak| PVC volume mode to ``Block``:

.. codewithvars:: bash

   kubectl annotate kafka kafka platform.confluent.io/pvc-volume-mode=Block

``Block`` and ``Filesystem`` are the valid values. Override will be ignored for
invalid values.

Once a PVC is created, the volume mode cannot be updated. -->

<a id="co-tiered-storage-configure"></a>

## Tiered Storage

You can use Confluent for Kubernetes (CFK) to enable [Tiered
Storage](/platform/current/kafka/tiered-storage.html) for Kafka. Confluent supports various object
Storage solutions, such as:

* AWS S3
* GCP GCS
* Azure Blob Storage
* Pure Storage FlashBlade
* Nutanix Objects
* NetApp Object Storage
* Dell EMC ECS
* MinIO
* Hitachi Content Platform Object Storage

### Enable Tiered Storage

When you enable Tiered Storage, you need to configure Kafka with the following:

* The type of blob storage to use.
* The name of the storage bucket to use.

  You must have created this bucket in advance. CFK does not create this
  bucket on your behalf.

You also need to ensure that the Kafka brokers have appropriate access to the
storage bucket. You can use one of the following options:

* [Service Account](co-configure-misc.md#co-service-account)

  This is the recommended option.
* Kubernetes Secret

Use Service Account to give Kafka brokers access to the storage bucket
: Map cloud IAM permissions to the Kubernetes ServiceAccount associated with
  your Kafka broker pods.
  <br/>
  AWS provides the ability to natively associate AWS IAM permissions with
  ServiceAccounts in EKS.
  <br/>
  Similarly, GCP provides the ability to map IAM permissions with
  ServiceAccounts in GKE.
  <br/>
  You can map the appropriate bucket permissions to the default ServiceAccount
  in the Kubernetes namespace where you plan to deploy Kafka, or you can map them
  to a separate ServiceAccount and use CFK to ensure the Kafka broker pods are
  associated with that ServiceAccount. The primary benefit of this approach is
  that you do not need to actually manage sensitive credentials for bucket
  access when deploying Confluent Platform via CFK.
  <br/>
  For more on associating AWS IAM roles for service accounts on EKS, see [IAM
  roles for service accounts](https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts.html).
  <br/>
  For more on associating GCP IAM roles for service accounts on GKE, see
  [Workload Identity](https://cloud.google.com/kubernetes-engine/docs/how-to/workload-identity).
  <br/>
  For more information on configuring which Kubernetes Service Account to
  associate with Confluent Platform components managed by CFK, see [Provide custom service account](co-configure-misc.md#co-service-account).

Use the Kubernetes Secret object to give Kafka brokers access to the storage bucket
: Put your AWS or GCP credentials in a Secret object and configure Kafka to use
  the credentials in that object, when deploying Kafka via the CFK.
  <br/>
  When your storage credentials change, you need to restart the Kafka cluster.

In addition to the above required settings, you can configure other Tiered Storage
settings using `configOverrides` in the `kafka` section. For the available
settings, see [Tiered Storage](/platform/current/kafka/tiered-storage.html).

When a Kafka cluster is deleted, CFK does not perform a garbage collection of
the Tiered Storage bucket contents. You can either wait for the set interval or
manually delete the objects in the Tiered Storage bucket. For more information, see
[Time Interval for Topic
Deletes](/platform/current/kafka/tiered-storage.html#tiered-storage-delete-check-interval).

### Configure Tiered Storage

You configure Tiered Storage in the Kafka CR in `spec.configOverrides.server`.

This section presents the steps to enable Tiered Storage for AWS S3 and GCS. For the
configuration properties you need to set for other storage providers via
`spec.configOverrides.server`, see [Enabling Tiered Storage on a Broker](https://docs.confluent.io/platform/current/kafka/tiered-storage.html#enabling-tiered-storage-on-a-broker).

#### Configure Tiered Storage for AWS S3

To enable and configure Tiered Storage with AWS S3, set the following config overrides
in the Kafka CR:

```yaml
kind: Kafka
spec:
  configOverrides:
    server:
      - confluent.tier.feature=true                ----- [1]
      - confluent.tier.enable=true                 ----- [2]
      - confluent.tier.backend=S3                  ----- [3]
      - confluent.tier.s3.bucket=                  ----- [4]
      - confluent.tier.s3.region=                  ----- [5]
      - confluent.tier.s3.cred.file.path=          ----- [6]
      - confluent.tier.topic.delete.check.interval.ms -- [7]
      - confluent.tier.s3.sse.algorithm=           ----- [9]
      - confluent.tier.s3.sse.customer.encryption.key= -- [10]

  mountedSecrets:                                  ----- [8]
    - secretRef:
```

* [1] Set `confluent.tier.feature=true` to enable Tiered Storage.
* [2] Set `confluent.tier.enable` to the default value for created topics.
  Setting this to `true` causes all non-compacted topics to be tiered.
* [3] Set `confluent.tier.backend` to `S3`.
* [4] Set `confluent.tier.s3.bucket` to the S3 bucket you want to use.
* [5] Set `confluent.tier.s3.region` to the region.
* [6] Optional. Specify `confluent.tier.s3.cred.file.path` if using Secrets to provide credentials for Tiered Storage. If using Service Accounts, this property is not necessary.

  To see what to add in the Secrets file, refer to [Tiered Storage](/platform/current/kafka/tiered-storage.html).
* [7] Optional. Set `confluent.tier.topic.delete.check.interval.ms` to a time
  interval for which the deletion of log segment files takes place after a topic
  or a cluster is deleted. The default value for this time interval is three hours.
* [8] Optional. Only required if using Secrets to provide credentials for
  Tiered Storage.
* [9] Optional. Set `confluent.tier.s3.sse.algorithm` to configure the
  server-side encryption algorithm for objects in S3. The default value is
  `AES256`, which uses SSE-S3 encryption. Set to `aws:kms` to use SSE-KMS
  with AWS KMS encryption when compliance or regulatory requirements mandate
  using a specific KMS key for encryption at rest.
* [10] Optional. Set `confluent.tier.s3.sse.customer.encryption.key` to the
  ARN of the AWS KMS key to use for encryption. For example,
  `arn:aws:kms:us-west-2:123456789012:key/12345678-1234-1234-1234-123456789012`.
  This parameter is required when `confluent.tier.s3.sse.algorithm` is set to
  `aws:kms`.

#### IMPORTANT
When using SSE-KMS encryption (`confluent.tier.s3.sse.algorithm=aws:kms`), ensure the
AWS IAM role or service account associated with your Kafka brokers has the following KMS permissions to access the specified key:

* `kms:Decrypt`
* `kms:Encrypt`
* `kms:GenerateDataKey`

Without these permissions, the brokers cannot read or write Tiered Storage objects.

If `confluent.tier.s3.sse.algorithm` is not specified, S3 encrypts objects using SSE-S3 (AES256) regardless of the default encryption settings configured on the S3 bucket.

For example:

```yaml
spec:
  configOverrides:
    server:
      - confluent.tier.feature=true
      - confluent.tier.enable=true
      - confluent.tier.backend=S3
      - confluent.tier.s3.bucket=my-bucket
      - confluent.tier.s3.region=us-west-2
      - confluent.tier.s3.cred.file.path=/mnt/secrets/my-secret-aws/aws/creds

  mountedSecrets:
    - secretRef: my-secret-aws
```

To configure Tiered Storage with SSE-KMS encryption for compliance requirements:

```yaml
spec:
  configOverrides:
    server:
      - confluent.tier.feature=true
      - confluent.tier.enable=true
      - confluent.tier.backend=S3
      - confluent.tier.s3.bucket=my-bucket
      - confluent.tier.s3.region=us-west-2
      - confluent.tier.s3.sse.algorithm=aws:kms
      - confluent.tier.s3.sse.customer.encryption.key=arn:aws:kms:us-west-2:123456789012:key/12345678-1234-1234-1234-123456789012
```

#### Configure Tiered Storage for GCS

To enable and configure Tiered Storage with GCS, set the following config overrides
in the Kafka CR:

```yaml
spec:
  configOverrides:
    server:
      - confluent.tier.feature=true                ----- [1]
      - confluent.tier.enable=true                 ----- [2]
      - confluent.tier.backend=GCS                 ----- [3]
      - confluent.tier.gcs.bucket=                 ----- [4]
      - confluent.tier.gcs.region=                 ----- [5]
      - confluent.tier.gcs.cred.file.path=         ----- [6]
      - confluent.tier.topic.delete.check.interval.ms -- [7]

  mountedSecrets:                                  ----- [8]
    - secretRef:
```

* [1] Set `confluent.tier.feature=true` to enable Tiered Storage.
* [2] Set `confluent.tier.enable` to the default value for created topics.
  Setting this to `true` causes all non-compacted topics to be tiered.
* [3] Set `confluent.tier.backend` to `GCS`.
* [4] Set `confluent.tier.gcs.bucket` to the GCS bucket you want to use.
* [5] Set `confluent.tier.gcs.region` to the GCS region.
* [6] Optional. Specify `confluent.tier.gcs.cred.file.path` if using
  Secrets for Tiered Storage. If using Service Accounts, this
  property is not necessary.

  To see what to add in the Secrets file, refer to [Tiered Storage](https://docs.confluent.io/platform/current/kafka/tiered-storage.html#tiered-storage).
* [7] Optional. Set `confluent.tier.topic.delete.check.interval.ms` to a time
  interval for which the deletion of log segment files takes place after a topic
  or a cluster is deleted. The default value for this time interval is 3 hours.
* [8] Optional. Only required if using Secrets to provide credentials for
  Tiered Storage.

For example:

```yaml
spec:
  configOverrides:
    server:
      - confluent.tier.feature=true
      - confluent.tier.enable=true
      - confluent.tier.backend=GCS
      - confluent.tier.gcs.bucket=my-bucket
      - confluent.tier.gcs.region=us-central1
      - confluent.tier.gcs.cred.file.path=/mnt/secrets/my-secret-gcs/credentials

  mountedSecrets:
  - secretRef: my-secret-gcs
```