Kafka 分布式消息队列

生产者（Producer）→ 主题 / 分区主副本（Leader）→ 从副本（Follower）
                    ↓
              消费者组（Consumer Group）A
              消费者组（Consumer Group）B

bin/kafka-topics.sh --bootstrap-server broker-1:9092 \
  --create --if-not-exists --topic orders \
  --partitions 12 --replication-factor 3 \
  --config min.insync.replicas=2 \
  --config retention.ms=604800000 \
  --config cleanup.policy=delete

bin/kafka-configs.sh --bootstrap-server broker-1:9092 \
  --entity-type topics --entity-name orders --alter \
  --add-config segment.bytes=1073741824,compression.type=producer

保留数据量 ≈ 峰值写入字节/天 × 保留天数 ÷ 实测压缩比
物理存储量 ≈ 保留数据量 × 副本因子 × (1 + 索引、重分配和增长余量)
单 Broker 容量 ≈ 物理存储量 ÷ Broker 数，再预留副本分布不均衡空间

Follower 进入 ISR 条件：
  - Follower 持续从 Leader 拉取数据，并在 replica.lag.time.max.ms 时间窗口内保持可追赶状态
  - Kafka 以时间窗口判断副本是否落后；不要把早期版本的 replica.lag.max.messages 当作当前集群的调优依据

Follower 踢出 ISR 条件：
  - 超过 replica.lag.time.max.ms 未能保持同步，Leader 将其移出 ISR

Leader 维护 ISR 集合 →
  Follower 持续 Fetch 并追赶日志末端位点（LEO）→
  超过 replica.lag.time.max.ms 仍无法保持同步 →
  从 ISR 移除并提交分区状态 →
  Controller 将状态变更传播给集群 →
  Follower 追平后重新加入 ISR

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

KRaft Controller Quorum 复制元数据日志 →
  多数派选出活跃 Controller →
  活跃 Controller 处理 Broker 注册、分区状态和 Leader 变更 →
  活跃 Controller 故障 → 多数派重新选举 → 新 Controller 从元数据日志继续处理

旧 ZooKeeper 集群：Broker 通过临时节点竞争 Controller，
Controller 切换依赖 ZooKeeper 会话与 Watch 通知。

Broker 不可用或受控下线 → Controller 收到状态变更 →
遍历受影响的 Leader 分区 →
从可用 ISR 中选择新 Leader →
提交新的分区状态与 Leader Epoch →
向 Broker 和客户端发布元数据更新 →
副本继续追赶并恢复正常 ISR

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 配置稳定的 transactional.id → initTransactions() →
  事务协调器分配或恢复 Producer ID（PID）与 Epoch →
  beginTransaction() →
  发送消息（关联事务）→
  消息写入分区但不向消费者可见 →
  sendOffsetsToTransaction() →
  commitTransaction() / abortTransaction() →
  Broker 写入 Commit/Abort Marker →
  消费者读取到 Marker 后过滤事务消息

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

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

# 按主题覆盖 Broker 默认值；只影响该主题后续写入的批次
bin/kafka-configs.sh --bootstrap-server broker-1:9092 \
  --entity-type topics --entity-name orders --alter \
  --add-config compression.type=producer

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

# 事件流：时间和大小任一条件先满足即触发删除
bin/kafka-configs.sh --bootstrap-server broker-1:9092 \
  --entity-type topics --entity-name events --alter \
  --add-config cleanup.policy=delete,retention.ms=604800000,retention.bytes=107374182400

# 状态主题：保留每个 Key 的最新值；删除标记需保留足够久以便慢消费者看到
bin/kafka-configs.sh --bootstrap-server broker-1:9092 \
  --entity-type topics --entity-name user-state --alter \
  --add-config cleanup.policy=compact,min.cleanable.dirty.ratio=0.5,delete.retention.ms=86400000

日志清理线程定期扫描 → 选择脏比率较高的日志段 →
  逐段读取 → 构建位点映射（每个 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 配置：
  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

