Disk skewness across and within brokers while using kafka rebalance

[AAR256]

Description -

While trying out some performance testing, we observed that some of the brokers were getting more volume of data as compared to others. Also, within a single broker using JBOD setup, there is a significant disk imbalance observed.

Environment-

AKS K8s version - 1.22.6
Strimzi version - 0.27

Kafka CR -

---
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: west-1-kafka
  namespace: kafka
spec:
  kafka:
    version: 3.0.0
    replicas: 5
    resources: 
      requests: 
        memory: 2Gi
        cpu: 1
      limits:
        memory: 4Gi
        cpu: 2
    listeners:
      - name: external
        port: 9094
        type: ingress
        tls: true
        authentication:
          type: scram-sha-512
        configuration:
          bootstrap:
            host: xxx.net
          brokers:
          - broker: 0
            host: xxx.net
          - broker: 1
            host: xxx.net
          - broker: 2
            host: xxx.net
          - broker: 3
            host: xxx.net
          - broker: 4
            host: xxx.net
          
      - name: tls
        port: 9093
        type: internal
        tls: true
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: saslint
        port: 9095
        type: internal
        tls: true
        authentication:
          type: scram-sha-512
        configuration:
          useServiceDnsDomain: true    
    authorization:
      type: simple
      superUsers:
        - west-1-shared-admin
    config: 
    # https://strimzi.io/blog/2021/06/08/broker-tuning/
    # https://kafka.apache.org/documentation/#brokerconfigs
      num.partitions: 3
      default.replication.factor: 3
      min.insync.replicas: 2
      auto.create.topics.enable: false
      delete.topic.enable: true
      compression.type: producer
      message.max.bytes: 1048576
      unclean.leader.election.enable: false
      auto.leader.rebalance.enable: true
      acks: -1
      # Internal topic - __consumer_offsets settings for offset commits
      offsets.commit.required.acks: -1
      offsets.topic.num.partitions: 50
      offsets.topic.replication.factor: 3
      # Internal topic - __transaction_state settings for transactions (https://strimzi.io/docs/operators/latest/using.html#reliability_guarantees)
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      # It allows us to run brokers of different versions in one cluster, to avoid a downtime of the cluster during the upgrade. An update to the inter.broker.protocol.version is required when upgrading the Kafka version
      inter.broker.protocol.version: "3.0"
      # Message format version the broker will use to append messages to the logs. From Kafka 3.0.0, when the inter.broker.protocol.version is set to 3.0 or higher, the log.message.format.version option is ignored and doesn’t need to be set
      log.message.format.version: "3.0"
      # Log retention policies
      log.retention.check.interval.ms: 300000 # 5 min
      log.retention.bytes: -1
      log.retention.ms: 259200000 # 3 days
      log.roll.ms: 604800000 # 7 days
      log.segment.bytes: 1073741824 # 1 Gibibyte
      log.segment.delete.delay.ms: 60000  
      # Thread settings
      num.network.threads: 3 # Handles requests to the Kafka cluster, such as produce and fetch requests from client applications
      num.io.threads: 8 # The number of I/O threads to pick up the requests from request queue and process them
      num.recovery.threads.per.data.dir: 1 # The number of threads used for log loading at startup and flushing at shutdown
      queued.max.requests: 500 # To limit the number of requests allowed in the request queue
      # The default buffer settings for sending and receiving messages
      socket.send.buffer.bytes: 102400 # 100 Kibibytes
      socket.receive.buffer.bytes: 102400 # 100 Kibibytes
      socket.request.max.bytes: 104857600 # 100 Mebibytes
      # Size of the buffer for network requests
      replica.socket.receive.buffer.bytes: 65536 # 64 kibibytes
      # Log cleaner settings
      log.cleanup.policy: delete
      log.cleaner.backoff.ms: 15000
      log.cleaner.delete.retention.ms: 86400000 # 1 day
      log.cleaner.dedupe.buffer.size: 134217728
      log.cleaner.io.buffer.load.factor: 0.9
      log.cleaner.threads: 8
      log.cleaner.io.max.bytes.per.second: 1.7976931348623157E308
      # Settings for log flush from in-memory to disk
      log.flush.scheduler.interval.ms: 2000
      log.flush.interval.ms: 50000
      log.flush.interval.messages: 100000
      # The lag time before a follower is considered out of sync
      replica.lag.time.max.ms: 30000
      # A maximum size for the messages fetched by each follower that replicates the data from a leader partition.
      replica.fetch.max.bytes: 1048576 # 1 Mebibyte
      # Leader imbalance check      
      leader.imbalance.check.interval.seconds: 300
      leader.imbalance.per.broker.percentage: 10
      # A delay for the group coordinator to wait for members to join before the initial rebalance
      group.initial.rebalance.delay.ms: 3000
      # The max time to establish connection to zookeeper
      zookeeper.connection.timeout.ms: 18000      
    storage:
      type: jbod
      volumes: 
      - id: 1
        type: persistent-claim
        size: 300Gi
        deleteClaim: false
        class: managed-csi-zrs-retain
      - id: 2
        type: persistent-claim
        size: 300Gi
        deleteClaim: false
        class: managed-csi-zrs-retain
      
    rack:
      topologyKey: topology.kubernetes.io/zone
    tlate:
      pod:
        affinity:
          nodeAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              nodeSelectorTerms:
              - matchExpressions:
                - key: team
                  operator: In
                  values:
                  - test
        tolerations:
        - key: "kubernetes.azure.com/scalesetpriority"
          operator: "Equal"
          value: "spot"
          effect: "NoSchedule"
        topologySpreadConstraints:
        - maxSkew: 1
          topologyKey: topology.kubernetes.io/zone
          whenUnsatisfiable: ScheduleAnyway
          labelSelector:
            matchLabels:
              strimzi.io/name: west-1-shared-kafka-kafka
        - maxSkew: 1
          topologyKey: kubernetes.io/hostname
          whenUnsatisfiable: ScheduleAnyway
          labelSelector:
            matchLabels:
              strimzi.io/name: west-1-shared-kafka-kafka
    metricsConfig:
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          name: kafka-metrics
          key: kafka-metrics-config.yml
  zookeeper:
    replicas: 3
    resources:
      requests:   
        memory: 2Gi
        cpu: 1
      limits:
        memory: 4Gi
        cpu: 2
    storage:
      type: persistent-claim
      size: 100Gi
      deleteClaim: false
      class: managed-csi-zrs-retain
    tlate:
      pod:
        affinity:
          nodeAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              nodeSelectorTerms:
              - matchExpressions:
                - key: team
                  operator: In
                  values:
                  - test
        tolerations:
        - key: "kubernetes.azure.com/scalesetpriority"
          operator: "Equal"
          value: "spot"
          effect: "NoSchedule"
        topologySpreadConstraints:
        - maxSkew: 1
          topologyKey: topology.kubernetes.io/zone
          whenUnsatisfiable: ScheduleAnyway
          labelSelector:
            matchLabels:
              strimzi.io/name: west-1-shared-kafka-zookeeper
        - maxSkew: 1
          topologyKey: kubernetes.io/hostname
          whenUnsatisfiable: ScheduleAnyway
          labelSelector:
            matchLabels:
              strimzi.io/name: west-1-shared-kafka-zookeeper
    metricsConfig:
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          name: kafka-metrics
          key: zookeeper-metrics-config.yml
  entityOperator:
    tlate:
      pod:
        affinity:
          nodeAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              nodeSelectorTerms:
              - matchExpressions:
                - key: team
                  operator: In
                  values:
                  - test
        tolerations:
        - key: "kubernetes.azure.com/scalesetpriority"
          operator: "Equal"
          value: "spot"
          effect: "NoSchedule"
        topologySpreadConstraints:
        - maxSkew: 1
          topologyKey: topology.kubernetes.io/zone
          whenUnsatisfiable: ScheduleAnyway
          labelSelector:
            matchLabels:
              strimzi.io/name: west-1-shared-kafka-entity-operator
        - maxSkew: 1
          topologyKey: kubernetes.io/hostname
          whenUnsatisfiable: ScheduleAnyway
          labelSelector:
            matchLabels:
              strimzi.io/name: west-1-shared-kafka-entity-operator
    tlsSidecar:
      resources: 
        requests:
          cpu: 100m
          memory: 100Mi
        limits:
          cpu: 500m
          memory: 500Mi
    topicOperator:
      resources:
        requests: 
          memory: 1Gi
          cpu: 500m
        limits:
          memory: 2Gi
          cpu: 1
    userOperator:
      resources:
        requests: 
          memory: 1Gi
          cpu: 500m
        limits:
          memory: 2Gi
          cpu: 1
  cruiseControl:
    tlate:
      pod:
        affinity:
          nodeAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              nodeSelectorTerms:
              - matchExpressions:
                - key: team
                  operator: In
                  values:
                  - test
        tolerations:
        - key: "kubernetes.azure.com/scalesetpriority"
          operator: "Equal"
          value: "spot"
          effect: "NoSchedule"
    tlsSidecar:
      resources: 
        requests:
          memory: 500Mi
          cpu: 500m
        limits:
          memory: 1Gi
          cpu: 1
    resources: 
      requests:
        memory: 1Gi
        cpu: 500m
      limits:
        memory: 2Gi
        cpu: 1
    metricsConfig:
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          name: cruise-control-metrics
          key: metrics-config.yml
  kafkaExporter:
    tlate:
      pod:
        affinity:
          nodeAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              nodeSelectorTerms:
              - matchExpressions:
                - key: team
                  operator: In
                  values:
                  - test
    resources: 
      requests:
        memory: 1Gi
        cpu: 500m
      limits:
        memory: 2Gi
        cpu: 1
    topicRegex: ".*"
    groupRegex: ".*"
  clusterCa:
    renewalDays: 30
    validityDays: 365
    generateCertificateAuthority: true
    generateSecretOwnerReference: false
  clientsCa:
    renewalDays: 30
    validityDays: 365
    generateCertificateAuthority: true
    generateSecretOwnerReference: false
