2026-09-11 15:42:24 +08:00
|
|
|
|
#!/usr/bin/env bash
|
|
|
|
|
|
|
|
|
|
|
|
set -euo pipefail
|
|
|
|
|
|
|
|
|
|
|
|
# Kafka 集群只需在一台装有 Kafka CLI、能够访问集群的服务器上执行一次。
|
|
|
|
|
|
# 示例：
|
|
|
|
|
|
# export KAFKA_BIN=<现场Kafka客户端bin目录>
|
|
|
|
|
|
# export BOOTSTRAP=<现场broker1地址>:<端口>,<现场broker2地址>:<端口>
|
|
|
|
|
|
# export RAW_TOPIC=<现场原始Topic名称>
|
|
|
|
|
|
# export COMMAND_CONFIG=<现场client.properties路径> # 无认证时不设置
|
|
|
|
|
|
# ./kafka-cluster-upgrade.sh
|
|
|
|
|
|
# BOOTSTRAP 必须使用客户 Kafka 运维提供的地址；集群建议填写两个以上 broker。
|
|
|
|
|
|
|
|
|
|
|
|
: "${KAFKA_BIN:?请根据现场环境设置 KAFKA_BIN}"
|
|
|
|
|
|
: "${BOOTSTRAP:?请填写客户 Kafka 运维提供的 broker 地址和端口}"
|
|
|
|
|
|
: "${RAW_TOPIC:?请填写现场实际使用的原始 Topic 名称}"
|
|
|
|
|
|
|
|
|
|
|
|
TOPICS_CMD="${KAFKA_BIN}/kafka-topics.sh"
|
|
|
|
|
|
CONFIGS_CMD="${KAFKA_BIN}/kafka-configs.sh"
|
|
|
|
|
|
|
|
|
|
|
|
if [[ ! -x "${TOPICS_CMD}" ]] || [[ ! -x "${CONFIGS_CMD}" ]]; then
|
|
|
|
|
|
echo "ERROR: ${KAFKA_BIN} 下缺少 kafka-topics.sh 或 kafka-configs.sh" >&2
|
|
|
|
|
|
exit 1
|
|
|
|
|
|
fi
|
|
|
|
|
|
|
|
|
|
|
|
COMMON_ARGS=(--bootstrap-server "${BOOTSTRAP}")
|
|
|
|
|
|
if [[ -n "${COMMAND_CONFIG:-}" ]]; then
|
|
|
|
|
|
if [[ ! -r "${COMMAND_CONFIG}" ]]; then
|
|
|
|
|
|
echo "ERROR: 无法读取认证配置 ${COMMAND_CONFIG}" >&2
|
|
|
|
|
|
exit 1
|
|
|
|
|
|
fi
|
|
|
|
|
|
COMMON_ARGS+=(--command-config "${COMMAND_CONFIG}")
|
|
|
|
|
|
fi
|
|
|
|
|
|
|
|
|
|
|
|
echo "[1/4] 测试 Kafka 集群连接：${BOOTSTRAP}"
|
|
|
|
|
|
"${TOPICS_CMD}" "${COMMON_ARGS[@]}" --list >/dev/null
|
|
|
|
|
|
|
|
|
|
|
|
echo "[2/4] 读取原始 Topic：${RAW_TOPIC}"
|
|
|
|
|
|
RAW_INFO=$("${TOPICS_CMD}" "${COMMON_ARGS[@]}" --describe --topic "${RAW_TOPIC}")
|
|
|
|
|
|
printf '%s\n' "${RAW_INFO}"
|
|
|
|
|
|
|
|
|
|
|
|
PARTITIONS=$(printf '%s\n' "${RAW_INFO}" | awk '
|
|
|
|
|
|
/PartitionCount:/ {
|
|
|
|
|
|
for (i = 1; i <= NF; i++) if ($i == "PartitionCount:") { print $(i + 1); exit }
|
|
|
|
|
|
}
|
|
|
|
|
|
')
|
|
|
|
|
|
REPLICAS=$(printf '%s\n' "${RAW_INFO}" | awk '
|
|
|
|
|
|
/ReplicationFactor:/ {
|
|
|
|
|
|
for (i = 1; i <= NF; i++) if ($i == "ReplicationFactor:") { print $(i + 1); exit }
|
|
|
|
|
|
}
|
|
|
|
|
|
')
|
|
|
|
|
|
|
|
|
|
|
|
if [[ ! "${PARTITIONS}" =~ ^[1-9][0-9]*$ ]] || [[ ! "${REPLICAS}" =~ ^[1-9][0-9]*$ ]]; then
|
|
|
|
|
|
echo "ERROR: 无法读取原始 Topic 的分区数或副本数" >&2
|
|
|
|
|
|
exit 1
|
|
|
|
|
|
fi
|
|
|
|
|
|
|
|
|
|
|
|
echo "新 Topic 使用分区数=${PARTITIONS}，副本数=${REPLICAS}"
|
|
|
|
|
|
|
2026-09-11 20:42:22 +08:00
|
|
|
|
SUFFIXES=("RETRY.1M" "RETRY.10M" "RETRY.1H" "DLT" "REPLAY" "STATS.MATERIALIZE")
|
|
|
|
|
|
RETENTIONS=(259200000 604800000 604800000 7776000000 604800000 604800000)
|
2026-09-11 15:42:24 +08:00
|
|
|
|
|
|
|
|
|
|
echo "[3/4] 创建或更新升级 Topic"
|
2026-09-11 20:42:22 +08:00
|
|
|
|
for i in 0 1 2 3 4 5; do
|
2026-09-11 15:42:24 +08:00
|
|
|
|
TOPIC="${RAW_TOPIC}.${SUFFIXES[$i]}"
|
|
|
|
|
|
RETENTION_MS="${RETENTIONS[$i]}"
|
|
|
|
|
|
|
|
|
|
|
|
echo "处理 ${TOPIC}"
|
|
|
|
|
|
"${TOPICS_CMD}" "${COMMON_ARGS[@]}" \
|
|
|
|
|
|
--create --if-not-exists \
|
|
|
|
|
|
--topic "${TOPIC}" \
|
|
|
|
|
|
--partitions "${PARTITIONS}" \
|
|
|
|
|
|
--replication-factor "${REPLICAS}" \
|
|
|
|
|
|
--config cleanup.policy=delete \
|
|
|
|
|
|
--config "retention.ms=${RETENTION_MS}"
|
|
|
|
|
|
|
|
|
|
|
|
"${CONFIGS_CMD}" "${COMMON_ARGS[@]}" \
|
|
|
|
|
|
--alter --entity-type topics --entity-name "${TOPIC}" \
|
|
|
|
|
|
--add-config "cleanup.policy=delete,retention.ms=${RETENTION_MS}"
|
|
|
|
|
|
done
|
|
|
|
|
|
|
|
|
|
|
|
echo "[4/4] 显示升级结果"
|
|
|
|
|
|
for SUFFIX in "${SUFFIXES[@]}"; do
|
|
|
|
|
|
"${TOPICS_CMD}" "${COMMON_ARGS[@]}" \
|
|
|
|
|
|
--describe --topic "${RAW_TOPIC}.${SUFFIX}"
|
|
|
|
|
|
done
|
|
|
|
|
|
|
|
|
|
|
|
echo "PASS: Kafka 集群 Topic 已创建或更新；未删除 Topic，未修改消费组 offset。"
|
