Kafka on Kubernetes 运维面试题

Kafka CR 创建 → Cluster Operator Watch → 
  创建 ZK StatefulSet（非 KRaft 模式）→ 
  创建 Broker StatefulSet → 
  Entity Operator（Topic Operator + User Operator）启动 → 
  监听 KafkaTopic / KafkaUser CRD → 
  通过 Admin API 操作 Kafka 集群

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    replicas: 3
    version: 3.7.0
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
      - name: external
        port: 9094
        type: nodeport
        tls: true
    storage:
      type: jbod
      volumes:
        - id: 0
          type: persistent-claim
          size: 100Gi
          deleteClaim: false
        - id: 1
          type: persistent-claim
          size: 100Gi
          deleteClaim: false
    config:
      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
    template:
      pod:
        affinity:
          podAntiAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              - labelSelector:
                  matchExpressions:
                    - key: strimzi.io/name
                      operator: In
                      values:
                        - my-cluster-kafka
                topologyKey: kubernetes.io/hostname
    resources:
      requests:
        memory: 8Gi
        cpu: 2
      limits:
        memory: 16Gi
        cpu: 4

KafkaTopic CR 创建/更新 → Topic Operator Watch 事件 → 
  反序列化 CR → 校验分区数/副本因子/配置 → 
  调用 AdminClient.createTopics() / AdminClient.describeConfigs() → 
  对比期望状态与当前状态 → 执行变更 → 更新 KafkaTopic.status

KafkaUser CR 创建/更新 → User Operator Watch 事件 → 
  生成凭证（SCRAM-SHA-512 密码 / TLS 证书）→ 
  存储到 Secret → 调用 Admin API 创建/更新用户 → 
  绑定 ACL 规则 → 设置 Quota → 更新 KafkaUser.status

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    replicas: 3
    rack:
      topologyKey: topology.kubernetes.io/zone
    listeners:
      - name: plain
        port: 9092
        type: internal
        affinity:
          podAntiAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              - labelSelector:
                  matchExpressions:
                    - key: strimzi.io/name
                      operator: In
                      values:
                        - my-cluster-kafka
                  topologyKey: kubernetes.io/hostname
        topologySpreadConstraints:        # topologySpreadConstraints 与 podAntiAffinity 并列
          - maxSkew: 1
            topologyKey: topology.kubernetes.io/zone
            whenUnsatisfiable: DoNotSchedule
            labelSelector:
              matchLabels:
                strimzi.io/name: my-cluster-kafka

template:
  pod:
    topologySpreadConstraints:
      - maxSkew: 1
        topologyKey: topology.kubernetes.io/zone
        whenUnsatisfiable: DoNotSchedule
        labelSelector:
          matchLabels:
            strimzi.io/name: my-cluster-kafka
      - maxSkew: 1
        topologyKey: kubernetes.io/hostname
        whenUnsatisfiable: DoNotSchedule
        labelSelector:
          matchLabels:
            strimzi.io/name: my-cluster-kafka

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: dr-mirror
spec:
  version: 3.7.0
  replicas: 2
  connectCluster: target-cluster
  clusters:
    - alias: source-cluster
      bootstrapServers: source-kafka-bootstrap.source-ns.svc:9092
    - alias: target-cluster
      bootstrapServers: target-kafka-bootstrap.target-ns.svc:9092
  mirrors:
    - sourceCluster: source-cluster
      targetCluster: target-cluster
      sourceConnector:
        config:
          replication.factor: 3
          offset-syncs.topic.replication.factor: 3
          sync.topic.acls.enabled: true
          refresh.topics.enabled: true
          refresh.topics.interval.seconds: 60
          tasks.max: 4
      topicsPattern: "orders.*|payments.*|inventory.*"
      groupsPattern: "order-.*|payment-.*"
      checkpointConnector:
        config:
          checkpoints.topic.replication.factor: 3
          refresh.groups.enabled: true
          refresh.groups.interval.seconds: 60
          sync.group.offsets.enabled: true
          emit.checkpoints.enabled: true
          tasks.max: 2
      heartbeatConnector:
        config:
          heartbeats.topic.replication.factor: 3
          tasks.max: 1