---

kind: ConfigMap
apiVersion: v1
metadata:
  name: kafka-metrics
  namespace: kafka
  annotations:
    kustomize.toolkit.fluxcd.io/substitute: disabled
  labels:
    app: strimzi
data:
  kafka-metrics-config.yml: |
    # See https://github.com/prometheus/jmx_exporter for more info about JMX Prometheus Exporter metrics
    lowercaseOutputName: true
    rules:
    # Special cases and very specific rules
    - pattern: kafka.server<type=(.+), name=(.+), clientId=(.+), topic=(.+), partition=(.*)><>Value
      name: kafka_server_$1_$2
      type: GAUGE
      labels:
       clientId: "$3"
       topic: "$4"
       partition: "$5"
    - pattern: kafka.server<type=(.+), name=(.+), clientId=(.+), brokerHost=(.+), brokerPort=(.+)><>Value
      name: kafka_server_$1_$2
      type: GAUGE
      labels:
       clientId: "$3"
       broker: "$4:$5"
    - pattern: kafka.server<type=(.+), cipher=(.+), protocol=(.+), listener=(.+), networkProcessor=(.+)><>connections
      name: kafka_server_$1_connections_tls_info
      type: GAUGE
      labels:
        cipher: "$2"
        protocol: "$3"
        listener: "$4"
        networkProcessor: "$5"
    - pattern: kafka.server<type=(.+), clientSoftwareName=(.+), clientSoftwareVersion=(.+), listener=(.+), networkProcessor=(.+)><>connections
      name: kafka_server_$1_connections_software
      type: GAUGE
      labels:
        clientSoftwareName: "$2"
        clientSoftwareVersion: "$3"
        listener: "$4"
        networkProcessor: "$5"
    - pattern: "kafka.server<type=(.+), listener=(.+), networkProcessor=(.+)><>(.+):"
      name: kafka_server_$1_$4
      type: GAUGE
      labels:
       listener: "$2"
       networkProcessor: "$3"
    - pattern: kafka.server<type=(.+), listener=(.+), networkProcessor=(.+)><>(.+)
      name: kafka_server_$1_$4
      type: GAUGE
      labels:
       listener: "$2"
       networkProcessor: "$3"
    # Some percent metrics use MeanRate attribute
    # Ex) kafka.server<type=(KafkaRequestHandlerPool), name=(RequestHandlerAvgIdlePercent)><>MeanRate
    - pattern: kafka.(\w+)<type=(.+), name=(.+)Percent\w*><>MeanRate
      name: kafka_$1_$2_$3_percent
      type: GAUGE
    # Generic gauges for percents
    - pattern: kafka.(\w+)<type=(.+), name=(.+)Percent\w*><>Value
      name: kafka_$1_$2_$3_percent
      type: GAUGE
    - pattern: kafka.(\w+)<type=(.+), name=(.+)Percent\w*, (.+)=(.+)><>Value
      name: kafka_$1_$2_$3_percent
      type: GAUGE
      labels:
        "$4": "$5"
    # Generic per-second counters with 0-2 key/value pairs
    - pattern: kafka.(\w+)<type=(.+), name=(.+)PerSec\w*, (.+)=(.+), (.+)=(.+)><>Count
      name: kafka_$1_$2_$3_total
      type: COUNTER
      labels:
        "$4": "$5"
        "$6": "$7"
    - pattern: kafka.(\w+)<type=(.+), name=(.+)PerSec\w*, (.+)=(.+)><>Count
      name: kafka_$1_$2_$3_total
      type: COUNTER
      labels:
        "$4": "$5"
    - pattern: kafka.(\w+)<type=(.+), name=(.+)PerSec\w*><>Count
      name: kafka_$1_$2_$3_total
      type: COUNTER
    # Generic gauges with 0-2 key/value pairs
    - pattern: kafka.(\w+)<type=(.+), name=(.+), (.+)=(.+), (.+)=(.+)><>Value
      name: kafka_$1_$2_$3
      type: GAUGE
      labels:
        "$4": "$5"
        "$6": "$7"
    - pattern: kafka.(\w+)<type=(.+), name=(.+), (.+)=(.+)><>Value
      name: kafka_$1_$2_$3
      type: GAUGE
      labels:
        "$4": "$5"
    - pattern: kafka.(\w+)<type=(.+), name=(.+)><>Value
      name: kafka_$1_$2_$3
      type: GAUGE
    # Emulate Prometheus 'Summary' metrics for the exported 'Histogram's.
    # Note that these are missing the '_sum' metric!
    - pattern: kafka.(\w+)<type=(.+), name=(.+), (.+)=(.+), (.+)=(.+)><>Count
      name: kafka_$1_$2_$3_count
      type: COUNTER
      labels:
        "$4": "$5"
        "$6": "$7"
    - pattern: kafka.(\w+)<type=(.+), name=(.+), (.+)=(.*), (.+)=(.+)><>(\d+)thPercentile
      name: kafka_$1_$2_$3
      type: GAUGE
      labels:
        "$4": "$5"
        "$6": "$7"
        quantile: "0.$8"
    - pattern: kafka.(\w+)<type=(.+), name=(.+), (.+)=(.+)><>Count
      name: kafka_$1_$2_$3_count
      type: COUNTER
      labels:
        "$4": "$5"
    - pattern: kafka.(\w+)<type=(.+), name=(.+), (.+)=(.*)><>(\d+)thPercentile
      name: kafka_$1_$2_$3
      type: GAUGE
      labels:
        "$4": "$5"
        quantile: "0.$6"
    - pattern: kafka.(\w+)<type=(.+), name=(.+)><>Count
      name: kafka_$1_$2_$3_count
      type: COUNTER
    - pattern: kafka.(\w+)<type=(.+), name=(.+)><>(\d+)thPercentile
      name: kafka_$1_$2_$3
      type: GAUGE
      labels:
        quantile: "0.$4"
  -zookeeper-metrics-config.yml: |
    # See https://github.com/prometheus/jmx_exporter for more info about JMX Prometheus Exporter metrics
    lowercaseOutputName: true
    rules:
    # replicated Zookeeper
    - pattern: "org.apache.ZooKeeperService<name0=ReplicatedServer_id(\\d+)><>(\\w+)"
      name: "zookeeper_$2"
      type: GAUGE
    - pattern: "org.apache.ZooKeeperService<name0=ReplicatedServer_id(\\d+), name1=replica.(\\d+)><>(\\w+)"
      name: "zookeeper_$3"
      type: GAUGE
      labels:
        replicaId: "$2"
    - pattern: "org.apache.ZooKeeperService<name0=ReplicatedServer_id(\\d+), name1=replica.(\\d+), name2=(\\w+)><>(Packets\\w+)"
      name: "zookeeper_$4"
      type: COUNTER
      labels:
        replicaId: "$2"
        memberType: "$3"
    - pattern: "org.apache.ZooKeeperService<name0=ReplicatedServer_id(\\d+), name1=replica.(\\d+), name2=(\\w+)><>(\\w+)"
      name: "zookeeper_$4"
      type: GAUGE
      labels:
        replicaId: "$2"
        memberType: "$3"
    - pattern: "org.apache.ZooKeeperService<name0=ReplicatedServer_id(\\d+), name1=replica.(\\d+), name2=(\\w+), name3=(\\w+)><>(\\w+)"
      name: "zookeeper_$4_$5"
      type: GAUGE
      labels:
        replicaId: "$2" 
        memberType: "$3"
