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
    template:
      pod:
        affinity:
          podAntiAffinity:
            preferredDuringSchedulingIgnoredDuringExecution:
              - weight: 100
                podAffinityTerm:
                  labelSelector:
                    matchExpressions:
                      - key: strimzi.io/name
                        operator: In
                        values:
                          - my-cluster-kafka
                  topologyKey: kubernetes.io/hostname

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: KafkaTopic
metadata:
  name: orders-topic
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 12
  replicas: 3
  config:
    retention.ms: 604800000        # 7 天保留
    retention.bytes: 107374182400  # 100GB 保留
    segment.bytes: 1073741824      # 1GB 每段
    min.insync.replicas: 2
    cleanup.policy: delete
    compression.type: producer
    max.message.bytes: 10485760    # 10MB
    message.timestamp.type: CreateTime

单分区日均存储 = 日均消息数 × 平均消息大小 × 副本因子 × (1 + segment索引开销≈10%) × 保留天数
总存储 = 单分区存储 × 分区数 + 预留 30% buffer

Follower 进入 ISR 条件：
  - Follower 在 replica.lag.time.max.ms（默认 30s）内向 Leader 发送过 Fetch 请求
  - Follower 的 LEO（Log End Offset）与 Leader 的差异在 replica.lag.max.messages 范围内

Follower 踢出 ISR 条件：
  - 超过 replica.lag.time.max.ms 未向 Leader 发起 Fetch 请求
  - Follower LEO 落后 Leader LEO 超过阈值

Leader 维护 ISR 集合 → 
  Follower Fetch 请求到达 → 更新 lastCaughtUpTimeMs → 
  定期检查：now - lastCaughtUpTimeMs > replica.lag.time.max.ms → 
  从 ISR 移除 → 写入 ZooKeeper / KRaft → 
  通知 Controller 更新 ISR 元数据 → Propagate 至所有 Broker → 
  Follower 追上后重新加入 ISR

Producer acks=all + min.insync.replicas=2 + replicationFactor=3
  → 消息写入 Leader → Leader 等待 Follower 确认 → 
  至少 2 个副本（含 Leader）持久化成功 → Producer 收到 ACK

Broker 启动 → 尝试创建 /controller 临时节点 → 
  创建成功 → 成为 Controller，监听 ZooKeeper 变更 → 
  创建失败 → 监听 /controller 节点 → 
  当前 Controller 宕机 → ZooKeeper 临时节点删除 → 
  剩余 Broker 收到 Watch 通知 → 重新竞争创建 /controller → 
  新 Controller 当选

Broker 宕机检测 → ZooKeeper 临时节点超时 → 
Controller 收到 Broker 变更通知 → 
遍历该 Broker 所有 Leader 分区 → 
从 ISR 中选取第一个存活 Broker 作为新 Leader → 
更新 ZooKeeper /brokers/topics/<topic>/partitions/<pid>/state → 
向所有 Broker 广播 LeaderAndIsr 请求 → 
各 Broker 更新本地元数据缓存

Producer 初始化 → 向 Broker 请求 Producer ID（PID）→ 
  Broker 分配 PID + Epoch → Producer 每条消息附带：
    (PID, Epoch, SequenceNumber) → 
  Broker 按 (PID, Partition) 维护已提交的最大 SequenceNumber → 
  收到消息 → 检查：
    Sequence == lastSeq+1 → 接受
    Sequence <= lastSeq → 重复，丢弃，返回 ACK
    Sequence > lastSeq+1 → 乱序，抛出 OutOfOrderSequenceException

# 幂等生产者配置
enable.idempotence=true
acks=all
max.in.flight.requests.per.connection=5
retries=2147483647

Producer 初始化事务 → initTransactions() → 
  Broker 分配 TransactionalId → 
  beginTransaction() → 
  发送消息（附加 TransactionalId）→ 
  消息写入分区但不向消费者可见 → 
  sendOffsetsToTransaction() → 
  commitTransaction() / abortTransaction() → 
  Broker 写入 Commit/Abort Marker → 
  消费者读取到 Marker 后过滤事务消息

# Producer 端压缩（推荐）
compression.type=zstd          # none / gzip / snappy / lz4 / zstd

# Broker 端压缩策略
compression.type=producer      # producer(保留原始压缩) / uncompressed(强制解压)

# Topic 级别压缩覆盖
# 通过 KafkaTopic CRD 的 config 指定

