Kafka

Kafka is the primary resource users work with. It is a namespaced resource that lets you create and manage Kafka broker instances in Deckhouse Kubernetes Platform and bind them to a specific KafkaClass.
The class defines the allowed sizing options, which configuration parameters can be overridden, and the validation rules that apply.

A Kafka resource has the following main sections in its spec:

• instance — compute and storage requirements for the broker pod (CPU, memory, persistent volume). Must satisfy the sizing policies of the referenced class.
• configuration — broker-level settings you want to tune. Only parameters listed in the class overridableConfiguration can be set here; all other settings are governed by the class defaults.
• tls — server-side TLS encryption settings (cert-manager or custom certificate). See TLS.

KafkaClassName

The name of the KafkaClass to associate with this instance.
Without an existing KafkaClass, deploying the service is impossible.

spec:
  kafkaClassName: default

Instance

Specifies the compute and storage resources for the broker pod.
Must pass validation according to the sizingPolicy of the corresponding class:

spec:
  instance:
    memory:
      size: 1Gi
    cpu:
      cores: 1
      coreFraction: "50%"
    persistentVolumeClaim:
      size: 10Gi
      storageClassName: default

Configuration

Optional broker settings to override the class defaults.
Only parameters listed in the class overridableConfiguration are accepted here. All values are validated against the class validations before being applied.

spec:
  configuration:
    logRetentionHours: 168
    logRetentionBytes: -1
    messageMaxBytes: "1Mi"
    compressionType: producer
    autoCreateTopicsEnable: false

Configuration Fields

logRetentionHours

Type: integer | Kafka param: log.retention.hours | Min: 1 | Example: 168

Number of hours to keep a log file before deleting it.
Mutually exclusive with logRetentionMs — set only one. When both are set, logRetentionMs takes priority.

logRetentionMs

Type: integer | Kafka param: log.retention.ms | Min: -1 | Example: 604800000

Millisecond-precision log retention period. Takes priority over logRetentionHours and is applied dynamically without a broker restart.
Set to -1 to disable time-based retention entirely. Mutually exclusive with logRetentionHours.

logRetentionBytes

Type: int-or-string | Kafka param: log.retention.bytes | Example: 1Gi

Maximum total size of the log per partition before old segments are deleted.
Use -1 to disable size-based retention.

logSegmentBytes

Type: int-or-string | Kafka param: log.segment.bytes | Example: 512Mi

Maximum size of a single log segment file. Once reached, Kafka opens a new segment.
A message must fit within a single segment (messageMaxBytes ≤ logSegmentBytes).

messageMaxBytes

Type: int-or-string | Kafka param: message.max.bytes | Example: 1Mi

Maximum size of a single message the broker will accept from producers.
Must not exceed logSegmentBytes. Must not exceed socketRequestMaxBytes defined in the class.

compressionType

Type: string | Kafka param: compression.type
Allowed values: producer, uncompressed, gzip, snappy, lz4, zstd

Broker-level compression applied to topic messages.
producer (recommended) retains whatever compression the producer used, avoiding re-compression CPU cost.

autoCreateTopicsEnable

Type: boolean | Kafka param: auto.create.topics.enable

Controls whether topics are automatically created when first accessed by a producer or consumer.
Disable in production to keep explicit control over topic lifecycle.

logCleanupPolicy

Type: string | Kafka param: log.cleanup.policy
Allowed values: delete, compact, delete,compact

Cleanup policy applied to log segments when retention limits are exceeded.
delete removes old segments; compact retains only the latest value per key; delete,compact applies both.

Note: Only fields listed in overridableConfiguration of the associated KafkaClass can be set here. Attempting to set a field not in that list will fail validation.

TLS

The spec.tls section enables server-side TLS encryption for traffic between clients and the broker. When TLS is enabled, the broker accepts client connections using the SSL protocol on port 9092.

Prerequisites: the cluster must have cert-manager installed and a ClusterIssuer or Issuer created (when using the CertManager mode).

