163 lines
5.4 KiB
Bash
163 lines
5.4 KiB
Bash
|
#!/usr/bin/env bash
|
|||
|
|
|
|||
|
|
set -euo pipefail
|
|||
|
|
|
|||
|
|
# 物理清空指定 Kafka Topic 中当前已有的消息，但保留 Topic 本身、分区、副本和配置。
|
|||
|
|
#
|
|||
|
|
# 默认值对应当前客户现场，复制到 Kafka 主机后可直接执行：
|
|||
|
|
# bash 04-clear-topic-messages.sh
|
|||
|
|
#
|
|||
|
|
# 清理其他 Topic 时可覆盖 TARGET_TOPIC：
|
|||
|
|
# TARGET_TOPIC=other-topic bash 04-clear-topic-messages.sh
|
|||
|
|
#
|
|||
|
|
# 有认证的集群需额外设置：
|
|||
|
|
# COMMAND_CONFIG=/path/to/client.properties bash 04-clear-topic-messages.sh
|
|||
|
|
#
|
|||
|
|
# 注意：
|
|||
|
|
# - 执行前必须停止目标 Topic 的生产者和消费者；
|
|||
|
|
# - 消息删除后通常不可恢复；
|
|||
|
|
# - 本脚本不会删除 Topic，也不会修改任何消费组 offset；
|
|||
|
|
# - 若消费组 offset 低于删除后的日志起始 offset，消费者后续行为取决于 auto.offset.reset。
|
|||
|
|
|
|||
|
|
SCRIPT_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)
|
|||
|
|
BOOTSTRAP="${BOOTSTRAP:-10.141.121.35:9092,10.141.121.50:9092,10.141.121.59:9092}"
|
|||
|
|
RAW_TOPIC="${RAW_TOPIC:-agent-01-syslog-topic}"
|
|||
|
|
TARGET_TOPIC="${TARGET_TOPIC:-${RAW_TOPIC}}"
|
|||
|
|
|
|||
|
|
if [[ ! "${TARGET_TOPIC}" =~ ^[a-zA-Z0-9._-]+$ ]]; then
|
|||
|
|
echo "ERROR: TARGET_TOPIC 包含非法字符：${TARGET_TOPIC}" >&2
|
|||
|
|
exit 1
|
|||
|
|
fi
|
|||
|
|
|
|||
|
|
# 优先使用调用方显式指定的目录；否则自动识别脚本同目录、现场 Kafka 安装目录或 PATH。
|
|||
|
|
if [[ -z "${KAFKA_BIN:-}" ]]; then
|
|||
|
|
for CANDIDATE_BIN in \
|
|||
|
|
"${SCRIPT_DIR}" \
|
|||
|
|
"/opt/kafka/kafka_2.13-3.6.1/bin" \
|
|||
|
|
"/opt/kafka/bin"; do
|
|||
|
|
if [[ -x "${CANDIDATE_BIN}/kafka-topics.sh" ]] \
|
|||
|
|
&& [[ -x "${CANDIDATE_BIN}/kafka-delete-records.sh" ]]; then
|
|||
|
|
KAFKA_BIN="${CANDIDATE_BIN}"
|
|||
|
|
break
|
|||
|
|
fi
|
|||
|
|
done
|
|||
|
|
fi
|
|||
|
|
|
|||
|
|
if [[ -z "${KAFKA_BIN:-}" ]] \
|
|||
|
|
&& command -v kafka-topics.sh >/dev/null 2>&1 \
|
|||
|
|
&& command -v kafka-delete-records.sh >/dev/null 2>&1; then
|
|||
|
|
KAFKA_BIN=$(dirname -- "$(command -v kafka-topics.sh)")
|
|||
|
|
fi
|
|||
|
|
|
|||
|
|
if [[ -z "${KAFKA_BIN:-}" ]]; then
|
|||
|
|
echo "ERROR: 未找到 Kafka CLI；现场预期路径为 /opt/kafka/kafka_2.13-3.6.1/bin" >&2
|
|||
|
|
exit 1
|
|||
|
|
fi
|
|||
|
|
|
|||
|
|
TOPICS_CMD="${KAFKA_BIN}/kafka-topics.sh"
|
|||
|
|
DELETE_RECORDS_CMD="${KAFKA_BIN}/kafka-delete-records.sh"
|
|||
|
|
|
|||
|
|
if [[ ! -x "${TOPICS_CMD}" ]] || [[ ! -x "${DELETE_RECORDS_CMD}" ]]; then
|
|||
|
|
echo "ERROR: ${KAFKA_BIN} 下缺少 kafka-topics.sh 或 kafka-delete-records.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
|
|||
|
|
|
|||
|
|
RUN_ID=$(date '+%Y%m%d_%H%M%S')
|
|||
|
|
AUDIT_DIR="${CLEAR_AUDIT_DIR:-./kafka-message-clear-${RUN_ID}}"
|
|||
|
|
mkdir -p "${AUDIT_DIR}"
|
|||
|
|
|
|||
|
|
BEFORE_FILE="${AUDIT_DIR}/topic-before.txt"
|
|||
|
|
DELETE_JSON="${AUDIT_DIR}/delete-records.json"
|
|||
|
|
DELETE_RESULT="${AUDIT_DIR}/delete-result.txt"
|
|||
|
|
AFTER_FILE="${AUDIT_DIR}/topic-after.txt"
|
|||
|
|
|
|||
|
|
echo "目标 Kafka：${BOOTSTRAP}"
|
|||
|
|
echo "目标 Topic：${TARGET_TOPIC}"
|
|||
|
|
echo "Kafka CLI：${KAFKA_BIN}"
|
|||
|
|
echo "审计目录：${AUDIT_DIR}"
|
|||
|
|
|
|||
|
|
echo "[1/5] 测试 Kafka 连接并确认 Topic 存在"
|
|||
|
|
"${TOPICS_CMD}" "${COMMON_ARGS[@]}" --list >/dev/null
|
|||
|
|
TOPIC_INFO=$("${TOPICS_CMD}" "${COMMON_ARGS[@]}" --describe --topic "${TARGET_TOPIC}")
|
|||
|
|
printf '%s\n' "${TOPIC_INFO}" | tee "${BEFORE_FILE}"
|
|||
|
|
|
|||
|
|
PARTITION_COUNT=$(printf '%s\n' "${TOPIC_INFO}" | awk '
|
|||
|
|
/PartitionCount:/ {
|
|||
|
|
for (i=1; i<=NF; i++) if ($i=="PartitionCount:") { print $(i+1); exit }
|
|||
|
|
}
|
|||
|
|
')
|
|||
|
|
|
|||
|
|
if [[ ! "${PARTITION_COUNT}" =~ ^[1-9][0-9]*$ ]]; then
|
|||
|
|
echo "ERROR: 无法确认 ${TARGET_TOPIC} 的分区数，拒绝继续" >&2
|
|||
|
|
exit 1
|
|||
|
|
fi
|
|||
|
|
|
|||
|
|
echo "[2/5] 为 ${PARTITION_COUNT} 个分区生成删除清单"
|
|||
|
|
printf '%s\n' "${TOPIC_INFO}" | awk -v topic="${TARGET_TOPIC}" '
|
|||
|
|
BEGIN {
|
|||
|
|
printf "{\"partitions\":["
|
|||
|
|
first=1
|
|||
|
|
}
|
|||
|
|
/Partition:/ {
|
|||
|
|
for (i=1; i<=NF; i++) {
|
|||
|
|
if ($i=="Partition:" && $(i+1)~/^[0-9]+$/) {
|
|||
|
|
if (!first) printf ","
|
|||
|
|
printf "{\"topic\":\"%s\",\"partition\":%s,\"offset\":-1}", topic, $(i+1)
|
|||
|
|
first=0
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
END {
|
|||
|
|
print "],\"version\":1}"
|
|||
|
|
}
|
|||
|
|
' > "${DELETE_JSON}"
|
|||
|
|
|
|||
|
|
DELETE_ROWS=$(awk -F'"partition":' '{count+=NF-1} END {print count+0}' "${DELETE_JSON}")
|
|||
|
|
if [[ "${DELETE_ROWS}" -ne "${PARTITION_COUNT}" ]]; then
|
|||
|
|
echo "ERROR: 删除清单只识别到 ${DELETE_ROWS}/${PARTITION_COUNT} 个分区，拒绝继续" >&2
|
|||
|
|
exit 1
|
|||
|
|
fi
|
|||
|
|
|
|||
|
|
cat "${DELETE_JSON}"
|
|||
|
|
|
|||
|
|
EXPECTED_CONFIRM="CLEAR:${TARGET_TOPIC}:${PARTITION_COUNT}"
|
|||
|
|
echo "[3/5] 等待人工确认"
|
|||
|
|
echo "警告：将永久删除 ${TARGET_TOPIC} 当前已有的全部消息，但保留 Topic。"
|
|||
|
|
echo "请确认生产者和消费者均已停止。"
|
|||
|
|
echo "如确认执行，请完整输入：${EXPECTED_CONFIRM}"
|
|||
|
|
|
|||
|
|
if [[ -n "${CONFIRM_CLEAR:-}" ]]; then
|
|||
|
|
CONFIRM_INPUT="${CONFIRM_CLEAR}"
|
|||
|
|
elif [[ -t 0 ]]; then
|
|||
|
|
read -r CONFIRM_INPUT
|
|||
|
|
else
|
|||
|
|
echo "ERROR: 当前不是交互终端；必须通过 CONFIRM_CLEAR 提供精确确认串" >&2
|
|||
|
|
exit 1
|
|||
|
|
fi
|
|||
|
|
|
|||
|
|
if [[ "${CONFIRM_INPUT}" != "${EXPECTED_CONFIRM}" ]]; then
|
|||
|
|
echo "CANCELLED: 确认串不匹配，未删除任何消息。" >&2
|
|||
|
|
exit 1
|
|||
|
|
fi
|
|||
|
|
|
|||
|
|
echo "[4/5] 清空 Topic 当前已有消息"
|
|||
|
|
"${DELETE_RECORDS_CMD}" "${COMMON_ARGS[@]}" \
|
|||
|
|
--offset-json-file "${DELETE_JSON}" | tee "${DELETE_RESULT}"
|
|||
|
|
|
|||
|
|
echo "[5/5] 再次检查 Topic"
|
|||
|
|
"${TOPICS_CMD}" "${COMMON_ARGS[@]}" \
|
|||
|
|
--describe --topic "${TARGET_TOPIC}" | tee "${AFTER_FILE}"
|
|||
|
|
|
|||
|
|
echo "PASS: ${TARGET_TOPIC} 当前已有消息已提交物理删除。"
|
|||
|
|
echo "Topic、分区、副本和配置均保留，消费组 offset 未修改。"
|
|||
|
|
echo "执行记录：${AUDIT_DIR}"
|