Producer 压缩消息 → 封装为 MessageSet → 发送到 Broker → 
  Broker.compression.type=producer → 原样写入 Page Cache → 
  批量刷盘到日志段文件 → 
  Consumer Fetch 请求 → Broker 直接发送压缩数据 → 
  Consumer 解压消息集

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
spec:
  kafka:
    config:
      compression.type: producer       # Broker 端保持 Producer 压缩
      message.max.bytes: 10485760      # 单条消息最大 10MB

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: events-topic
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 12
  replicas: 3
  config:
    # 基于时间 + 大小双重删除
    cleanup.policy: delete
    retention.ms: 604800000          # 7 天
    retention.bytes: 107374182400    # 100GB / 分区

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: user-state-compacted
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 6
  replicas: 3
  config:
    cleanup.policy: compact
    min.cleanable.dirty.ratio: 0.5
    delete.retention.ms: 86400000    # 删除标记保留 1 天
    segment.ms: 86400000             # 每天产生新段

Log Cleaner 线程定期扫描 → 选择脏比率最高的 Log → 
  逐段读取 → 构建 Offset Map（每个 Key 的最新 Offset）→ 
  保留最新 Offset 的消息，删除旧消息 → 合并为干净的新段 → 
  交换新旧段 → 更新索引

Phase 1: JoinGroup
  Consumer → Group Coordinator: JoinGroupRequest（memberId, protocols）
  Coordinator 选第一个 Consumer 为 Group Leader
  Coordinator → Group Leader: JoinGroupResponse（所有 Consumer 信息）
  Coordinator → 其他 Consumer: JoinGroupResponse（成员 ID）

Phase 2: SyncGroup  
  Group Leader 执行分配策略 → 生成 Partition Assignments → 
  Group Leader → Coordinator: SyncGroupRequest（分配方案）
  Coordinator → 所有 Consumer: SyncGroupResponse（各 Consumer 的分区列表）

Consumer 收到分配方案 → 清空或保持未提交的 Offset → 开始消费

与传统 Eager Rebalance 差异：
  - 不触发全局 Stop-The-World
  - Consumer 不必在 Rebalance 期间暂停所有分区消费
  - 仅迁移需要重新分配的分区
  - 未被重新分配的分区继续正常消费

Consumer 配置：
  partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor

# 延长 Session 超时，避免短暂网络波动触发 Rebalance
session.timeout.ms=30000          # Consumer 心跳超时（默认 45s）

# 延长心跳间隔，减少心跳开销
heartbeat.interval.ms=3000        # Consumer 向 Coordinator 发送心跳间隔

# 增加 Poll 间隔上限，防止慢速处理触发 Rebalance
max.poll.interval.ms=600000       # 两次 poll 最大间隔（默认 5min）

# 使用 Cooperative Sticky Rebalance
partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor

# 增加 poll 记录数，减少 poll 调用频率
max.poll.records=500

Consumer Group ID → 
  hash(groupId) % __consumer_offsets 分区数 → 目标分区 → 
  该分区 Leader 所在 Broker = Group Coordinator

旧架构（ZooKeeper 存储 Offset）：
  Consumer → ZK 直接写入 Offset
  缺点：ZK 不适合高频写入，OOM 风险

新架构（__consumer_offsets Topic 存储 Offset）：
  Consumer → Group Coordinator → __consumer_offsets Topic
  存储格式：Key = <GroupId, Topic, Partition>
           Value = <Offset, Metadata, Timestamp>
  压缩策略：compact（每个分区仅保留最新 Offset）

Consumer 心跳超时 → Group Coordinator 移除该 Consumer → 
  触发 Rebalance → 但 Coordinator 自身可能也在故障 → 
  Consumer 收到 NOT_COORDINATOR 错误 → 
  重新 FindCoordinator Request → 定位新 Coordinator → 
  重新 JoinGroup

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    metricsConfig:
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          name: kafka-metrics
          key: kafka-metrics-config.yml
    template:
      pod:
        metadata:
          annotations:
            prometheus.io/scrape: "true"
            prometheus.io/port: "9404"