CertManager Mode

The operator automatically creates a cert-manager Certificate resource that issues and rotates the TLS certificate. The certificate is mounted into the broker pod in PEM format.

Specify one of the following:

• clusterIssuerName — name of a ClusterIssuer (cluster-scoped);
• issuerName — name of an Issuer (in the instance namespace).

spec:
  tls:
    mode: CertManager
    certManager:
      clusterIssuerName: selfsigned

CustomCertificate Mode

Use your own TLS certificate, pre-created as a kubernetes.io/tls Secret in the instance namespace.

spec:
  tls:
    mode: CustomCertificate
    customCertificate:
      serverTLSSecret: my-kafka-tls-secret
      serverCASecret: my-kafka-ca-secret

Connecting Clients over TLS

When TLS is enabled, clients must connect using the SSL protocol. To verify the server certificate, clients need the CA certificate.

Retrieving the CA certificate (CertManager mode):

kubectl get secret d8ms-kfk-<name>-server-tls-secret -n <namespace> \
  -o jsonpath='{.data.ca\.crt}' | base64 -d > ca.crt

Connecting with kcat (kafkacat):

kcat -L -b d8ms-kfk-<name>.<namespace>.svc:9092 \
  -X security.protocol=SSL \
  -X ssl.ca.location=ca.crt

Connecting from a Java application:

bootstrap.servers=d8ms-kfk-<name>.<namespace>.svc:9092
security.protocol=SSL
ssl.truststore.type=PEM
ssl.truststore.certificates=<contents of ca.crt>

Note: when using self-signed certificates, make sure the CA certificate is available to the client. For production, use a trusted ClusterIssuer (e.g. one backed by your organization’s internal PKI).

Supported Kafka Versions

The only supported Kafka version is 4.2.0.

Our images for running Kafka containers are based on distroless architecture.

Status

The status of the Managed Kafka service is reflected in the Kafka resource.
The Conditions structure clearly shows the current status of the service.

Significant types:

• LastValidConfigurationApplied — An aggregating type that shows whether the last valid configuration has been successfully applied at least once.
• ConfigurationValid — shows whether the configuration has passed all validations of the associated KafkaClass.
• ScaledToLastValidConfiguration — shows whether the number of running replicas matches the specified configuration.
• Available — shows whether the broker is running and accepting connections.

conditions:
    - lastTransitionTime: '2025-09-22T23:20:36Z'
      observedGeneration: 2
      status: 'True'
      type: Available
    - lastTransitionTime: '2025-09-22T14:38:04Z'
      observedGeneration: 2
      status: 'True'
      type: ConfigurationValid
    - lastTransitionTime: '2025-09-22T14:38:47Z'
      observedGeneration: 2
      status: 'True'
      type: LastValidConfigurationApplied
    - lastTransitionTime: '2025-09-22T23:20:36Z'
      observedGeneration: 2
      status: 'True'
      type: ScaledToLastValidConfiguration

A False status indicates a problem at one stage or another, or incomplete state synchronization.
For such a state, a reason and message with a description will be specified.

---
    - lastTransitionTime: '2025-09-23T14:53:33Z'
      message: Syncing
      observedGeneration: 1
      reason: Syncing
      status: 'False'
      type: LastValidConfigurationApplied
    - lastTransitionTime: '2025-09-23T14:54:58Z'
      message: Not all the instances are running still waiting for 1 to become ready
      observedGeneration: 1
      reason: ScalingInProgress
      status: 'False'
      type: ScaledToLastValidConfiguration
---

Usage Examples

Basic Usage

Standard Kafka broker with persistent storage and class default configuration.

1. Create a namespace named kafka.

2. Create a Kafka resource:

   kubectl apply -f managed-services_v1alpha1_kafka.yaml -n kafka

   apiVersion: managed-services.deckhouse.io/v1alpha1
   kind: Kafka
   metadata:
     name: kafka-sample
   spec:
     kafkaClassName: default
     instance:
       memory:
         size: "1Gi"
       cpu:
         cores: 2
         coreFraction: "50%"
       persistentVolumeClaim:
         size: "10Gi"

