Configure Storage

Storage overview

Confluent for Kubernetes (CFK) requires Kubernetes storage classes to provision persistent storage volumes for most Confluent Platform components. A storage class defines the type and configuration of storage that Kubernetes dynamically provisions for a persistent volume.

Beyond that storage class requirement, CFK has no specific requirements or recommendations for storage solutions or configuration details. 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 storage requirements

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

• AWS EBS

  On Kubernetes 1.23 or later, the EBS Container Storage Interface (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.

• 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.

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, see Configure Kafka Tiered Storage.

ZooKeeper storage requirements

ZooKeeper uses the same block storage solutions as Kafka (AWS EBS, Azure Disk, GCE Disk, Ceph RBD, or Portworx).

Important

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

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 Control Center stores operational state and metrics in Kafka.

Persistent Volumes

When configuring Confluent components to use persistent storage volumes, CFK supports the following options:

• 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.

Storage class configuration is an important step in the Confluent for Kubernetes configuration process.

Storage class change considerations

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 the following, contact your account team:

• Migrating from one storage class to another.

• Changing the storage class, for example, enabling encryption on the persistent volumes.

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 an example, see Kubernetes Storage Classes.

2. In your Confluent component custom resource (CR), specify the name of the StorageClass to use:

   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.

Use default Kubernetes StorageClass for dynamic provisioning

If you don’t provide spec.storageClass in the CR, CFK uses the Kubernetes default storage class (supported by default since Kubernetes 1.11).

Important

Using the default StorageClass in production environments is not recommended.

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

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:

   apiVersion: storage.k8s.io/v1
   kind: StorageClass
   metadata:
     name: my-storage-class
   provisioner: kubernetes.io/no-provisioner
   volumeBindingMode: WaitForFirstConsumer
   

2. Create PersistentVolumes 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:

   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 PersistentVolumeClaim that CFK eventually creates and which Kubernetes eventually binds to this PersistentVolume.

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

     Warning

     With persistentVolumeReclaimPolicy: Delete, CFK deletes your data on the volume when you delete the component 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:

     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:

   spec:
     storageClass:
       name: my-storage-class
   

4. After deploying the new CR, validate that the PersistentVolumes are bound:

   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:

   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

Set access mode for persistent volume claims

In Kubernetes, a persistent volume claim (PVC) access mode defines how pods interact with the underlying storage provided by a persistent volume. It specifies whether multiple pods can access the volume simultaneously, and whether they can read from and write to it.

By default, CFK creates a PVC with the ReadWriteOnce access mode.

You can override the access mode in a PVC with the platform.confluent.io/pvc-access-mode annotation while creating the CFK CRs.

CFK supports the following access modes:

• ReadWriteOnce: The volume can be mounted in read/write mode by a single node. Multiple CR pods scheduled on that node can access the volume.

• ReadWriteMany: The volume can be mounted in read/write mode by many nodes.

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

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

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

Tiered Storage overview

Tiered Storage lets Kafka brokers offload older log segments to object storage, such as S3 or GCS, while keeping them available to consumers.

You can use Confluent for Kubernetes to enable Tiered Storage 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

Prerequisites for Tiered Storage

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

• The type of object 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

  This is the recommended option.

• Kubernetes Secret

Use a service account to give Kafka brokers access to the storage bucket

Map cloud Identity and Access Management (IAM) permissions to the Kubernetes ServiceAccount associated with your Kafka broker pods.

AWS provides the ability to natively associate AWS IAM permissions with ServiceAccounts in EKS.

Similarly, GCP provides the ability to map IAM permissions with ServiceAccounts in GKE.

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.

For more on associating AWS IAM roles for service accounts on EKS, see IAM roles for service accounts.

For more on associating GCP IAM roles for service accounts on GKE, see Workload Identity.

For more information on configuring which Kubernetes ServiceAccount to associate with Confluent Platform components managed by CFK, see Provide custom service account.

Use a Kubernetes Secret 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 through CFK.

When your storage credentials change, you need to restart the Kafka cluster.

In addition to the storage type and bucket name required to enable Tiered Storage, you can configure other Tiered Storage settings using configOverrides in the kafka section. For the available settings, see Tiered Storage.

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 configured deletion interval (confluent.tier.topic.delete.check.interval.ms, three hours by default) or manually delete the objects in the Tiered Storage bucket. For more information, see Time Interval for Topic Deletes.

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.

Configure Tiered Storage for AWS S3

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

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=           ----- [8]
      - confluent.tier.s3.sse.customer.encryption.key= -- [9]

  mountedSecrets:                                  ----- [10]
    - 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 ServiceAccounts, this property is not necessary.

  For what to add in the Secrets file, see 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 three hours.

• [8] 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 Key Management Service (KMS) encryption when compliance or regulatory requirements mandate using a specific KMS key for encryption at rest.

• [9] Optional. Set confluent.tier.s3.sse.customer.encryption.key to the Amazon Resource Name (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.

• [10] Optional. Only required if using Secrets to provide credentials for Tiered Storage.

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:

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:

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:

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 ServiceAccounts, this property is not necessary.

  For what to add in the Secrets file, see 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 three hours.

• [8] Optional. Only required if using Secrets to provide credentials for Tiered Storage.

For example:

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