lowercaseOutputName: true
lowercaseOutputLabelNames: true
rules:
  # Broker 核心指标
  - pattern: kafka.server<type=BrokerTopicMetrics, name=(BytesInPerSec|BytesOutPerSec)><>OneMinuteRate
    name: kafka_server_broker_topic_metrics_$1
    type: GAUGE
  - pattern: kafka.server<type=BrokerTopicMetrics, name=(MessagesInPerSec)><>OneMinuteRate
    name: kafka_server_broker_topic_metrics_$1
    type: GAUGE
  # 网络请求
  - pattern: kafka.network<type=RequestMetrics, name=(TotalTimeMs|RequestQueueTimeMs), request=(Produce|Fetch|FetchConsumer|FetchFollower)><>Count
    name: kafka_network_request_metrics_$2_$1
    labels:
      request: "$3"
  # ISR 与副本
  - pattern: kafka.server<type=ReplicaManager, name=(UnderReplicatedPartitions|UnderMinIsrPartitionCount)><>Value
    name: kafka_server_replica_manager_$1
  - pattern: kafka.server<type=ReplicaManager, name=(IsrShrinksPerSec|IsrExpandsPerSec)><>OneMinuteRate
    name: kafka_server_replica_manager_$1
  # Producer 请求延迟
  - pattern: kafka.network<type=RequestMetrics, name=TotalTimeMs, request=(Produce|FetchConsumer)><>(\w+)
    name: kafka_network_$3_$1_$2
    labels:
      request: "$3"

apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: kafka-cluster-metrics
  labels:
    release: prometheus
spec:
  selector:
    matchLabels:
      strimzi.io/cluster: my-cluster
      strimzi.io/kind: Kafka
  endpoints:
    - port: tcp-prometheus
      interval: 15s
      scrapeTimeout: 10s

# 吞吐优先
batch.size=131072                  # 批量大小 128KB
linger.ms=5                        # 等待 5ms 攒批
compression.type=zstd              # 压缩算法
buffer.memory=134217728            # 发送缓冲区 128MB
max.in.flight.requests.per.connection=5
acks=1                             # Leader 确认即可

# 可靠性优先
acks=all                           # 所有 ISR 确认
enable.idempotence=true            # 幂等生产者
max.in.flight.requests.per.connection=5
retries=2147483647                 # 无限重试
delivery.timeout.ms=120000         # 超时 2 分钟

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
spec:
  kafka:
    config:
      # 网络与 I/O
      num.network.threads: 8       # 网络线程数（处理网络请求）
      num.io.threads: 16           # I/O 线程数（处理磁盘 I/O）
      socket.send.buffer.bytes: 1048576
      socket.receive.buffer.bytes: 1048576
      socket.request.max.bytes: 104857600

      # Page Cache 与刷盘
      log.flush.interval.messages: 9223372036854775807  # 不基于消息数刷盘
      log.flush.interval.ms: 9223372036854775807        # 不基于时间刷盘（依赖 OS）

      # 日志段与索引
      log.segment.bytes: 1073741824                     # 1GB 段
      log.index.interval.bytes: 4096                   # 稀疏索引密度

      # 副本机制
      num.replica.fetchers: 4       # Follower 拉取线程数
      replica.fetch.max.bytes: 10485760
      replica.socket.timeout.ms: 30000

      # 压缩
      compression.type: producer

      # Leader 均衡
      auto.leader.rebalance.enable: true
      leader.imbalance.per.broker.percentage: 10
      leader.imbalance.check.interval.seconds: 300

    resources:
      requests:
        memory: 8Gi
        cpu: 2
      limits:
        memory: 16Gi
        cpu: 4
    jvmOptions:
      -Xms: 6g
      -Xmx: 6g
      -XX:+UseG1GC
      -XX:MaxGCPauseMillis: 20
      -XX:InitiatingHeapOccupancyPercent: 35
      -XX:+DisableExplicitGC

# 吞吐优先
fetch.min.bytes=1048576            # 每次 Fetch 最少拉取 1MB
fetch.max.wait.ms=500              # 等待 500ms 攒批
max.partition.fetch.bytes=10485760 # 每分区最大 10MB
max.poll.records=500               # 每次 Poll 最多 500 条

# 延迟优先
fetch.min.bytes=1                  # 有数据就返回
fetch.max.wait.ms=50               # 最多等待 50ms

# Rebalance 优化
session.timeout.ms=30000
heartbeat.interval.ms=3000
max.poll.interval.ms=600000
partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor

# Offset 管理
enable.auto.commit=false           # 手动提交，避免自动提交带来的重复消费

spec:
  kafka:
    storage:
      type: persistent-claim
      size: 500Gi
      class: fast-ssd
      deleteClaim: false

spec:
  kafka:
    storage:
      type: jbod
      volumes:
        - id: 0
          type: persistent-claim
          size: 200Gi
          class: fast-ssd
          deleteClaim: false
        - id: 1
          type: persistent-claim
          size: 200Gi
          class: fast-ssd
          deleteClaim: false

