Red Hat Summit 红帽全球峰会支持控制台（Console）开发人员开始试用联系人

9.2. Configuring Kafka

Update the spec properties of the Kafka custom resource to configure your Kafka deployment.

As well as configuring Kafka, you can add configuration for ZooKeeper and the Streams for Apache Kafka Operators. Common configuration properties, such as logging and healthchecks, are configured independently for each component.

Configuration options that are particularly important include the following:

  • Resource requests (CPU / Memory)
  • JVM options for maximum and minimum memory allocation
  • Listeners for connecting clients to Kafka brokers (and authentication of clients)
  • Authentication
  • Storage
  • Rack awareness
  • Metrics
  • Cruise Control for cluster rebalancing
  • Metadata version for KRaft-based Kafka clusters
  • Inter-broker protocol version for ZooKeeper-based Kafka clusters

The .spec.kafka.metadataVersion property or the inter.broker.protocol.version property in config must be a version supported by the specified Kafka version (spec.kafka.version). The property represents the Kafka metadata or inter-broker protocol version used in a Kafka cluster. If either of these properties is not set in the configuration, the Cluster Operator updates the version to the default for the Kafka version used.

注意

The oldest supported metadata version is 3.3. Using a metadata version that is older than the Kafka version might cause some features to be disabled.

For a deeper understanding of the Kafka cluster configuration options, refer to the Streams for Apache Kafka Custom Resource API Reference.

Managing TLS certificates

When deploying Kafka, the Cluster Operator automatically sets up and renews TLS certificates to enable encryption and authentication within your cluster. If required, you can manually renew the cluster and clients CA certificates before their renewal period starts. You can also replace the keys used by the cluster and clients CA certificates. For more information, see Renewing CA certificates manually and Replacing private keys.

Example Kafka custom resource configuration

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    replicas: 3
1 

    version: 3.7.0
2 

    logging:
3 

      type: inline
      loggers:
        kafka.root.logger.level: INFO
    resources:
4 

      requests:
        memory: 64Gi
        cpu: "8"
      limits:
        memory: 64Gi
        cpu: "12"
    readinessProbe:
5 

      initialDelaySeconds: 15
      timeoutSeconds: 5
    livenessProbe:
      initialDelaySeconds: 15
      timeoutSeconds: 5
    jvmOptions:
6 

      -Xms: 8192m
      -Xmx: 8192m
    image: my-org/my-image:latest
7 

    listeners:
8 

      - name: plain
9 

        port: 9092
10 

        type: internal
11 

        tls: false
12 

        configuration:
          useServiceDnsDomain: true
13 

      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
14 

          type: tls
      - name: external1
15 

        port: 9094
        type: route
        tls: true
        configuration:
          brokerCertChainAndKey:
16 

            secretName: my-secret
            certificate: my-certificate.crt
            key: my-key.key
    authorization:
17 

      type: simple
    config:
18 

      auto.create.topics.enable: "false"
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      default.replication.factor: 3
      min.insync.replicas: 2
      inter.broker.protocol.version: "3.7"
    storage:
19 

      type: persistent-claim
20 

      size: 10000Gi
    rack:
21 

      topologyKey: topology.kubernetes.io/zone
    metricsConfig:
22 

      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
23 

          name: my-config-map
          key: my-key
    # ...
  zookeeper:
24 

    replicas: 3
25 

    logging:
26 

      type: inline
      loggers:
        zookeeper.root.logger: INFO
    resources:
      requests:
        memory: 8Gi
        cpu: "2"
      limits:
        memory: 8Gi
        cpu: "2"
    jvmOptions:
      -Xms: 4096m
      -Xmx: 4096m
    storage:
      type: persistent-claim
      size: 1000Gi
    metricsConfig:
      # ...
  entityOperator:
27 

    tlsSidecar:
