Apache Kafka 分割區生命週期管理與日誌分析深度實戰指南

{
  "version": 1,
  "topics": [
    {"topic": "orders"},
    {"topic": "payments"},
    {"topic": "inventory-updates"},
    {"topic": "customer-events"}
  ]
}

# 使用 kafka-reassign-partitions.sh 工具生成分割區重新分配計畫
# 這個指令會分析當前的分割區分布狀態，並產生建議的重新分配方案
kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --topics-to-move-json-file topics.json \
  --broker-list 5,6,7,8 \
  --generate

# 參數詳細說明：
# --bootstrap-server localhost:9092
#   指定 Kafka 叢集的連線位址
#   可以是叢集中任意一個 Broker 的位址
#   工具會自動發現其他 Broker 節點
#
# --topics-to-move-json-file topics.json
#   指定包含要重新分配 Topic 清單的 JSON 設定檔
#   檔案路徑可以是相對路徑或絕對路徑
#   檔案內容必須符合 Kafka 定義的 JSON Schema
#
# --broker-list 5,6,7,8
#   指定目標 Broker ID 清單，以逗號分隔
#   分割區只會被分配到這個清單中的 Broker 上
#   Broker ID 必須是叢集中實際存在的節點
#   順序不影響分配結果，工具會自動最佳化
#
# --generate
#   生成重新分配計畫但不實際執行
#   這是一個安全的操作，不會對叢集造成任何變更
#   輸出可以被檢視、修改，然後用於後續的執行步驟

{
  "version": 1,
  "partitions": [
    {
      "topic": "orders",
      "partition": 0,
      "replicas": [5, 6, 7],
      "log_dirs": ["any", "any", "any"]
    },
    {
      "topic": "orders",
      "partition": 1,
      "replicas": [6, 7, 8],
      "log_dirs": ["any", "any", "any"]
    },
    {
      "topic": "orders",
      "partition": 2,
      "replicas": [7, 8, 5],
      "log_dirs": ["any", "any", "any"]
    },
    {
      "topic": "orders",
      "partition": 3,
      "replicas": [8, 5, 6],
      "log_dirs": ["any", "any", "any"]
    },
    {
      "topic": "payments",
      "partition": 0,
      "replicas": [6, 7, 8],
      "log_dirs": ["any", "any", "any"]
    },
    {
      "topic": "payments",
      "partition": 1,
      "replicas": [7, 8, 5],
      "log_dirs": ["any", "any", "any"]
    },
    {
      "topic": "payments",
      "partition": 2,
      "replicas": [8, 5, 6],
      "log_dirs": ["any", "any", "any"]
    },
    {
      "topic": "payments",
      "partition": 3,
      "replicas": [5, 6, 7],
      "log_dirs": ["any", "any", "any"]
    }
  ]
}

# 執行分割區重新分配計畫
# 這個操作會開始實際的資料遷移程序，對叢集產生實質影響
kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file cluster-reassignment-plan.json \
  --execute

# 參數詳細說明：
# --bootstrap-server localhost:9092
#   指定 Kafka 叢集的連線位址
#   建議使用叢集的負載平衡位址或多個 Broker 位址
#   以提升連線的可靠性
#
# --reassignment-json-file cluster-reassignment-plan.json
#   指定包含重新分配計畫的 JSON 檔案
#   這個檔案應該是經過仔細檢視與調整的最終版本
#   執行前建議再次確認檔案內容的正確性
#
# --execute
#   實際執行重新分配操作
#   這個參數會觸發資料遷移程序
#   執行後無法簡單地取消，需要使用 --cancel 指令
#
# 執行輸出範例：
# Current partition replica assignment
# 
# {"version":1,"partitions":[...]}
# 
# Save this to use as the --reassignment-json-file option during rollback
# Successfully started reassignment of partitions.
#
# 輸出說明：
# 第一部分顯示當前的分割區分配狀態，應該保存起來以便回滾
# 第二部分確認重新分配已成功啟動

# 驗證分割區重新分配的執行狀態
# 這個指令會檢查每個分割區的遷移進度並報告結果
kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file cluster-reassignment-plan.json \
  --verify