spec:
  kafka:
    config:
      # 启用 Tiered Storage
      remote.log.storage.system.enable: true
      remote.log.storage.manager.class.path: /opt/kafka/libs/remote-storage-*.jar
      remote.log.metadata.manager.impl.class: org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManager
      rlmm.config.remote.log.metadata.topic.replication.factor: 3

      # 本地保留
      local.retention.ms: 86400000  # 本地保留 1 天
      local.retention.bytes: -2     # 本地无大小限制

      # 远程保留
      retention.ms: 2592000000      # 远程保留 30 天

Active Segments（本地 SSD） <── 热数据，最近 1 天
Closed Segments（远程 S3/MinIO） <── 冷数据，最近 30 天
Deleted Segments <── 过期数据，由 Log Cleaner 线程清除

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
      - name: external
        port: 9094
        type: nodeport
        tls: true
        authentication:
          type: tls
        configuration:
          bootstrap:
            nodePort: 32094
          brokers:
            - broker: 0
              nodePort: 32095
            - broker: 1
              nodePort: 32096
            - broker: 2
              nodePort: 32097
    authorization:
      type: simple

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: scram-sha-512
  authorization:
    type: simple
    acls:
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operations:
          - Read
          - Write
          - Describe
        host: "*"
      - resource:
          type: group
          name: my-consumer-group
          patternType: literal
        operations:
          - Read
        host: "*"

bootstrap.servers=my-cluster-kafka-tls-bootstrap.kafka.svc:9093
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="my-user" \
  password="<password-from-secret>";
ssl.truststore.location=/path/to/ca.crt
ssl.truststore.type=PEM

plain (9092):   No TLS + No Auth → 仅测试环境
tls (9093):     TLS + No Auth → 加密但不认证
tls (9093):     TLS + SASL → 加密 + 用户认证（生产推荐）
external (9094): TLS + SASL → 外部客户端安全接入

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: app-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: tls
  authorization:
    type: simple
    acls:
      # 主题读写权限
      - resource:
          type: topic
          name: orders-*
          patternType: prefix
        operations:
          - Read
          - Write
          - Describe
          - DescribeConfigs
        host: "*"

      # 消费者组权限
      - resource:
          type: group
          name: order-processor
          patternType: literal
        operations:
          - Read
        host: "*"

      # 集群操作权限
      - resource:
          type: cluster
          name: kafka-cluster
        operations:
          - Describe
          - DescribeConfigs

      # 事务 ID 权限
      - resource:
          type: transactionalId
          name: app-tx-*
          patternType: prefix
        operations:
          - Write
          - Describe

spec:
  kafka:
    authorization:
      type: simple
      superUsers:
        - CN=entity-operator
        - CN=cruise-control
        - CN=my-mirror-maker-2

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

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  cruiseControl:
    config:
      goals: >-
        com.linkedin.kafka.cruisecontrol.analyzer.goals.RackAwareGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaCapacityGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.DiskCapacityGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkInboundCapacityGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkOutboundCapacityGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.CpuCapacityGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaDistributionGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.LeaderBytesInDistributionGoal        
      capacity.configFile: /opt/cruise-control/config/capacity.json
      hard.goals: >-
        com.linkedin.kafka.cruisecontrol.analyzer.goals.RackAwareGoal        
      anomaly.detection.goals: >-
        com.linkedin.kafka.cruisecontrol.analyzer.goals.RackAwareGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaCapacityGoal        
    resources:
      requests:
        memory: 1Gi
        cpu: 500m
      limits:
        memory: 2Gi
        cpu: 1

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaRebalance
metadata:
  name: my-rebalance
  labels:
    strimzi.io/cluster: my-cluster
spec:
  goals:
    - RackAwareGoal
    - ReplicaCapacityGoal
    - DiskCapacityGoal
    - NetworkInboundCapacityGoal
    - NetworkOutboundCapacityGoal
    - CpuCapacityGoal
  skipHardGoalCheck: false

KafkaRebalance CR 创建 → Cruise Control 收到请求 →
  Step 1: 收集集群指标（CPU / 磁盘 / 网络 / 副本数）→
  Step 2: 构建负载模型 → 
  Step 3: 运行优化器（Goal-based）→ 生成分区重分配方案 →
  Step 4: 更新 KafkaRebalance.status（ProposalReady）→
  用户审批方案 → Approve → 
  Step 5: 执行分区重分配（Kafka Admin API）→ 
  监控进度 → 完成更新 KafkaRebalance.status（Ready）