28 

      resources:
        requests:
          cpu: 200m
          memory: 64Mi
        limits:
          cpu: 500m
          memory: 128Mi
    topicOperator:
      watchedNamespace: my-topic-namespace
      reconciliationIntervalSeconds: 60
      logging:
29 

        type: inline
        loggers:
          rootLogger.level: INFO
      resources:
        requests:
          memory: 512Mi
          cpu: "1"
        limits:
          memory: 512Mi
          cpu: "1"
    userOperator:
      watchedNamespace: my-topic-namespace
      reconciliationIntervalSeconds: 60
      logging:
30 

        type: inline
        loggers:
          rootLogger.level: INFO
      resources:
        requests:
          memory: 512Mi
          cpu: "1"
        limits:
          memory: 512Mi
          cpu: "1"
  kafkaExporter:
31 

    # ...
  cruiseControl:
32 

    # ...

1
The number of replica nodes.
2
Kafka version, which can be changed to a supported version by following the upgrade procedure.
3
Kafka loggers and log levels added directly (inline) or indirectly (external) through a ConfigMap. A custom Log4j configuration must be placed under the log4j.properties key in the ConfigMap. For the Kafka kafka.root.logger.level logger, you can set the log level to INFO, ERROR, WARN, TRACE, DEBUG, FATAL or OFF.
4
Requests for reservation of supported resources, currently cpu and memory, and limits to specify the maximum resources that can be consumed.
5
Healthchecks to know when to restart a container (liveness) and when a container can accept traffic (readiness).
6
JVM configuration options to optimize performance for the Virtual Machine (VM) running Kafka.
7
ADVANCED OPTION: Container image configuration, which is recommended only in special situations.
8
Listeners configure how clients connect to the Kafka cluster via bootstrap addresses. Listeners are configured as internal or external listeners for connection from inside or outside the OpenShift cluster.
9
Name to identify the listener. Must be unique within the Kafka cluster.
10
Port number used by the listener inside Kafka. The port number has to be unique within a given Kafka cluster. Allowed port numbers are 9092 and higher with the exception of ports 9404 and 9999, which are already used for Prometheus and JMX. Depending on the listener type, the port number might not be the same as the port number that connects Kafka clients.
11
Listener type specified as internal or cluster-ip (to expose Kafka using per-broker ClusterIP services), or for external listeners, as route (OpenShift only), loadbalancer, nodeport or ingress (Kubernetes only).
12
Enables or disables TLS encryption for each listener. For route and ingress type listeners, TLS encryption must always be enabled by setting it to true.
13
Defines whether the fully-qualified DNS names including the cluster service suffix (usually .cluster.local) are assigned.
14
Listener authentication mechanism specified as mTLS, SCRAM-SHA-512, or token-based OAuth 2.0.
15
External listener configuration specifies how the Kafka cluster is exposed outside OpenShift, such as through a route, loadbalancer or nodeport.
16
Optional configuration for a Kafka listener certificate managed by an external CA (certificate authority). The brokerCertChainAndKey specifies a Secret that contains a server certificate and a private key. You can configure Kafka listener certificates on any listener with enabled TLS encryption.
17
Authorization enables simple, OAUTH 2.0, or OPA authorization on the Kafka broker. Simple authorization uses the AclAuthorizer and StandardAuthorizer Kafka plugins.
18
Broker configuration. Standard Apache Kafka configuration may be provided, restricted to those properties not managed directly by Streams for Apache Kafka.
19
Storage size for persistent volumes may be increased and additional volumes may be added to JBOD storage.
20
Persistent storage has additional configuration options, such as a storage id and class for dynamic volume provisioning.
21
Rack awareness configuration to spread replicas across different racks, data centers, or availability zones. The topologyKey must match a node label containing the rack ID. The example used in this configuration specifies a zone using the standard topology.kubernetes.io/zone label.
22
Prometheus metrics enabled. In this example, metrics are configured for the Prometheus JMX Exporter (the default metrics exporter).
23
Rules for exporting metrics in Prometheus format to a Grafana dashboard through the Prometheus JMX Exporter, which are enabled by referencing a ConfigMap containing configuration for the Prometheus JMX exporter. You can enable metrics without further configuration using a reference to a ConfigMap containing an empty file under metricsConfig.valueFrom.configMapKeyRef.key.
24
ZooKeeper-specific configuration, which contains properties similar to the Kafka configuration.
25
The number of ZooKeeper nodes. ZooKeeper clusters or ensembles usually run with an odd number of nodes, typically three, five, or seven. The majority of nodes must be available in order to maintain an effective quorum. If the ZooKeeper cluster loses its quorum, it will stop responding to clients and the Kafka brokers will stop working. Having a stable and highly available ZooKeeper cluster is crucial for Streams for Apache Kafka.
26
ZooKeeper loggers and log levels.
27
Entity Operator configuration, which specifies the configuration for the Topic Operator and User Operator.
28
Entity Operator TLS sidecar configuration. Entity Operator uses the TLS sidecar for secure communication with ZooKeeper.
29
Specified Topic Operator loggers and log levels. This example uses inline logging.
30
Specified User Operator loggers and log levels.
31
Kafka Exporter configuration. Kafka Exporter is an optional component for extracting metrics data from Kafka brokers, in particular consumer lag data. For Kafka Exporter to be able to work properly, consumer groups need to be in use.
32
Optional configuration for Cruise Control, which is used to rebalance the Kafka cluster.