# 參數說明：
# --bootstrap-server localhost:9092
#   指定 Kafka 叢集的連線位址
#
# --reassignment-json-file cluster-reassignment-plan.json
#   指定原始的重新分配計畫檔案
#   驗證過程需要這個檔案來了解目標狀態
#
# --verify
#   檢查重新分配的完成狀態
#   這是一個唯讀操作，不會對叢集產生任何變更
#
# 輸出範例（進行中）：
# Status of partition reassignment:
# Reassignment of partition orders-0 is still in progress.
# Reassignment of partition orders-1 is still in progress.
# Reassignment of partition payments-0 completed successfully.
# Reassignment of partition payments-1 completed successfully.
#
# 輸出範例（全部完成）：
# Status of partition reassignment:
# Reassignment of partition orders-0 completed successfully.
# Reassignment of partition orders-1 completed successfully.
# Reassignment of partition payments-0 completed successfully.
# Reassignment of partition payments-1 completed successfully.
#
# 輸出範例（發生錯誤）：
# Status of partition reassignment:
# Reassignment of partition orders-0 completed successfully.
# Reassignment of partition orders-1 failed.
# Reassignment of partition payments-0 is still in progress.
# Reassignment of partition payments-1 completed successfully.

#!/bin/bash

# Kafka 分割區重新分配狀態監控腳本
# 此腳本會定期檢查重新分配狀態，直到所有分割區完成遷移或發生失敗
# 適用於自動化維運場景，可整合到 CI/CD 流程或監控系統

# 設定 Kafka 叢集連線位址
# 建議使用叢集的負載平衡位址以提升可靠性
BOOTSTRAP_SERVER="kafka-lb.example.com:9092"

# 設定重新分配計畫檔案路徑
# 這應該是執行 --execute 時使用的同一個檔案
REASSIGNMENT_FILE="/opt/kafka/plans/cluster-reassignment-plan.json"

# 設定檢查間隔時間（秒）
# 對於大型分割區，建議設定較長的間隔以減少系統負擔
CHECK_INTERVAL=60

# 設定日誌檔案路徑
# 所有的檢查結果都會記錄到這個檔案中
LOG_FILE="/var/log/kafka/reassignment-monitor-$(date +%Y%m%d_%H%M%S).log"

# 設定通知電子郵件位址（可選）
# 當遷移完成或失敗時會發送通知
NOTIFICATION_EMAIL="[email protected]"

# 函式：記錄訊息到日誌檔案與標準輸出
log_message() {
    local timestamp=$(date '+%Y-%m-%d %H:%M:%S')
    local message="[$timestamp] $1"
    echo "$message" | tee -a "$LOG_FILE"
}

# 函式：發送通知郵件
send_notification() {
    local subject="$1"
    local body="$2"
    if [ -n "$NOTIFICATION_EMAIL" ]; then
        echo "$body" | mail -s "$subject" "$NOTIFICATION_EMAIL"
    fi
}

# 檢查必要的工具是否存在
if ! command -v kafka-reassign-partitions.sh &> /dev/null; then
    log_message "錯誤：找不到 kafka-reassign-partitions.sh 工具"
    exit 1
fi

# 檢查重新分配計畫檔案是否存在
if [ ! -f "$REASSIGNMENT_FILE" ]; then
    log_message "錯誤：重新分配計畫檔案不存在：$REASSIGNMENT_FILE"
    exit 1
fi

log_message "開始監控分割區重新分配狀態"
log_message "Kafka 叢集：$BOOTSTRAP_SERVER"
log_message "計畫檔案：$REASSIGNMENT_FILE"
log_message "檢查間隔：$CHECK_INTERVAL 秒"
log_message "----------------------------------------"

# 初始化計數器
check_count=0
total_partitions=0
completed_partitions=0