export KAFKA_OPTS='-javaagent:/opt/jmx/jmx_prometheus_javaagent.jar=9404:/opt/jmx/kafka-metrics.yml'
bin/kafka-server-start.sh config/server.properties

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"

scrape_configs:
  - job_name: kafka
    static_configs:
      - targets:
          - broker-1.example.com:9404
          - broker-2.example.com:9404
          - broker-3.example.com:9404

# 吞吐优先
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 分钟

# server.properties；数值必须根据 CPU、磁盘、网络与压测结果确定
num.network.threads=8
num.io.threads=16
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600

# 日志段与副本拉取
log.segment.bytes=1073741824
log.index.interval.bytes=4096
num.replica.fetchers=4
replica.fetch.max.bytes=10485760
compression.type=producer

# Leader 均衡应避免与大规模重分配同时进行
auto.leader.rebalance.enable=true
leader.imbalance.per.broker.percentage=10
leader.imbalance.check.interval.seconds=300

# 吞吐优先
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           # 手动提交，避免自动提交带来的重复消费

# server.properties
log.dirs=/data-1/kafka-logs,/data-2/kafka-logs
num.recovery.threads.per.data.dir=2
log.segment.bytes=1073741824

活跃日志段：保留在本地磁盘，服务低延迟尾部读取
已关闭日志段：由 RemoteStorageManager 上传到远端存储
本地保留到期：本地副本可删除，历史读取转为远端读取
主题总保留到期：按主题策略删除本地与远端历史数据

bin/kafka-configs.sh --bootstrap-server broker-1:9092 \
  --entity-type topics --entity-name audit-events --alter \
  --add-config remote.storage.enable=true,local.retention.ms=86400000,retention.ms=2592000000

# server.properties
listeners=SASL_SSL://0.0.0.0:9093
advertised.listeners=SASL_SSL://broker-1.example.com:9093
listener.security.protocol.map=SASL_SSL:SASL_SSL
inter.broker.listener.name=SASL_SSL

ssl.keystore.location=/etc/kafka/secrets/broker.keystore.p12
ssl.keystore.password=<keystore-password>
ssl.key.password=<key-password>
ssl.truststore.location=/etc/kafka/secrets/broker.truststore.p12
ssl.truststore.password=<truststore-password>
ssl.client.auth=none

sasl.enabled.mechanisms=SCRAM-SHA-512
listener.name.sasl_ssl.scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required;

bootstrap.servers=broker-1.example.com:9093,broker-2.example.com: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-store>";
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:        仍使用 TLS + SASL，并让 advertised.listeners 对外可解析

# 生产者：只写 orders- 前缀主题
bin/kafka-acls.sh --bootstrap-server broker-1:9093 --command-config admin.properties \
  --add --allow-principal User:order-service \
  --operation Write --operation Describe \
  --topic orders- --resource-pattern-type prefixed

# 消费者：读取订单主题，并只访问自己的消费者组
bin/kafka-acls.sh --bootstrap-server broker-1:9093 --command-config admin.properties \
  --add --allow-principal User:order-service \
  --operation Read --topic orders- --resource-pattern-type prefixed \
  --group order-processor

# 事务生产者：允许使用自己的 transactional.id 前缀
bin/kafka-acls.sh --bootstrap-server broker-1:9093 --command-config admin.properties \
  --add --allow-principal User:order-service \
  --operation Write --transactional-id order-tx- --resource-pattern-type prefixed

# server.properties
authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
super.users=User:kafka-admin

采集指标与节点容量 → 选择目标集 → 生成 dry-run 方案 →
人工检查迁移量、硬约束和风险 → 设置限速并执行 →
观察副本同步、磁盘与客户端延迟 → 验证完成

# 先查看集群状态与容量模型；实际 URL、认证和目标名以部署版本为准
curl -sS 'https://cruise-control.example.com/kafkacruisecontrol/kafka_cluster_state'