源集群 Topic: orders
目标集群 Topic: source-cluster.orders

复制逻辑:
  MM2 从源集群消费 orders → 写入目标集群 source-cluster.orders →
  依赖 Kafka Connect 的 Exactly-Once 支持保证不丟不重 →
  心跳 Topic（source-cluster.heartbeats）持续写入监控延迟 →
  Checkpoint Topic 记录 Consumer Group Offset 映射

[Source DC]
  Kafka Cluster (source-cluster)
         |
  MM2 Workers (active-standby)
         |
[Target DC]
  Kafka Cluster (target-cluster)
   - source-cluster.orders
   - source-cluster.payments
   - offsets synced

spec:
  kafka:
    replicas: 3
    resources:
      requests:
        memory: 8Gi
        cpu: 2
      limits:
        memory: 16Gi
        cpu: 4
    jvmOptions:
      -Xms: 6144m
      -Xmx: 6144m         # heap = limit * 0.5（剩余给 Page Cache）
      -XX:+UseG1GC
      -XX:MaxGCPauseMillis: 20
      -XX:G1HeapRegionSize: 32m
      -XX:MetaspaceSize: 128m
      -XX:MaxMetaspaceSize: 256m
      -XX:+ExitOnOutOfMemoryError
    storage:
      type: jbod
      volumes:
        - id: 0
          type: persistent-claim
          size: 500Gi
          class: nvme-ssd
    template:
      pod:
        affinity:
          podAntiAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              - labelSelector:
                  matchExpressions:
                    - key: strimzi.io/name
                      operator: In
                      values:
                        - my-cluster-kafka
                topologyKey: kubernetes.io/hostname
        tolerations:
          - key: dedicated
            operator: Equal
            value: kafka
            effect: NoSchedule
      serviceAccount:
        metadata:
          annotations:
            iam.gke.io/gcp-service-account: [email protected]

# Cruise Control 异常检测
spec:
  cruiseControl:
    config:
      anomaly.detection.goals: >-
        com.linkedin.kafka.cruisecontrol.analyzer.goals.RackAwareGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaCapacityGoal
      failed.brokers.zk.session.timeout.ms: 15000
      metric.anomaly.finder.class: >-
        com.linkedin.kafka.cruisecontrol.detector.NoopMetricAnomalyFinder
      goal.violation.detection.interval.ms: 300000

groups:
  - name: kafka-critical
    rules:
      - alert: KafkaUnderReplicatedPartitions
        expr: kafka_server_replica_manager_underreplicatedpartitions > 0
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Kafka Under Replicated Partitions = {{ $value }}"

      - alert: KafkaUnderMinISR
        expr: kafka_server_replica_manager_underminisrpartitioncount > 0
        for: 2m
        labels:
          severity: critical

      - alert: KafkaBrokerDown
        expr: kafka_server_replicamanager_leadercount < 1
        for: 2m
        labels:
          severity: critical

      - alert: KafkaActiveControllerCount
        expr: sum(kafka_controller_activecontroller) != 1
        for: 1m
        labels:
          severity: critical

      - alert: KafkaHighConsumerLag
        expr: kafka_consumergroup_group_lag > 100000
        for: 10m
        labels:
          severity: warning

      - alert: KafkaDiskUsageHigh
        expr: (kafka_log_log_size / kafka_log_log_size_limit) > 0.85
        for: 5m
        labels:
          severity: warning

Strimzi 版本升级流程：
  1. 阅读 Strimzi 版本升级说明（Upgrade Guide）
  2. 备份 Kafka CRD 和配置
  3. 升级 Strimzi Cluster Operator（Helm / OLM）
  4. 升级 CRD 定义
  5. 升级 Kafka CR 中的 version 字段（Kafka 版本）
  6. Operator 自动触发 Broker 滚动升级
  7. 验证集群状态和客户端兼容性

跨大版本升级原则：
  - 不跳版本升级（如 3.5 → 3.6 → 3.7）
  - 先升级 inter.broker.protocol.version，再升级 log.message.format.version
  - 客户端版本与 Broker 版本的兼容性：**旧客户端连新 Broker 保证兼容（N-1）**；新客户端连旧 Broker 不保证