# 等待方案生成
kubectl wait kafkarebalance my-rebalance --for=condition=ProposalReady --timeout=300s

# 审批并执行
kubectl annotate kafkarebalance my-rebalance strimzi.io/rebalance=approve

# 停止正在执行的 Rebalance
kubectl annotate kafkarebalance my-rebalance strimzi.io/rebalance=stop

apiVersion: velero.io/v1
kind: Backup
metadata:
  name: kafka-full-backup
  namespace: velero
spec:
  includedNamespaces:
    - kafka
  labelSelector:
    matchLabels:
      strimzi.io/cluster: my-cluster
  includeClusterResources: true
  storageLocation: default
  snapshotVolumes: true
  hooks:
    resources:
      - name: pause-kafka
        pre:
          exec:
            container: kafka
            command:
              - /bin/bash
              - -c
              - |
                # 触发刷盘后暂停写入
                /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 \
                  --entity-type brokers --entity-default \
                  --alter --add-config unclean.leader.election.enable=false                

1. 准备恢复目标 K8s 集群（确保 StorageClass 兼容）
2. 创建 Velero Restore CR
   velero restore create --from-backup kafka-full-backup
3. Velero 恢复 PVC 和 PV 数据
4. Strimzi Operator 恢复 StatefulSet / ConfigMap / Secret
5. Kafka Pod 启动，加载恢复的数据
6. 验证集群状态
   - kubectl exec -it my-cluster-kafka-0 -- bin/kafka-topics.sh --list
   - 检查 ISR 和 Under Replicated Partitions
7. 恢复 Consumer 从最近的 Committed Offset 开始消费

[Active DC]                        [Standby DC]
Kafka Cluster ──MM2──→ Kafka Cluster
     │                              │
Clients Produce               Clients Consume
(主写入)                    (DR 切换后接管)

切换流程:
1. 停止 Active 侧 Producer
2. 确认 MM2 同步延迟 = 0（无数据丢失）
3. 停止 MM2 Connector
4. 切换 Client DNS / Bootstrap Server 到 Standby
5. 启动 Standby 侧 Producer
6. Consumer 从 Standby 集群继续消费

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-kraft-cluster
  annotations:
    strimzi.io/node-pools: enabled
    strimzi.io/kraft: enabled
spec:
  kafka:
    version: 3.7.0
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: 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
    storage:
      type: persistence-claim
      size: 200Gi

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaNodePool
metadata:
  name: controller
  labels:
    strimzi.io/cluster: my-kraft-cluster
spec:
  replicas: 3
  roles:
    - controller
  storage:
    type: persistent-claim
    size: 20Gi
  resources:
    requests:
      memory: 2Gi
      cpu: 1

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaNodePool
metadata:
  name: broker
  labels:
    strimzi.io/cluster: my-kraft-cluster
spec:
  replicas: 3
  roles:
    - broker
  storage:
    type: persistent-claim
    size: 200Gi
  resources:
    requests:
      memory: 8Gi
      cpu: 2

Controller 节点（3 节点 Quorum）:  majority = floor(3/2) + 1 = 2
  任意 1 个 Controller 故障 → Quorum 仍可用
  2 个 Controller 故障 → Quorum 不可用 → 元数据不可写

Controller 角色可选组合:
  - 纯 Controller 节点（推荐）: roles: [controller]
  - 纯 Broker 节点: roles: [broker]
  - 混合节点: roles: [controller, broker]

Kafka 3.5+ → 启用 ZK 到 KRaft 的迁移工具 → 
  Step 1: 部署 KRaft Controller Quorum → 
  Step 2: 配置 Broker 同时连接 ZK 和 KRaft → 
  Step 3: 元数据从 ZK 迁移至 KRaft → 
  Step 4: 验证 KRaft 中元数据完整性 → 
  Step 5: 从 Broker 配置中移除 ZK 连接 → 
  Step 6: 下线 ZooKeeper 集群

# 方式一：KafkaRebalance CRD
kubectl apply -f - <<EOF
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaRebalance
metadata:
  name: my-rebalance
  labels:
    strimzi.io/cluster: my-cluster
spec:
  goals:
    - RackAwareGoal
    - ReplicaCapacityGoal
    - DiskCapacityGoal
EOF

# 查看方案
kubectl describe kafkarebalance my-rebalance

# 批准执行
kubectl annotate kafkarebalance my-rebalance strimzi.io/rebalance=approve