---
kind: ConfigMap
apiVersion: v1
metadata:
  name: -cruise-control-metrics
  namespace: kafka
  annotations:
    kustomize.toolkit.fluxcd.io/substitute: disabled
  labels:
    app: strimzi
data:
  -metrics-config.yml: |
    # See https://github.com/prometheus/jmx_exporter for more info about JMX Prometheus Exporter metrics
    lowercaseOutputName: true
    rules:
    - pattern: kafka.cruisecontrol<name=(.+)><>(\w+)
      name: kafka_cruisecontrol_$1_$2
      type: GAUGE
---

Kafka Rebalance CR -

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaRebalance
metadata:
  name: west-1-kafka-rebalance
  namespace: kafka
  labels:
    strimzi.io/cluster: west-1-kafka
spec:
  skipHardGoalCheck: true
  goals:
    - CpuCapacityGoal
    - DiskCapacityGoal
    - RackAwareGoal
    - MinTopicLeadersPerBrokerGoal
    - DiskUsageDistributionGoal

After applying and approving the proposal, we produced some messages to test the balancing features according to the goals we have setup. But there was a huge imbalance observed between the brokers when it comes to disk usage.

While viewing the rebalance spec, it's showing the data to move is 0 MB. -

Spec:
  Goals:
    CpuCapacityGoal
    DiskCapacityGoal
    RackAwareGoal
    MinTopicLeadersPerBrokerGoal
    DiskUsageDistributionGoal
  Skip Hard Goal Check:  true