Use the Kafka Static Quota plugin to set throughput and storage limits on brokers in your Kafka cluster. You enable the plugin and set limits by configuring the Kafka resource. You can set a byte-rate threshold and storage quotas to put limits on the clients interacting with your brokers.

You can set byte-rate thresholds for producer and consumer bandwidth. The total limit is distributed across all clients accessing the broker. For example, you can set a byte-rate threshold of 40 MBps for producers. If two producers are running, they are each limited to a throughput of 20 MBps.

Storage quotas throttle Kafka disk storage limits between a soft limit and hard limit. The limits apply to all available disk space. Producers are slowed gradually between the soft and hard limit. The limits prevent disks filling up too quickly and exceeding their capacity. Full disks can lead to issues that are hard to rectify. The hard limit is the maximum storage limit.

注意

For JBOD storage, the limit applies across all disks. If a broker is using two 1 TB disks and the quota is 1.1 TB, one disk might fill and the other disk will be almost empty.

Prerequisites

  • The Cluster Operator that manages the Kafka cluster is running.

Procedure

  1. Add the plugin properties to the config of the Kafka resource.

    The plugin properties are shown in this example configuration.

    Example Kafka Static Quota plugin configuration

    apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    config:
      client.quota.callback.class: io.strimzi.kafka.quotas.StaticQuotaCallback
    1 
    
      client.quota.callback.static.produce: 1000000
    2 
    
      client.quota.callback.static.fetch: 1000000
    3 
    
      client.quota.callback.static.storage.soft: 400000000000
    4 
    
      client.quota.callback.static.storage.hard: 500000000000
    5 
    
      client.quota.callback.static.storage.check-interval: 5
    6

    1
    Loads the Kafka Static Quota plugin.
    2
    Sets the producer byte-rate threshold. 1 MBps in this example.
    3
    Sets the consumer byte-rate threshold. 1 MBps in this example.
    4
    Sets the lower soft limit for storage. 400 GB in this example.
    5
    Sets the higher hard limit for storage. 500 GB in this example.
    6
    Sets the interval in seconds between checks on storage. 5 seconds in this example. You can set this to 0 to disable the check.

  2. Update the resource. 

    oc apply -f <kafka_configuration_file>

9.2.2. Default ZooKeeper configuration values
复制链接

When deploying ZooKeeper with Streams for Apache Kafka, some of the default configuration set by Streams for Apache Kafka differs from the standard ZooKeeper defaults. This is because Streams for Apache Kafka sets a number of ZooKeeper properties with values that are optimized for running ZooKeeper within an OpenShift environment.