# 1. 生成 Topic 列表（JSON 格式）
cat > topics-to-move.json <<EOF
{
  "topics": [
    {"topic": "orders"},
    {"topic": "payments"}
  ],
  "version": 1
}
EOF

# 2. 生成重分配方案
kubectl exec my-cluster-kafka-0 -- bin/kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --topics-to-move-json-file /tmp/topics-to-move.json \
  --broker-list "0,1,2,3" \
  --generate

# 3. 执行重分配
kubectl exec my-cluster-kafka-0 -- bin/kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file /tmp/reassignment.json \
  --execute

# 4. 验证进度
kubectl exec my-cluster-kafka-0 -- bin/kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file /tmp/reassignment.json \
  --verify

# Kafka CR 中配置限流
spec:
  kafka:
    config:
      # 副本间数据搬迁的限流（B/s），防止影响正常流量
      leader.replication.throttled.rate: 104857600    # 100MB/s
      follower.replication.throttled.rate: 104857600  # 100MB/s

1. Controller 将分区标记为 Reassigning
2. 目标 Broker 启动 ReplicaFetcher 线程
3. 从 Leader 拉取分区数据到目标 Broker
4. 目标 Broker 数据追平 ISR 后，加入 ISR
5. Controller 从原 Broker 移除旧副本
6. 更新分区元数据
7. 分区回到正常状态

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  version: 3.7.0
  replicas: 2
  bootstrapServers: my-cluster-kafka-bootstrap:9092
  config:
    group.id: connect-cluster
    offset.storage.topic: connect-cluster-offsets
    config.storage.topic: connect-cluster-configs
    status.storage.topic: connect-cluster-status
    config.storage.replication.factor: 3
    offset.storage.replication.factor: 3
    status.storage.replication.factor: 3
    key.converter: org.apache.kafka.connect.json.JsonConverter
    value.converter: org.apache.kafka.connect.json.JsonConverter
    key.converter.schemas.enable: false
    value.converter.schemas.enable: false
  build:
    output:
      type: docker
      image: registry.example.com/kafka-connect:latest
    plugins:
      - name: debezium-mysql
        artifacts:
          - type: zip
            url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/2.6.0.Final/debezium-connector-mysql-2.6.0.Final-plugin.zip
            sha512sum: <sha512-checksum>

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: mysql-source-connector
  labels:
    strimzi.io/cluster: my-connect-cluster
spec:
  class: io.debezium.connector.mysql.MySqlConnector
  tasksMax: 2
  config:
    database.hostname: mysql.example.com
    database.port: 3306
    database.user: debezium
    database.password: ${secret:mysql-secret:password}
    database.server.id: 1
    topic.prefix: cdc
    database.include.list: orders,payments
    table.include.list: orders.orders,payments.transactions
    snapshot.mode: initial
    include.schema.changes: true
    transforms: Reroute,ExtractField
    transforms.Reroute.type: io.debezium.transforms.ByLogicalTableRouter
    transforms.Reroute.topic.regex: cdc.orders.orders
    transforms.Reroute.topic.replacement: orders-events

Connect Worker 启动 → 
  读取 config.storage.topic 中的 Connector 配置 → 
  Worker 分配 Connector → Tasks 数量 → 
  每个 Task 独立运行（并行度 = tasksMax）→ 
  Source Task: poll() → 写入 Kafka → commit() offset
  Sink Task: 消费 Kafka → put() 到外部系统 → flush() → commit() offset

# Schema Registry 以独立 Deployment 部署
apiVersion: apps/v1
kind: Deployment
metadata:
  name: schema-registry
  labels:
    app: schema-registry
spec:
  replicas: 2
  selector:
    matchLabels:
      app: schema-registry
  template:
    metadata:
      labels:
        app: schema-registry
    spec:
      containers:
        - name: schema-registry
          image: confluentinc/cp-schema-registry:7.6.0
          env:
            - name: SCHEMA_REGISTRY_HOST_NAME
              valueFrom:
                fieldRef:
                  fieldPath: status.podIP
            - name: SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS
              value: PLAINTEXT://my-cluster-kafka-bootstrap:9092
            - name: SCHEMA_REGISTRY_KAFKASTORE_TOPIC
              value: _schemas
            - name: SCHEMA_REGISTRY_KAFKASTORE_TOPIC_REPLICATION_FACTOR
              value: "3"
            - name: SCHEMA_REGISTRY_KAFKASTORE_SECURITY_PROTOCOL
              value: SSL
            - name: SCHEMA_REGISTRY_LISTENERS
              value: http://0.0.0.0:8081
            - name: SCHEMA_REGISTRY_SCHEMA_COMPATIBILITY_LEVEL
              value: backward
            - name: SCHEMA_REGISTRY_ACCESS_CONTROL_ENABLED
              value: "true"
          ports:
            - containerPort: 8081
          resources:
            requests:
              memory: 1Gi
              cpu: 500m
            limits:
              memory: 2Gi
              cpu: 1