# 生成方案，只查看结果，不执行
curl -sS -X POST \
  'https://cruise-control.example.com/kafkacruisecontrol/rebalance?dryrun=true'

主集群 ──异步复制──> 备用集群
   ↑                      ↑
生产者与消费者       灾难时切换客户端入口

# Controller 节点：省略 TLS 与鉴权配置，只展示角色和仲裁关系
process.roles=controller
node.id=1
controller.quorum.voters=1@controller-1.example.com:9093,2@controller-2.example.com:9093,3@controller-3.example.com:9093
controller.listener.names=CONTROLLER
listeners=CONTROLLER://0.0.0.0:9093
inter.broker.listener.name=INTERNAL
listener.security.protocol.map=CONTROLLER:SSL,INTERNAL:SSL
metadata.log.dir=/var/lib/kafka/metadata

# Broker 节点
process.roles=broker
node.id=101
controller.quorum.voters=1@controller-1.example.com:9093,2@controller-2.example.com:9093,3@controller-3.example.com:9093
controller.listener.names=CONTROLLER
listeners=INTERNAL://0.0.0.0:9092
advertised.listeners=INTERNAL://broker-101.example.com:9092
inter.broker.listener.name=INTERNAL
listener.security.protocol.map=CONTROLLER:SSL,INTERNAL:SSL
log.dirs=/var/lib/kafka/data

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

# 2. 生成候选方案；命令会打印建议的 reassignment JSON，需人工保存并审阅
bin/kafka-reassign-partitions.sh \
  --bootstrap-server broker-1.example.com:9092 \
  --topics-to-move-json-file topics-to-move.json \
  --broker-list "0,1,2,3" \
  --generate

# 3. 将审阅后的方案保存为 reassignment.json，再限速执行
bin/kafka-reassign-partitions.sh \
  --bootstrap-server broker-1.example.com:9092 \
  --reassignment-json-file reassignment.json \
  --execute \
  --throttle 104857600

# 4. 验证进度
bin/kafka-reassign-partitions.sh \
  --bootstrap-server broker-1.example.com:9092 \
  --reassignment-json-file reassignment.json \
  --verify

1. Controller 记录新的目标副本集合。
2. 新副本从当前 Leader 拉取数据并追赶日志。
3. 新副本追平后，Controller 切换副本集合并移除旧副本。
4. 如有 Leader 迁移，客户端会经历一次元数据刷新和短暂的角色切换。

# connect-distributed.properties
bootstrap.servers=broker-1.example.com:9093,broker-2.example.com:9093
group.id=connect-orders-prod

config.storage.topic=_connect-orders-configs
offset.storage.topic=_connect-orders-offsets
status.storage.topic=_connect-orders-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
plugin.path=/opt/kafka-connect/plugins

curl -sS -X POST 'https://connect.example.com/connectors' \
  -H 'Content-Type: application/json' \
  --data @- <<'JSON'
{
  "name": "mysql-orders-source",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "database.hostname": "mysql.example.com",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "${file:/etc/kafka-connect/secrets/mysql.properties:password}",
    "topic.prefix": "cdc",
    "database.include.list": "orders",
    "table.include.list": "orders.orders",
    "snapshot.mode": "initial"
  }
}
JSON

Worker 读取配置 Topic → 任务在 Worker 间分配 →
Source Task 从外部系统读取并写入 Kafka，提交源端进度 →
Sink Task 消费 Kafka、写入目标系统、确认后再提交 Kafka 位点

# schema-registry.properties；安全相关参数按实际环境补齐
listeners=http://0.0.0.0:8081
kafkastore.bootstrap.servers=SSL://broker-1.example.com:9093,SSL://broker-2.example.com:9093
kafkastore.topic=_schemas
kafkastore.topic.replication.factor=3
schema.compatibility.level=BACKWARD

# 以下是 Confluent 兼容客户端的配置示例
# Producer
key.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
value.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
schema.registry.url=https://schema-registry.example.com
auto.register.schemas=false