# 持續檢查直到所有分割區完成重新分配
while true; do
    check_count=$((check_count + 1))
    log_message "執行第 $check_count 次狀態檢查"

    # 執行驗證指令並擷取輸出結果
    RESULT=$(kafka-reassign-partitions.sh \
        --bootstrap-server "$BOOTSTRAP_SERVER" \
        --reassignment-json-file "$REASSIGNMENT_FILE" \
        --verify 2>&1)

    # 將完整結果寫入日誌
    echo "$RESULT" >> "$LOG_FILE"

    # 分析結果統計
    total_partitions=$(echo "$RESULT" | grep -c "Reassignment of partition")
    completed_partitions=$(echo "$RESULT" | grep -c "completed successfully")
    in_progress=$(echo "$RESULT" | grep -c "still in progress")
    failed_partitions=$(echo "$RESULT" | grep -c "failed")

    log_message "狀態統計："
    log_message "  總分割區數：$total_partitions"
    log_message "  已完成：$completed_partitions"
    log_message "  進行中：$in_progress"
    log_message "  失敗：$failed_partitions"

    # 檢查是否有失敗的重新分配
    if [ $failed_partitions -gt 0 ]; then
        log_message "警告：發現 $failed_partitions 個分割區重新分配失敗！"
        log_message "失敗的分割區："
        echo "$RESULT" | grep "failed" | tee -a "$LOG_FILE"

        # 發送失敗通知
        send_notification \
            "Kafka 分割區重新分配失敗" \
            "檢測到 $failed_partitions 個分割區重新分配失敗。詳細資訊請查看日誌檔案：$LOG_FILE"

        log_message "----------------------------------------"
        log_message "重新分配程序異常結束"
        exit 1
    fi

    # 檢查是否還有進行中的重新分配
    if [ $in_progress -gt 0 ]; then
        log_message "仍有 $in_progress 個分割區正在遷移資料..."
        log_message "將在 $CHECK_INTERVAL 秒後再次檢查"
        log_message "----------------------------------------"
        sleep $CHECK_INTERVAL
    else
        # 所有分割區都已完成
        log_message "所有分割區重新分配已成功完成！"
        log_message "總共處理 $total_partitions 個分割區"
        log_message "完成 $check_count 次狀態檢查"
        log_message "----------------------------------------"

        # 發送成功通知
        send_notification \
            "Kafka 分割區重新分配完成" \
            "所有 $total_partitions 個分割區已成功完成重新分配。詳細資訊請查看日誌檔案：$LOG_FILE"

        # 執行後續驗證
        log_message "執行最終驗證..."
        kafka-topics.sh \
            --bootstrap-server "$BOOTSTRAP_SERVER" \
            --describe >> "$LOG_FILE" 2>&1

        log_message "重新分配程序正常結束"
        exit 0
    fi
done

# 使用流量限制執行分割區重新分配
# 這可以降低資料遷移對正常業務流量的影響
kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file cluster-reassignment-plan.json \
  --execute \
  --throttle 52428800

# 參數說明：
# --throttle 52428800
#   設定每秒最大資料傳輸量，單位為位元組（Bytes）
#   52428800 位元組 = 50 MB/s
#   這個限制會套用到複製流量與追隨者拉取流量
#
# 限制值的設定建議：
# - 對於 1 Gbps 網路：建議設定在 50-100 MB/s
# - 對於 10 Gbps 網路：建議設定在 200-500 MB/s
# - 在高峰時段：建議設定較低的限制值
# - 在離峰時段：可以適度提高限制值
#
# 計算範例：
# 如果叢集有 1 Gbps (125 MB/s) 的網路頻寬
# 當前業務流量佔用約 50 MB/s
# 可以設定限制為 50 MB/s，保留約 25 MB/s 的緩衝空間

# 動態調整流量限制
# 在重新分配進行中可以動態修改限制值
kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file cluster-reassignment-plan.json \
  --execute \
  --throttle 104857600  # 調整為 100 MB/s

# 移除流量限制
# 設定為 -1 表示不限制流量
kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file cluster-reassignment-plan.json \
  --execute \
  --throttle -1

# 停用機架感知功能生成重新分配計畫
# 當嚴格的機架約束導致無法生成可行方案時使用
kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --topics-to-move-json-file topics.json \
  --broker-list 5,6,7,8 \
  --generate \
  --disable-rack-aware

# 使用場景：
# 1. 叢集只有兩個機架，但副本因子為 3
# 2. 某個機架的 Broker 暫時離線，導致可用機架不足
# 3. 在開發或測試環境中，不需要嚴格的機架分散
#
# 注意事項：
# 停用機架感知會降低容錯能力
# 應該在解決了機架資源問題後重新啟用
# 並重新執行分配以滿足機架分散要求

# 取消正在進行的分割區重新分配
# 這會停止資料遷移並嘗試還原到原始狀態
kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file cluster-reassignment-plan.json \
  --cancel

# 取消行為說明：
# 1. 停止所有進行中的資料複製
# 2. 移除尚未完成同步的新副本
# 3. 保留原有的副本配置
# 4. 恢復原始的 Leader 分配
#
# 重要提示：
# - 取消操作可能無法完全還原狀態
# - 已經完成同步的分割區可能保持新的配置
# - 建議在取消後執行驗證，確認叢集狀態
# - 必要時使用備份的原始配置重新執行分配