The default configuration for key ZooKeeper properties in Streams for Apache Kafka is as follows:

Expand
表 9.1. Default ZooKeeper Properties in Streams for Apache Kafka
PropertyDefault valueDescription

tickTime

2000

The length of a single tick in milliseconds, which determines the length of a session timeout.

initLimit

5

The maximum number of ticks that a follower is allowed to fall behind the leader in a ZooKeeper cluster.

syncLimit

2

The maximum number of ticks that a follower is allowed to be out of sync with the leader in a ZooKeeper cluster.

autopurge.purgeInterval

1

Enables the autopurge feature and sets the time interval in hours for purging the server-side ZooKeeper transaction log.

admin.enableServer

false

Flag to disable the ZooKeeper admin server. The admin server is not used by Streams for Apache Kafka.

重要

Modifying these default values as zookeeper.config in the Kafka custom resource may impact the behavior and performance of your ZooKeeper cluster.

9.2.3. Deleting Kafka nodes using annotations
复制链接

This procedure describes how to delete an existing Kafka node by using an OpenShift annotation. Deleting a Kafka node consists of deleting both the Pod on which the Kafka broker is running and the related PersistentVolumeClaim (if the cluster was deployed with persistent storage). After deletion, the Pod and its related PersistentVolumeClaim are recreated automatically.

警告

Deleting a PersistentVolumeClaim can cause permanent data loss and the availability of your cluster cannot be guaranteed. The following procedure should only be performed if you have encountered storage issues.

Prerequisites

  • A running Cluster Operator

Procedure

  1. Find the name of the Pod that you want to delete.

    Kafka broker pods are named <cluster_name>-kafka-<index_number>, where <index_number> starts at zero and ends at the total number of replicas minus one. For example, my-cluster-kafka-0.

  2. Use oc annotate to annotate the Pod resource in OpenShift: 

    oc annotate pod <cluster_name>-kafka-<index_number> strimzi.io/delete-pod-and-pvc="true"
  3. Wait for the next reconciliation, when the annotated pod with the underlying persistent volume claim will be deleted and then recreated.

9.2.4. Deleting ZooKeeper nodes using annotations
复制链接

This procedure describes how to delete an existing ZooKeeper node by using an OpenShift annotation. Deleting a ZooKeeper node consists of deleting both the Pod on which ZooKeeper is running and the related PersistentVolumeClaim (if the cluster was deployed with persistent storage). After deletion, the Pod and its related PersistentVolumeClaim are recreated automatically.

警告

Deleting a PersistentVolumeClaim can cause permanent data loss and the availability of your cluster cannot be guaranteed. The following procedure should only be performed if you have encountered storage issues.

Prerequisites

  • A running Cluster Operator

Procedure

  1. Find the name of the Pod that you want to delete.

    ZooKeeper pods are named <cluster_name>-zookeeper-<index_number>, where <index_number> starts at zero and ends at the total number of replicas minus one. For example, my-cluster-zookeeper-0.

  2. Use oc annotate to annotate the Pod resource in OpenShift: 

    oc annotate pod <cluster_name>-zookeeper-<index_number> strimzi.io/delete-pod-and-pvc="true"
  3. Wait for the next reconciliation, when the annotated pod with the underlying persistent volume claim will be deleted and then recreated.
Red Hat logoGithubredditYoutubeTwitter

学习

尝试、购买和销售

社区

關於紅帽

我们提供强化的解决方案，使企业能够更轻松地跨平台和环境（从核心数据中心到网络边缘）工作。

让开源更具包容性

红帽致力于替换我们的代码、文档和 Web 属性中存在问题的语言。欲了解更多详情，请参阅红帽博客.

关于红帽文档

通过我们的产品和服务，以及可以信赖的内容，帮助红帽用户创新并实现他们的目标。 了解我们当前的更新.

Legal Notice

Theme

© 2026 Red Hat返回顶部