# Producer 配置
key.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
value.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
schema.registry.url=http://schema-registry.kafka.svc:8081
auto.register.schemas=true

# Consumer 配置
key.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer
value.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer
schema.registry.url=http://schema-registry.kafka.svc:8081
specific.avro.reader=true

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: scram-sha-512
  authorization:
    type: simple
  quotas:
    producerByteRate: 10485760    # 10MB/s 生产限流
    consumerByteRate: 20971520    # 20MB/s 消费限流
    requestPercentage: 50         # 50% I/O 线程占用

<user, client-id> Quota → 最高优先级
<user> Quota → 次优先级
<client-id> Quota → 再次优先级
<user, client-id> default → 默认 Quota

# 设置 User 级别生产带宽限流
kubectl exec my-cluster-kafka-0 -- bin/kafka-configs.sh \
  --bootstrap-server localhost:9092 \
  --entity-type users --entity-name my-user \
  --alter --add-config producer_byte_rate=10485760

# 设置 Client ID 级别请求率限流
kubectl exec my-cluster-kafka-0 -- bin/kafka-configs.sh \
  --bootstrap-server localhost:9092 \
  --entity-type clients --entity-name app-producer \
  --alter --add-config request_percentage=30

# 查看当前 Quota 配置
kubectl exec my-cluster-kafka-0 -- bin/kafka-configs.sh \
  --bootstrap-server localhost:9092 \
  --entity-type users --entity-name my-user --describe

# 按 Topic 分组查看各分区的写入字节速率
kubectl exec my-cluster-kafka-0 -- bin/kafka-run-class.sh \
  kafka.tools.JmxTool \
  --object-name kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec,topic=orders \
  --reporting-interval 1000

# 通过 kafka-log-dirs.sh 查看各分区磁盘占用
kubectl exec my-cluster-kafka-0 -- bin/kafka-log-dirs.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --topic-list orders

# 查看 Topic 分区 Leader 分布
kubectl exec my-cluster-kafka-0 -- bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --describe --topic orders

# 单分区写入速率（检测热点）
topk(10, sum(rate(kafka_server_broker_topic_metrics_bytesinpersec[5m])) by (topic, partition))

# 分区大小 Top 10
topk(10, sum(kafka_log_log_size) by (topic, partition))

# 单 Broker 上 Leader 分区数
count(kafka_server_replicamanager_leadercount) by (pod)

spec:
  kafka:
    config:
      auto.leader.rebalance.enable: true
      leader.imbalance.per.broker.percentage: 10     # 不平衡阈值 10%
      leader.imbalance.check.interval.seconds: 300    # 每 5 分钟检查

// 自定义 Partitioner：热点 Key 拆分为多个子 Key
public class ConsistentHashPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        String keyStr = (String) key;
        if (isHotKey(keyStr)) {
            // 热点 Key 附加随机后缀散列
            keyStr = keyStr + "-" + ThreadLocalRandom.current().nextInt(10);
        }
        return Math.abs(keyStr.hashCode()) % cluster.partitionCountForTopic(topic);
    }
}

Producer 初始化 → 向 Broker 发送 InitProducerIdRequest → 
  Broker 分配 Producer ID（PID，单调递增的唯一 ID）→ 
  Producer 为每个 <PID, TopicPartition> 维护 Sequence Number（从 0 开始）→ 
  发送消息时附加 <PID, Epoch, SequenceNumber> → 
  Broker 端维护每个 <PID, TopicPartition> 已确认的最大 SequenceNumber → 
  收到消息：
    1. Sequence == lastSeq+1 → 接受，更新 lastSeq
    2. Sequence <= lastSeq → 重复消息，丢弃，返回 ACK（不重复写入）
    3. Sequence > lastSeq+1 → 乱序，抛出 OutOfOrderSequenceException

enable.idempotence=true                    # 启用幂等
acks=all                                    # 必须设为 all
max.in.flight.requests.per.connection=5     # ≤ 5
retries=2147483647                          # 无限重试

Producer ID 分配：
  - ZK 模式：存储在 /latest_producer_id_block ZNode 中
  - KRaft 模式：存储在 Controller 元数据日志中