{
  "version": 1,
  "partitions": [
    {
      "topic": "orders",
      "partition": 0,
      "replicas": [5, 6, 7],
      "log_dirs": ["any", "any", "any"]
    },
    {
      "topic": "orders",
      "partition": 1,
      "replicas": [6, 7, 5],
      "log_dirs": ["any", "any", "any"]
    },
    {
      "topic": "orders",
      "partition": 2,
      "replicas": [7, 5, 6],
      "log_dirs": ["any", "any", "any"]
    },
    {
      "topic": "orders",
      "partition": 3,
      "replicas": [5, 6, 7],
      "log_dirs": ["any", "any", "any"]
    }
  ]
}

# 執行副本因子增加操作
# 新的副本會被建立並開始同步資料
kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file increase-replication-factor.json \
  --execute

# 監控增加進度
# 重點關注新副本的同步狀態
kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file increase-replication-factor.json \
  --verify

# 驗證副本因子是否已正確更新
# 使用 kafka-topics.sh 檢視 Topic 配置
kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --topic orders \
  --describe

# 輸出範例：
# Topic: orders    PartitionCount: 4    ReplicationFactor: 3    Configs: ...
# Topic: orders    Partition: 0    Leader: 5    Replicas: 5,6,7    Isr: 5,6,7
# Topic: orders    Partition: 1    Leader: 6    Replicas: 6,7,5    Isr: 6,7,5
# Topic: orders    Partition: 2    Leader: 7    Replicas: 7,5,6    Isr: 7,5,6
# Topic: orders    Partition: 3    Leader: 5    Replicas: 5,6,7    Isr: 5,6,7
#
# 驗證重點：
# 1. ReplicationFactor 已更新為 3
# 2. 每個分割區的 Replicas 清單包含 3 個 Broker
# 3. 所有副本都在 ISR 清單中，表示同步正常
# 4. Leader 分布均勻，每個 Broker 都承擔部分 Leader 角色

# 檢視日誌段檔案的後設資料資訊
# 這個指令會解析日誌段的結構並顯示每個批次的摘要資訊
kafka-dump-log.sh \
  --files /var/kafka-logs/orders-0/00000000000000000000.log

# 參數說明：
# --files /var/kafka-logs/orders-0/00000000000000000000.log
#   指定要分析的日誌段檔案路徑
#   路徑格式：{log.dirs}/{topic}-{partition}/{offset}.log
#   log.dirs 是 Broker 配置的資料目錄
#   topic 是 Topic 名稱
#   partition 是分割區編號
#   offset 是該日誌段的起始 offset
#
# 輸出範例：
# Dumping /var/kafka-logs/orders-0/00000000000000000000.log
# Starting offset: 0
# baseOffset: 0 lastOffset: 4 count: 5 baseSequence: -1 lastSequence: -1 
# producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false 
# isControlBatch: false position: 0 CreateTime: 1732752000000 size: 156 
# magic: 2 compresscodec: NONE crc: 123456789 isvalid: true
#
# baseOffset: 5 lastOffset: 9 count: 5 baseSequence: -1 lastSequence: -1 
# producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false 
# isControlBatch: false position: 156 CreateTime: 1732752010000 size: 178 
# magic: 2 compresscodec: GZIP crc: 987654321 isvalid: true
#
# 欄位說明：
# baseOffset：該批次的起始 offset
# lastOffset：該批次的結束 offset  
# count：批次中包含的訊息數量
# baseSequence / lastSequence：生產者端的序號，用於冪等性保證
# producerId / producerEpoch：生產者識別碼與 epoch，用於事務支援
# partitionLeaderEpoch：分割區 Leader 的 epoch 值
# isTransactional：是否為事務訊息批次
# isControlBatch：是否為控制批次（用於事務協調）
# position：在日誌檔案中的位元組位置
# CreateTime：批次建立的時間戳記
# size：批次的位元組大小
# magic：訊息格式版本（0, 1, 或 2）
# compresscodec：壓縮演算法（NONE, GZIP, SNAPPY, LZ4, ZSTD）
# crc：CRC 校驗碼，用於驗證資料完整性
# isvalid：CRC 校驗是否通過

