fix: 批量回收 Kafka 临时消费组

config/platform-infra/kafka.yaml

@@ -251,7 +251,7 @@ management:
     deleteBatchSize: 50
     automatic:
       enabled: true
-      schedule: "*/15 * * * *"
+      schedule: "*/10 * * * *"
       image: quay.io/strimzi/kafka:1.1.0-kafka-4.3.0
       consumerGroupsCommand: /opt/kafka/bin/kafka-consumer-groups.sh
       configMapName: platform-infra-kafka-group-retention

scripts/native/platform-infra/kafka-consumer-group-retention.sh

@@ -43,6 +43,17 @@ state_allowed() {
   return 1
 }
 
+declare -A group_states=()
+while read -r group state; do
+  [[ -n "$group" && -n "$state" ]] || continue
+  group_states["$group"]="$state"
+done < <(
+  timeout "$KAFKA_COMMAND_TIMEOUT_SECONDS" "${groups_cli[@]}" --describe --all-groups --state 2>/dev/null |
+    awk 'NF >= 2 && $1 != "GROUP" { print $1, $(NF-1) }'
+)
+
+selected_groups=()
+
 while IFS= read -r group; do
   [[ -n "$group" ]] || continue
   scanned=$((scanned + 1))
@@ -56,18 +67,34 @@ while IFS= read -r group; do
   age_matched=$((age_matched + 1))
   (( selected < KAFKA_CANDIDATE_LIMIT )) || continue
 
-  state="$(timeout "$KAFKA_COMMAND_TIMEOUT_SECONDS" "${groups_cli[@]}" --describe --group "$group" --state 2>/dev/null | awk 'NR > 1 && NF >= 2 { value=$(NF-1) } END { print value }')"
+  state="${group_states[$group]:-}"
   state_allowed "$state" || continue
   inactive_matched=$((inactive_matched + 1))
   selected=$((selected + 1))
-  (( selected <= KAFKA_DELETE_BATCH_SIZE )) || continue
-  if timeout "$KAFKA_COMMAND_TIMEOUT_SECONDS" "${groups_cli[@]}" --delete --group "$group" >/dev/null 2>&1; then
-    deleted=$((deleted + 1))
-  else
-    failed=$((failed + 1))
-  fi
+  selected_groups+=("$group")
 done < <(timeout "$KAFKA_COMMAND_TIMEOUT_SECONDS" "${groups_cli[@]}" --list 2>/dev/null)
 
+delete_args=()
+delete_count="${#selected_groups[@]}"
+(( delete_count <= KAFKA_DELETE_BATCH_SIZE )) || delete_count="$KAFKA_DELETE_BATCH_SIZE"
+for ((index = 0; index < delete_count; index += 1)); do
+  delete_args+=(--group "${selected_groups[$index]}")
+done
+if (( delete_count > 0 )); then
+  timeout "$KAFKA_COMMAND_TIMEOUT_SECONDS" "${groups_cli[@]}" --delete "${delete_args[@]}" >/dev/null 2>&1 || true
+  declare -A remaining_groups=()
+  while IFS= read -r group; do
+    [[ -n "$group" ]] && remaining_groups["$group"]=1
+  done < <(timeout "$KAFKA_COMMAND_TIMEOUT_SECONDS" "${groups_cli[@]}" --list 2>/dev/null)
+  for ((index = 0; index < delete_count; index += 1)); do
+    if [[ -n "${remaining_groups[${selected_groups[$index]}]:-}" ]]; then
+      failed=$((failed + 1))
+    else
+      deleted=$((deleted + 1))
+    fi
+  done
+fi
+
 printf '{"ok":%s,"policy":"%s","scanned":%d,"prefixMatched":%d,"patternMatched":%d,"ageMatched":%d,"inactiveMatched":%d,"selected":%d,"deleted":%d,"failed":%d,"valuesPrinted":false}\n' \
   "$([[ "$failed" -eq 0 ]] && printf true || printf false)" \
   "$KAFKA_POLICY_ID" "$scanned" "$prefix_matched" "$pattern_matched" "$age_matched" \

scripts/src/platform-infra-kafka.ts

@@ -2094,7 +2094,7 @@ function renderStatus(result: Record<string, unknown>): RenderedCliResult {
   const topics = arrayRecords(summary.topics).map((topic) => [stringValue(topic.topicName), boolText(topic.ready), stringValue(topic.resource)]);
   const users = arrayRecords(summary.users).map((user) => [stringValue(user.name), boolText(user.ready), stringValue(user.secret)]);
   const pods = arrayRecords(summary.pods).map((pod) => [stringValue(pod.name), stringValue(pod.phase), boolText(pod.ready)]);
-  const retention = arrayRecords(summary.retention).map((job) => [stringValue(job.name), stringValue(job.schedule), boolText(!job.suspended), stringValue(job.lastSuccessfulTime)]);
+  const retention = arrayRecords(summary.retention).map((job) => [stringValue(job.name), stringValue(job.schedule), boolText(!job.suspended), stringValue(job.active, "0"), stringValue(job.lastSuccessfulTime)]);
   return rendered(result, "platform-infra kafka status", [
     "PLATFORM-INFRA KAFKA STATUS",
     ...table(["TARGET", "ROUTE", "NAMESPACE", "READY"], [[stringValue(summary.target), stringValue(summary.route), stringValue(summary.namespace), boolText(summary.ready)]]),
@@ -2119,7 +2119,7 @@ function renderStatus(result: Record<string, unknown>): RenderedCliResult {
     ...(pods.length === 0 ? ["-"] : table(["POD", "PHASE", "READY"], pods)),
     "",
     "GROUP RETENTION",
-    ...(retention.length === 0 ? ["-"] : table(["CRONJOB", "SCHEDULE", "ENABLED", "LAST_SUCCESS"], retention)),
+    ...(retention.length === 0 ? ["-"] : table(["CRONJOB", "SCHEDULE", "ENABLED", "ACTIVE", "LAST_SUCCESS"], retention)),
     ...remoteErrorLines(result),
     "",
     `NEXT  bun scripts/cli.ts platform-infra kafka validate --target ${stringValue(summary.target)}`,