# Consumer
key.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer
value.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer
schema.registry.url=https://schema-registry.example.com
specific.avro.reader=true

# 设置 User 级别的生产、消费带宽限流
bin/kafka-configs.sh \
  --bootstrap-server broker-1.example.com:9092 \
  --entity-type users --entity-name my-user \
  --alter --add-config producer_byte_rate=10485760,consumer_byte_rate=20971520

# 设置 Client ID 级别请求率限流
bin/kafka-configs.sh \
  --bootstrap-server broker-1.example.com:9092 \
  --entity-type clients --entity-name app-producer \
  --alter --add-config request_percentage=30

# 查看当前 Quota 配置
bin/kafka-configs.sh \
  --bootstrap-server broker-1.example.com:9092 \
  --entity-type users --entity-name my-user --describe

# JMX 可查看 Topic 总吞吐；分区级吞吐通常需由 exporter、客户端或链路监控补充
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 查看各分区磁盘占用
bin/kafka-log-dirs.sh \
  --bootstrap-server broker-1.example.com:9092 \
  --describe \
  --topic-list orders

# 查看 Topic 分区 Leader 分布
bin/kafka-topics.sh \
  --bootstrap-server broker-1.example.com:9092 \
  --describe --topic orders

# 以下假设 exporter 暴露了带 topic、partition 标签的分区字节计数器；实际指标名按 exporter 调整
topk(10, sum by (topic, partition) (rate(kafka_partition_bytes_in_total[5m])))

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

# 单 Broker 上 Leader 分区数
sum by (instance) (kafka_server_replicamanager_leadercount)

Producer 初始化 → 向事务协调器请求 Producer ID（PID）和 Epoch →
  Producer 为每个 <PID, TopicPartition> 维护 Sequence Number（从 0 开始）→
  发送消息时附加 <PID, Epoch, SequenceNumber> →
  Leader Broker 为每个 <PID, TopicPartition> 维护已确认的序列状态 →
  收到消息：
    1. Sequence 与预期值连续 → 接受并更新状态
    2. Sequence 已处理过 → 识别为重试，不重复写入
    3. Sequence 跳号或 Epoch 过期 → 拒绝该请求，客户端需要处理乱序或被 fencing 的情况

enable.idempotence=true                    # 启用幂等
acks=all                                    # 幂等性要求所有 ISR 确认
max.in.flight.requests.per.connection=5     # 启用幂等性时不能大于 5

事务协调器负责分配和管理 PID / Epoch。
分区 Leader 将最近的生产者序列状态保存在日志相关的 producer state snapshot 中；
Broker 重启后可从日志与快照恢复，避免把历史重试当成新消息。

幂等生产者 → 单分区 / 单 Session 内的去重保证
事务生产者 →
  跨分区原子写入 +
  Kafka read-process-write 链路中的 Exactly-Once +
  使用稳定 transactional.id fencing 旧实例

症状：客户端报 Connection to node failed、NotLeaderOrFollower、
      TimeoutException，或集群中出现 Offline Partition。

排查重点：
  1. 确认受影响的是单个 Broker、单个 Listener，还是整个集群。
  2. 查看 Broker 日志、JVM/GC、磁盘空间和进程监听端口。
  3. 核对 listeners、advertised.listeners、DNS、负载均衡和防火墙配置。
  4. TLS/SASL 集群还需核对证书有效期、客户端协议与 ACL。

处理原则：
  - 先恢复网络、进程或磁盘等根因，再确认 Controller 已完成 Leader 迁移。
  - 不要把 advertised.listeners 配成客户端无法解析的内网地址。

bin/kafka-topics.sh --bootstrap-server broker-1:9092 \
  --describe --under-replicated-partitions

bin/kafka-consumer-groups.sh --bootstrap-server broker-1:9092 \
  --group order-processor --describe

bin/kafka-log-dirs.sh --bootstrap-server broker-1:9092 --describe