# 檢視日誌段檔案的完整內容，包括訊息資料
# 這會解碼每則訊息的 key 與 value
kafka-dump-log.sh \
  --files /var/kafka-logs/orders-0/00000000000000000000.log \
  --print-data-log

# 參數說明：
# --print-data-log
#   印出訊息的實際內容，包括 key 與 value
#   內容會以可讀的文字格式顯示
#   對於二進位資料會以十六進位格式顯示
#
# 輸出範例：
# ...批次後設資料...
# | offset: 0 CreateTime: 1732752000000 keysize: 5 valuesize: 50 
#   sequence: -1 headerKeys: [] key: order1 
#   payload: {"orderId":"order1","customerId":"cust123","amount":99.99}
# | offset: 1 CreateTime: 1732752001000 keysize: 5 valuesize: 52 
#   sequence: -1 headerKeys: [] key: order2 
#   payload: {"orderId":"order2","customerId":"cust456","amount":149.99}
#
# 訊息欄位說明：
# offset：訊息的 offset
# CreateTime：訊息建立時間
# keysize：key 的位元組大小
# valuesize：value 的位元組大小
# sequence：生產者序號
# headerKeys：自訂標頭鍵列表
# key：訊息的 key（已解碼）
# payload：訊息的 value（已解碼）

# 只分析特定範圍的 offset
# 對於大型日誌段，可以限制分析範圍以提升效能
kafka-dump-log.sh \
  --files /var/kafka-logs/orders-0/00000000000000000000.log \
  --print-data-log \
  --offsets-decoder \
  --max-message-size 1048576

# 進階參數：
# --offsets-decoder
#   使用特殊的 offset 解碼器
#   適用於分析 __consumer_offsets 等內部 Topic
#
# --max-message-size 1048576
#   設定最大訊息大小限制（位元組）
#   超過此大小的訊息會被跳過
#   可避免處理異常大的訊息時記憶體溢位

# 驗證副本之間的資料一致性
# 這個工具會持續執行並定期報告驗證結果
kafka-replica-verification.sh \
  --broker-list kafka-broker1.example.com:9092,kafka-broker2.example.com:9092,kafka-broker3.example.com:9092 \
  --topic-white-list 'orders.*'

# 參數說明：
# --broker-list
#   指定要驗證的 Broker 清單
#   應該包含叢集中所有持有副本的 Broker
#   使用完整的主機名稱或 IP 位址加連接埠
#
# --topic-white-list 'orders.*'
#   使用正規表示式指定要驗證的 Topic
#   'orders.*' 會匹配所有以 orders 開頭的 Topic
#   '.*' 可以匹配所有 Topic（但不建議在大型叢集使用）
#
# 輸出範例（正常情況）：
# 2025-11-28 10:00:00,123: Verification process is started.
# 2025-11-28 10:00:30,456: max lag is 0 for partition orders-0 at offset 12345 
#   among 3 replicas
# 2025-11-28 10:01:00,789: max lag is 0 for partition orders-1 at offset 23456 
#   among 3 replicas
#
# 輸出範例（發現延遲）：
# 2025-11-28 10:05:00,123: max lag is 150 for partition orders-2 at offset 34567 
#   among 3 replicas
#
# 欄位說明：
# max lag：副本之間的最大 offset 差距
#   0 表示所有副本完全同步
#   小於 100 通常是可接受的範圍
#   持續增長的延遲表示存在同步問題
# partition：出現延遲的分割區
# offset：檢查的 offset 位置
# among 3 replicas：該分割區的副本總數

#!/bin/bash

# Kafka 副本一致性驗證與監控腳本
# 此腳本用於定期驗證叢集中副本的資料一致性
# 並在發現問題時記錄詳細資訊與發送告警

# 設定 Broker 清單
# 需要包含叢集中所有要驗證的 Broker 節點
BROKER_LIST="kafka-broker1.example.com:9092,kafka-broker2.example.com:9092,kafka-broker3.example.com:9092,kafka-broker4.example.com:9092"

# 設定要驗證的 Topic 正規表示式
# 可以指定特定的 Topic 或使用模式匹配
TOPIC_PATTERN="(orders|payments|inventory).*"

# 設定輸出檔案路徑
OUTPUT_DIR="/var/log/kafka/replica-verification"
OUTPUT_FILE="$OUTPUT_DIR/verification-$(date +%Y%m%d_%H%M%S).log"