3. Wait until the broker is ready and all conditions are True:

   kubectl get kafka kafka-sample -n kafka -o wide -w

4. Connect to the broker using the d8ms-kfk-kafka-sample service on port 9092:

   --bootstrap-server d8ms-kfk-kafka-sample:9092

Short-Term Retention

Kafka broker with reduced log retention — suitable for development or testing environments.

1. Create a namespace named kafka.

2. Create a Kafka resource:

   kubectl apply -f managed-services_v1alpha1_kafka.yaml -n kafka

   apiVersion: managed-services.deckhouse.io/v1alpha1
   kind: Kafka
   metadata:
     name: kafka-dev
   spec:
     kafkaClassName: default
     configuration:
       logRetentionHours: 24
       autoCreateTopicsEnable: true
     instance:
       memory:
         size: "1Gi"
       cpu:
         cores: 2
         coreFraction: "25%"
       persistentVolumeClaim:
         size: "5Gi"

3. Wait until the broker is ready and all conditions are True:

   kubectl get kafka kafka-dev -n kafka -o wide -w

4. Connect to the broker using the d8ms-kfk-kafka-dev service on port 9092:

   --bootstrap-server d8ms-kfk-kafka-dev:9092

Broker with TLS Encryption

Kafka broker with server-side TLS enabled via cert-manager.

1. Make sure cert-manager is installed in the cluster and a ClusterIssuer exists:

   apiVersion: cert-manager.io/v1
   kind: ClusterIssuer
   metadata:
     name: selfsigned
   spec:
     selfSigned: {}

2. Create a namespace and a Kafka resource with TLS:

   apiVersion: managed-services.deckhouse.io/v1alpha1
   kind: Kafka
   metadata:
     name: kafka-tls
     namespace: kafka
   spec:
     kafkaClassName: default
     tls:
       mode: CertManager
       certManager:
         clusterIssuerName: selfsigned
     configuration:
       logRetentionHours: 168
     instance:
       memory:
         size: "1Gi"
       cpu:
         cores: 2
         coreFraction: "50%"
       persistentVolumeClaim:
         size: "10Gi"

   kubectl apply -f kafka-tls.yaml

3. Wait until the broker is ready:

   kubectl get kafka kafka-tls -n kafka -o wide -w

4. Retrieve the CA certificate:

   kubectl get secret d8ms-kfk-kafka-tls-server-tls-secret -n kafka \
     -o jsonpath='{.data.ca\.crt}' | base64 -d > ca.crt

5. Connect to the broker over SSL:

   kcat -L -b d8ms-kfk-kafka-tls.kafka.svc:9092 \
     -X security.protocol=SSL \
     -X ssl.ca.location=ca.crt

Custom Compression and Message Size

Kafka broker configured with gzip compression and an increased maximum message size.

1. Create a namespace named kafka.

2. Create a Kafka resource:

   kubectl apply -f managed-services_v1alpha1_kafka.yaml -n kafka

   apiVersion: managed-services.deckhouse.io/v1alpha1
   kind: Kafka
   metadata:
     name: kafka-compressed
   spec:
     kafkaClassName: default
     configuration:
       compressionType: gzip
       messageMaxBytes: "10Mi"
       autoCreateTopicsEnable: false
     instance:
       memory:
         size: "2Gi"
       cpu:
         cores: 2
         coreFraction: "50%"
       persistentVolumeClaim:
         size: "20Gi"

3. Wait until the broker is ready and all conditions are True:

   kubectl get kafka kafka-compressed -n kafka -o wide -w

4. Connect to the broker using the d8ms-kfk-kafka-compressed service on port 9092:

   --bootstrap-server d8ms-kfk-kafka-compressed:9092