Status:
  Conditions:
    Last Transition Time:  2022-08-23T12:28:15.349343Z
    Status:                True
    Type:                  Ready
  Observed Generation:     1
  Optimization Result:
    After Before Load Config Map:  west-1-kafka-rebalance
    Data To Move MB:               0
    Excluded Brokers For Leadership:
    Excluded Brokers For Replica Move:
    Excluded Topics:
    Intra Broker Data To Move MB:         0
    Monitored Partitions Percentage:      100
    Num Intra Broker Replica Movements:   0
    Num Leader Movements:                 0
    Num Replica Movements:                0
    On Demand Balancedness Score After:   88.44952152142903
    On Demand Balancedness Score Before:  88.44952152142903
    Provision Recommendation:             
    Provision Status:                     RIGHT_SIZED
    Recent Windows:                       1
Events:                                   <none>

What could be the reason behind this? Anything we might be doing wrong here. Any help would be appreciated.

[scholzj]

This is how Kafka works. It does not do any balancing on the fly and depending on your architecture equal disk usage might even not be possible since single partition is always only on single disk. If you want, you can use the intrabroker balancing to balance the data with Cruise Control between the JBOD disks.

[AAR256]

Just to clarify, in our setup we have 5 brokers running with replication factor of 3 so a single partition would be replicated in 2 other broker disks. For intrabroker balancing, is there any separate configuration we have to provide as part of kafka CR?

