97 lines
3.6 KiB
Bash
97 lines
3.6 KiB
Bash
|
#!/usr/bin/env bash
|
||
|
|
set -euo pipefail
|
||
|
|
|
||
|
|
# v3 统计投影 Topic。新建时使用 1 个分区；如果现场已经建成多个分区则保留现状，
|
||
|
|
# 应用使用固定 Kafka key，投影消息仍只会落入其中一个稳定分区。
|
||
|
|
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}"
|
||
|
|
PROJECTION_TOPIC="${PROJECTION_TOPIC:-${RAW_TOPIC}.STATS.PROJECTION}"
|
||
|
|
RETENTION_MS="${RETENTION_MS:-604800000}"
|
||
|
|
|
||
|
|
if [[ "${PROJECTION_TOPIC}" == "${RAW_TOPIC}.STATS.MATERIALIZE" ]]; then
|
||
|
|
echo "ERROR: STATS.PROJECTION 不能与 STATS.MATERIALIZE 使用同一个 Topic" >&2
|
||
|
|
exit 1
|
||
|
|
fi
|
||
|
|
|
||
|
|
if [[ -z "${KAFKA_BIN:-}" ]]; then
|
||
|
|
for candidate in \
|
||
|
|
"$(pwd)" \
|
||
|
|
"$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" \
|
||
|
|
"/opt/kafka/kafka_2.13-3.6.1/bin" \
|
||
|
|
"/opt/kafka/bin" \
|
||
|
|
"/usr/local/kafka/bin"; do
|
||
|
|
if [[ -x "${candidate}/kafka-topics.sh" ]] \
|
||
|
|
&& [[ -x "${candidate}/kafka-configs.sh" ]]; then
|
||
|
|
KAFKA_BIN="${candidate}"
|
||
|
|
break
|
||
|
|
fi
|
||
|
|
done
|
||
|
|
fi
|
||
|
|
|
||
|
|
if [[ -z "${KAFKA_BIN:-}" ]] && command -v kafka-topics.sh >/dev/null 2>&1; then
|
||
|
|
KAFKA_BIN=$(dirname -- "$(command -v kafka-topics.sh)")
|
||
|
|
fi
|
||
|
|
if [[ -z "${KAFKA_BIN:-}" ]]; then
|
||
|
|
echo "ERROR: 未找到 kafka-topics.sh；请在 Kafka bin 目录执行，或设置 KAFKA_BIN" >&2
|
||
|
|
exit 1
|
||
|
|
fi
|
||
|
|
|
||
|
|
TOPICS="${KAFKA_BIN}/kafka-topics.sh"
|
||
|
|
CONFIGS="${KAFKA_BIN}/kafka-configs.sh"
|
||
|
|
COMMON=(--bootstrap-server "${BOOTSTRAP}")
|
||
|
|
if [[ -n "${COMMAND_CONFIG:-}" ]]; then
|
||
|
|
[[ -r "${COMMAND_CONFIG}" ]] || { echo "ERROR: 无法读取 ${COMMAND_CONFIG}" >&2; exit 1; }
|
||
|
|
COMMON+=(--command-config "${COMMAND_CONFIG}")
|
||
|
|
fi
|
||
|
|
|
||
|
|
echo "Kafka=${BOOTSTRAP}"
|
||
|
|
echo "原始 Topic=${RAW_TOPIC}"
|
||
|
|
echo "统计投影 Topic=${PROJECTION_TOPIC}"
|
||
|
|
|
||
|
|
RAW_INFO=$("${TOPICS}" "${COMMON[@]}" --describe --topic "${RAW_TOPIC}")
|
||
|
|
REPLICAS=$(printf '%s\n' "${RAW_INFO}" | awk '
|
||
|
|
/ReplicationFactor:/ {for(i=1;i<=NF;i++) if($i=="ReplicationFactor:"){print $(i+1);exit}}
|
||
|
|
')
|
||
|
|
if [[ ! "${REPLICAS}" =~ ^[1-9][0-9]*$ ]]; then
|
||
|
|
echo "ERROR: 无法从 ${RAW_TOPIC} 读取副本数" >&2
|
||
|
|
exit 1
|
||
|
|
fi
|
||
|
|
|
||
|
|
echo "[1/3] 创建 Topic（已经存在时不重复创建）"
|
||
|
|
"${TOPICS}" "${COMMON[@]}" --create --if-not-exists \
|
||
|
|
--topic "${PROJECTION_TOPIC}" \
|
||
|
|
--partitions 1 \
|
||
|
|
--replication-factor "${REPLICAS}" \
|
||
|
|
--config cleanup.policy=delete \
|
||
|
|
--config "retention.ms=${RETENTION_MS}"
|
||
|
|
|
||
|
|
echo "[2/3] 更新保留策略"
|
||
|
|
if ! "${CONFIGS}" "${COMMON[@]}" --alter \
|
||
|
|
--entity-type topics --entity-name "${PROJECTION_TOPIC}" \
|
||
|
|
--add-config "cleanup.policy=delete,retention.ms=${RETENTION_MS}"; then
|
||
|
|
echo "WARN: Kafka 配置更新超时；新建命令已携带相同策略，继续验证 Topic" >&2
|
||
|
|
fi
|
||
|
|
|
||
|
|
echo "[3/3] 验证 Topic"
|
||
|
|
INFO=$("${TOPICS}" "${COMMON[@]}" --describe --topic "${PROJECTION_TOPIC}")
|
||
|
|
printf '%s\n' "${INFO}"
|
||
|
|
PARTITIONS=$(printf '%s\n' "${INFO}" | awk '
|
||
|
|
/PartitionCount:/ {for(i=1;i<=NF;i++) if($i=="PartitionCount:"){print $(i+1);exit}}
|
||
|
|
')
|
||
|
|
ACTUAL_REPLICAS=$(printf '%s\n' "${INFO}" | awk '
|
||
|
|
/ReplicationFactor:/ {for(i=1;i<=NF;i++) if($i=="ReplicationFactor:"){print $(i+1);exit}}
|
||
|
|
')
|
||
|
|
if [[ ! "${PARTITIONS}" =~ ^[1-9][0-9]*$ ]]; then
|
||
|
|
echo "ERROR: ${PROJECTION_TOPIC} 没有有效分区" >&2
|
||
|
|
exit 1
|
||
|
|
fi
|
||
|
|
if [[ "${PARTITIONS}" != "1" ]]; then
|
||
|
|
echo "WARN: Topic 已有 ${PARTITIONS} 个分区；不会修改分区数，固定 key 仍只使用其中一个分区" >&2
|
||
|
|
fi
|
||
|
|
if [[ "${ACTUAL_REPLICAS}" != "${REPLICAS}" ]]; then
|
||
|
|
echo "WARN: Topic 副本数=${ACTUAL_REPLICAS}，原始 Topic 副本数=${REPLICAS}" >&2
|
||
|
|
fi
|
||
|
|
|
||
|
|
echo "PASS: ${PROJECTION_TOPIC} 已创建/确认。分区数=${PARTITIONS}，保留时间=${RETENTION_MS}ms。"
|
||
|
|
echo "脚本未删除消息，也未修改任何消费组 offset。"
|