# 設定執行時間（秒），0 表示持續執行
# 對於定期檢查，建議設定為 300-600 秒
DURATION=600

# 設定報告間隔（毫秒）
# 決定多久輸出一次驗證結果
REPORT_INTERVAL=30000

# 設定延遲告警閾值
# 當 max lag 超過此值時觸發告警
LAG_THRESHOLD=100

# 設定告警通知方式
ALERT_EMAIL="[email protected]"
ALERT_WEBHOOK="https://alerts.example.com/webhook/kafka"

# 建立輸出目錄
mkdir -p "$OUTPUT_DIR"

# 函式：記錄訊息
log_message() {
    local timestamp=$(date '+%Y-%m-%d %H:%M:%S')
    local message="[$timestamp] $1"
    echo "$message" | tee -a "$OUTPUT_FILE"
}

# 函式：發送告警
send_alert() {
    local subject="$1"
    local body="$2"
    
    # 發送郵件告警
    if [ -n "$ALERT_EMAIL" ]; then
        echo "$body" | mail -s "$subject" "$ALERT_EMAIL"
    fi
    
    # 發送 webhook 告警
    if [ -n "$ALERT_WEBHOOK" ]; then
        curl -X POST "$ALERT_WEBHOOK" \
            -H "Content-Type: application/json" \
            -d "{\"subject\":\"$subject\",\"body\":\"$body\"}" \
            2>/dev/null
    fi
}

log_message "開始副本一致性驗證"
log_message "Broker 清單：$BROKER_LIST"
log_message "Topic 樣式：$TOPIC_PATTERN"
log_message "執行時間：$DURATION 秒"
log_message "報告間隔：$REPORT_INTERVAL 毫秒"
log_message "延遲閾值：$LAG_THRESHOLD"
log_message "----------------------------------------"

# 建立背景程序執行驗證
if [ $DURATION -gt 0 ]; then
    # 使用 timeout 限制執行時間
    timeout $DURATION kafka-replica-verification.sh \
        --broker-list "$BROKER_LIST" \
        --topic-white-list "$TOPIC_PATTERN" \
        --report-interval-ms $REPORT_INTERVAL > "$OUTPUT_FILE.raw" 2>&1 &
else
    # 持續執行直到手動終止
    kafka-replica-verification.sh \
        --broker-list "$BROKER_LIST" \
        --topic-white-list "$TOPIC_PATTERN" \
        --report-interval-ms $REPORT_INTERVAL > "$OUTPUT_FILE.raw" 2>&1 &
fi

VERIFY_PID=$!
log_message "驗證程序已啟動，PID：$VERIFY_PID"

# 監控驗證輸出並檢測異常
tail -f "$OUTPUT_FILE.raw" | while read -r line; do
    echo "$line" >> "$OUTPUT_FILE"
    
    # 檢查是否有延遲超過閾值
    if echo "$line" | grep -q "max lag is"; then
        lag=$(echo "$line" | grep -oP 'max lag is \K[0-9]+')
        partition=$(echo "$line" | grep -oP 'partition \K[^ ]+')
        
        if [ "$lag" -gt "$LAG_THRESHOLD" ]; then
            alert_msg="檢測到副本延遲超過閾值！\n"
            alert_msg+="分割區：$partition\n"
            alert_msg+="延遲：$lag 訊息\n"
            alert_msg+="閾值：$LAG_THRESHOLD 訊息\n"
            alert_msg+="時間：$(date '+%Y-%m-%d %H:%M:%S')\n"
            alert_msg+="\n詳細資訊請查看日誌：$OUTPUT_FILE"
            
            log_message "警告：$partition 延遲 $lag 訊息（閾值：$LAG_THRESHOLD）"
            send_alert "Kafka 副本延遲告警" "$alert_msg"
        fi
    fi
done

# 等待驗證程序結束
wait $VERIFY_PID

log_message "----------------------------------------"
log_message "副本一致性驗證完成"
log_message "詳細結果請查看：$OUTPUT_FILE"

# 產生摘要報告
log_message "生成摘要報告..."
grep "max lag is" "$OUTPUT_FILE" | sort -t' ' -k5 -nr | head -20 >> "$OUTPUT_FILE.summary"

log_message "延遲最嚴重的 20 個分割區："
cat "$OUTPUT_FILE.summary" | tee -a "$OUTPUT_FILE"