Sequence Number 维护：
  - 每个 Broker 的内存中维护 Snapshot 文件
  - 保存最近 N 个 Producer 的 <PID, Epoch, SequenceNumber> 状态
  - Snapshot 定期写入磁盘，Broker 启动时恢复

幂等生产者 → 单分区 / 单 Session 内的去重保证
事务生产者 → 
  跨分区原子写入 +
  Consumer-Transform-Produce 循环中的 Exactly-Once +
  跨 Session 去重（基于 TransactionalId 而非 PID）

症状：Pod 状态 CrashLoopBackOff / Pending

排查步骤：
  1. kubectl describe pod my-cluster-kafka-0
  2. kubectl logs my-cluster-kafka-0 -c kafka
  3. 常见原因：
     - PVC 无法挂载（StorageClass 不兼容 / PV 未创建）
     - 内存不足（JVM OOMKilled）
     - ZooKeeper 连接失败（ZK Pod 未就绪 / DNS 不可解析）
     - 证书过期（TLS 证书校验失败）
     - config 中的 broker.id 冲突

  4. 解决：
     - kubectl describe pvc data-my-cluster-kafka-0
     - 检查 JVM heap 配置：-Xms 与 -Xmx 与容器 memory limit 匹配
     - 验证 ZK 服务：kubectl exec my-cluster-zk-0 -- bin/zookeeper-shell.sh localhost:2181 stat

症状：kafka_server_replica_manager_underreplicated_partitions > 0 持续超过 5 分钟

排查步骤：
  1. kubectl exec my-cluster-kafka-0 -- bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --under-replicated-partitions
  2. 检查受影响 Broker 的状态
  3. 常见原因：
     - 某一 Broker 宕机或 OOM 重启
     - 网络分区导致 Broker 间通信中断
     - 磁盘 I/O 瓶颈导致 Follower 追不上 Leader
     - Replica Fetcher 线程不足

  4. 解决：
     - 恢复故障 Broker
     - 调大 num.replica.fetchers（默认 1 → 4）
     - 检查磁盘 I/O 指标，必要时迁移分区

症状：Consumer Group Lag 单调递增

排查步骤：
  1. kubectl exec my-cluster-kafka-0 -- bin/kafka-consumer-groups.sh \
       --bootstrap-server localhost:9092 --group order-processor --describe
  2. 对比 Produce 速率与 Consumer 消费速率
  3. 常见原因：
     - Consumer 实例数量不足
     - Consumer 处理逻辑慢（外部 API 调用 / DB 查询慢）
     - Consumer 频繁 Rebalance 导致消费暂停
     - 某些分区 Lag 高但其他分区 Lag 低（热点分区）

  4. 解决：
     - 增加 Consumer 实例（不超过分区数）
     - 优化 Consumer 处理逻辑
     - 启用 Cooperative Rebalance 减少 Rebalance 暂停时间
     - 增加 max.poll.records 减少 Poll 频率
     - 热点分区通过修改 Key 策略处理

症状：Producer 端报 TimeoutException / request timed out

排查步骤：
  1. 检查 Broker 端 produce 请求延迟
     kafka_network_request_metrics_totaltimems_produce
  2. 检查 Broker GC 暂停时间
     jvm_gc_pause_seconds
  3. 检查磁盘 I/O Util
  4. 常见原因：
     - acks=all 且 ISR 中有 Follower 延迟过高
     - Broker 磁盘 I/O 饱和
     - 网络带宽打满
     - JVM GC 频繁导致 Broker 无响应

  5. 解决：
     - Producer acks 从 all 降为 1（非关键业务）
     - 调整 JVM GC 参数减少暂停
     - 扩容 Broker 或增加网络带宽

症状：KafkaTopic CR status.conditions 中出现 NotReady

排查步骤：
  1. kubectl describe kafkatopic my-topic
  2. kubectl logs deploy/my-cluster-entity-operator -c topic-operator
  3. 常见原因：
     - Topic Operator 无法连接 Kafka Admin API
     - 分区数或副本因子超过集群容量
     - Topic 名称不符合规范
     - Kafka Broker 端 ACL 限制 Entity Operator 操作

  4. 解决：
     - 验证 Entity Operator 的 TLS 证书未过期
     - 检查 KafkaTopic.spec 参数合法性
     - 确认 Entity Operator 在 superUsers 列表中

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 = XQOPEN $value XQCLOSE"

      - 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 版本的兼容性区间为 ±1 大版本