[AAR256]

When trying to view the current partitions, we found the below number partitions for all the brokers -

No. of Topics - 5, Replication Factor - 3, Partitions - 3.

Broker 0 - 8
Broker 1 - 7
Broker 2 - 15
Broker 3- 7
Broker 4 - 8

By looking at this, it seems the partition rebalancing isn't happening. That's why the disk usage is high on broker 2 compared to others. Shouldn't kafka by default rebalance the partitions across the brokers as we setup auto.leader.rebalance.enable: true in Kafka CR ?

[scholzj]

For intrabroker balancing, is there any separate configuration we have to provide as part of kafka CR?

There is a flag for it. Please check the docs.

Shouldn't kafka by default rebalance the partitions across the brokers as we setup auto.leader.rebalance.enable: true in Kafka CR ?

No, this only means that the leader shifts back to the preferred leader node for given partition. It does not move partitions. Check https://kafka.apache.org/documentation/#basic_ops_leader_balancing for more details.

By looking at this, it seems the partition rebalancing isn't happening.

First of all, number of partitions is only one of many factors for the balancedness - and relatively minor one as each partition has different load / size etc. So nothin in the table suggests your cluster is out of balance. If you used Cruise Control on this, it is probably optimised towards more important goals such as CPU utilization, traffic, disk usage etc.