Cluster operator reconciliation gets bugged and requires restart.

[AAR256]

Description -

While trying to add a disk onto the currently running Kafka Cluster, the changes got applied to the cluster CR wherein we can view the newly added disk but the same wasn't being applied to the Kafka stateful sets.

Environment-

AKS K8s version - 1.22.6
Strimzi version - 0.27

Kafka CR -

---
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: west-1-shared-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: west-1-shared-kafka-bootstrap.net
          brokers:
          - broker: 0
            host: west-1-shared-kafka-kafka-0.net
          - broker: 1
            host: west-1-shared-kafka-kafka-1.net
          - broker: 2
            host: west-1-shared-kafka-kafka-2.net
          - broker: 3
            host: west-1-shared-kafka-kafka-3.net
          - broker: 4
            host: west-1-shared-kafka-kafka-4.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: 200Gi
        deleteClaim: false
        class: managed-csi-zrs-retain
      - id: 2
        type: persistent-claim
        size: 200Gi
        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
---

Onto the existing cluster the change that we made was adding a new disk with id 3 with the below configuration -

       - id: 3
        type: persistent-claim
        size: 100Gi
        deleteClaim: false
        class: managed-csi-zrs-retain

Note - For our current cluster operator deployment we're running two pods and the logs for the same when we encountered this issue were below.

2022-07-26 08:17:38 ERROR AbstractOperator:499 - Watcher closed with exception in namespace *
io.fabric8.kubernetes.client.WatcherException: too old resource version: 435743338 (441459178)
	at io.fabric8.kubernetes.client.dsl.internal.AbstractWatchManager.onStatus(AbstractWatchManager.java:273) [io.fabric8.kubernetes-client-5.10.2.jar:?]
	at io.fabric8.kubernetes.client.dsl.internal.AbstractWatchManager.onMessage(AbstractWatchManager.java:257) [io.fabric8.kubernetes-client-5.10.2.jar:?]
	at io.fabric8.kubernetes.client.dsl.internal.WatcherWebSocketListener.onMessage(WatcherWebSocketListener.java:93) [io.fabric8.kubernetes-client-5.10.2.jar:?]
	at okhttp3.internal.ws.RealWebSocket.onReadMessage(RealWebSocket.java:323) [com.squareup.okhttp3.okhttp-3.12.6.jar:?]
	at okhttp3.internal.ws.WebSocketReader.readMessageFrame(WebSocketReader.java:219) [com.squareup.okhttp3.okhttp-3.12.6.jar:?]
	at okhttp3.internal.ws.WebSocketReader.processNextFrame(WebSocketReader.java:105) [com.squareup.okhttp3.okhttp-3.12.6.jar:?]
	at okhttp3.internal.ws.RealWebSocket.loopReader(RealWebSocket.java:274) [com.squareup.okhttp3.okhttp-3.12.6.jar:?]
	at okhttp3.internal.ws.RealWebSocket$2.onResponse(RealWebSocket.java:214) [com.squareup.okhttp3.okhttp-3.12.6.jar:?]
	at okhttp3.RealCall$AsyncCall.execute(RealCall.java:203) [com.squareup.okhttp3.okhttp-3.12.6.jar:?]
	at okhttp3.internal.NamedRunnable.run(NamedRunnable.java:32) [com.squareup.okhttp3.okhttp-3.12.6.jar:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
	at java.lang.Thread.run(Thread.java:829) [?:?]

After restarting both the pods, the reconciliation for the storage change triggered and everything was setup in place. Any help on why this has happened and if there is any way to maintain and observe the status of cluster operator would be really helpful.

[scholzj]

First of all, you should not be running two cluster operators watching the same CR. If you do, they will fight with each other and it will cause problems. You should always have only one cluster operator operate a single cluster.

The error you pasted is fairly normal -> it just means the Kubernetes resource watch is too old and needs to be recreated. How often that happens depends on your cluster size and how busy it is. On its own it should not cause any issues.

However, in theory, the first issue can cause this error to happen more often and can cause other issues as well. So I think the first thing you need to do is to make sure you have only one operator pod monitor given custom resource.

[AAR256]

One question - Does having multiple replicas of a single cluster operator advisable? Could this cause some issues.

[scholzj]

It does not matter whether you have multiple replicas of the same deployment or two deployments. The effects are the same -> you will have multiple pods trying to do the same and causing issues. You should have only one operator pod monitoring given custom resource.

[AAR256]

Alright, thanks for the input. Will make the suggested changes and test it out.