perf(syslog-consumer-rule): 非标日志改为单线程异步批量写入

- 非标写库结果不再参与 Kafka ACK - 使用独立单线程和有界队列处理非标日志 - 非标数据按固定大小分批写入达梦 - 增加异步成功、失败和队列拒绝指标 - 补充异步解耦及批量写入测试 perf(syslog-consumer-rule): 按批次字段并集优化达梦批量写入 - 普通日志和告警日志按非空字段并集生成批写模板 - 为同批记录补齐缺失字段并绑定 NULL - 使用 MyBatis ExecutorType.BATCH 复用 PreparedStatement - 避免固定写入全部 373 个候选字段 - 非标日志保留固定 24 列多行批量写入 - 批量大小和非标队列容量默认调整为 500 - 补充字段并集与动态 SQL 一致性测试
2026-09-11 15:42:24 +08:00
parent 61863c469b
commit ce473b2c7d
24 changed files with 2419 additions and 1592 deletions
@@ -50,18 +50,31 @@ public class LogNormalProcessor {
    public static final class NormalizationResult {
        private final ProcessingOutcome outcome;
        private final List<Map<String, Object>> rows;
+        private final List<SyslogNonNormalMessage> nonNormalMessages;

-        private NormalizationResult(ProcessingOutcome outcome, List<Map<String, Object>> rows) {
+        private NormalizationResult(ProcessingOutcome outcome, List<Map<String, Object>> rows,
+                                    List<SyslogNonNormalMessage> nonNormalMessages) {
            this.outcome = outcome;
            this.rows = Collections.unmodifiableList(new ArrayList<>(rows));
+            this.nonNormalMessages = Collections.unmodifiableList(new ArrayList<>(nonNormalMessages));
        }

        static NormalizationResult success(List<Map<String, Object>> rows) {
-            return new NormalizationResult(ProcessingOutcome.SUCCESS, rows);
+            return success(rows, Collections.emptyList());
+        }
+
+        static NormalizationResult success(List<Map<String, Object>> rows,
+                                           List<SyslogNonNormalMessage> nonNormalMessages) {
+            return new NormalizationResult(ProcessingOutcome.SUCCESS, rows, nonNormalMessages);
        }

        static NormalizationResult filtered() {
-            return new NormalizationResult(ProcessingOutcome.FILTERED, Collections.emptyList());
+            return filtered(Collections.emptyList());
+        }
+
+        static NormalizationResult filtered(List<SyslogNonNormalMessage> nonNormalMessages) {
+            return new NormalizationResult(ProcessingOutcome.FILTERED,
+                    Collections.emptyList(), nonNormalMessages);
        }

        public ProcessingOutcome getOutcome() {
@@ -71,6 +84,14 @@ public class LogNormalProcessor {
        public List<Map<String, Object>> getRows() {
            return rows;
        }
+
+        /**
+         * 活动 Kafka 批处理链路中需要集中落库的非标日志。
+         * 历史单条入口仍会立即保存，因此其结果通常为空。
+         */
+        public List<SyslogNonNormalMessage> getNonNormalMessages() {
+            return nonNormalMessages;
+        }
    }

    private static final ZoneId RECEIVE_TIME_ZONE = ZoneId.of("Asia/Shanghai");
@@ -126,7 +147,9 @@ public class LogNormalProcessor {
        RuleExecutionMetrics executionMetrics;
        MessageParseContext messageParseContext;
        boolean shadowMode;
+        boolean deferNonNormalPersistence;
        final List<Map<String, Object>> normalizedRows = new ArrayList<>();
+        final List<SyslogNonNormalMessage> deferredNonNormalMessages = new ArrayList<>();
    }

    // ==================== 构造器注入（Spring 自动完成）====================
@@ -254,7 +277,7 @@ public class LogNormalProcessor {
     * 只执行标准化计算并返回待写入行，不在本方法中写两张标准化表。
     *
     * <p>这是批量写入的关键边界：并行线程只负责 CPU/规则计算，达梦事务由 Kafka 批次汇总后的
-     * 批写器统一管理。非标日志仍沿用原有独立保存路径，不属于本次两张标准化表的改造范围。</p>
+     * 批写器统一管理。活动 Kafka 链路产生的非标日志也随结果返回，在 poll 边界集中批写。</p>
     */
    public NormalizationResult prepare(String logMsg, String syslogUUID, String syslogTopic,
                                       LocalDateTime fallbackFirstReceivedAt) {
@@ -326,6 +349,8 @@ public class LogNormalProcessor {
        ctx.executionMetrics = executionMetrics == null ? new RuleExecutionMetrics() : executionMetrics;
        ctx.messageParseContext = new MessageParseContext(
                optimizationProperties.getMessageParseCache().isEnabled(), globalMetrics);
+        // 活动 Kafka 链路必须把非标日志带回 poll 边界集中批写，禁止计算线程逐条开启达梦事务。
+        ctx.deferNonNormalPersistence = true;
        return init(ctx);
    }

@@ -373,7 +398,7 @@ public class LogNormalProcessor {
                // 原文已在进入标准化前写入 InfluxDB。设备没有绑定规则是可预期的业务终态，
                // 继续 Retry/DLT 不会凭空产生规则，只会重复消耗 Kafka、线程和日志 I/O。
                // 因此只跳过达梦标准化并允许 ACK；系统依赖和规则执行异常仍由下方异常链处理。
-                return NormalizationResult.filtered();
+                return NormalizationResult.filtered(ctx.deferredNonNormalMessages);
            }
            NormalizationOptimizationProperties.RuleSelection selectionConfig =
                    optimizationProperties.getRuleSelection();
@@ -420,9 +445,9 @@ public class LogNormalProcessor {
        if (!decision.isSelected()) {
            saveSelectionFailure(ctx, deviceId, "未命中规则",
                    "同设备多规则均未形成可用归属候选");
-            // 规则均未命中是确定性的业务终态；非标保存成功后直接 ACK，避免 Retry
-            // 为同一个 syslog_uuid 重复写入随机 ID 的非标记录。保存失败仍会由上层异常链重试。
-            return NormalizationResult.filtered();
+            // 规则均未命中是确定性的业务终态；非标记录随结果返回并交给独立线程批写。
+            // 按现场临时策略，非标写入结果不参与原 Kafka 消息的 ACK。
+            return NormalizationResult.filtered(ctx.deferredNonNormalMessages);
        }

        RuleSelectionEngine.Candidate<ExecutionPlan> selected = decision.getWinner();
@@ -442,7 +467,7 @@ public class LogNormalProcessor {
                    saveSelectionFailure(ctx, deviceId, "多规则归属歧义",
                            "候选评分相同但业务结果不同, candidateRuleIds="
                                    + candidateRuleIds(decision.getTiedCandidates()));
-                    return NormalizationResult.filtered();
+                    return NormalizationResult.filtered(ctx.deferredNonNormalMessages);
                }
                globalMetrics.ruleSelection("equivalent_tie");
                prepared = byRule.get(selected.getSource().getId());
@@ -716,8 +741,12 @@ public class LogNormalProcessor {
    }

    private NormalizationResult finish(ProcessContext ctx) {
-        if (ctx.savedCount > 0) return NormalizationResult.success(ctx.normalizedRows);
-        if (ctx.filteredCount > 0 && ctx.rejectedCount == 0) return NormalizationResult.filtered();
+        if (ctx.savedCount > 0) {
+            return NormalizationResult.success(ctx.normalizedRows, ctx.deferredNonNormalMessages);
+        }
+        if (ctx.filteredCount > 0 && ctx.rejectedCount == 0) {
+            return NormalizationResult.filtered(ctx.deferredNonNormalMessages);
+        }
        throw new IllegalArgumentException("Syslog 未产生可用标准化结果: eventId=" + ctx.strSyslogUUID);
    }

@@ -1133,7 +1162,9 @@ public class LogNormalProcessor {
            normalMessage.setCollectTaskId(collect_id);
            if (collectTask != null)
                normalMessage.setCollectTaskName(collectTask.getTaskName());
-            if (!this.messageService.saveMessage(normalMessage)) {
+            if (ctx.deferNonNormalPersistence) {
+                ctx.deferredNonNormalMessages.add(normalMessage);
+            } else if (!this.messageService.saveMessage(normalMessage)) {
                throw new IllegalStateException("非标日志服务返回保存失败");
            }
            ctx.isSaveNonNormal = true;
@@ -9,7 +9,9 @@ import com.config.AppProperties;
 import com.config.KafkaConsumerProperties;
 import com.common.service.DmStandardizedBatchWriter;
 import com.common.service.IngressStatsProjectionWriter;
+import com.common.service.SyslogNonNormalMessageService;
 import com.common.entity.IngressStatsEvent;
+import com.common.entity.SyslogNonNormalMessage;
 import com.influxdb.client.write.Point;
 import com.influx.raw.InfluxRawPointFactory;
 import com.influx.raw.PointTimeResolver;
@@ -71,6 +73,13 @@ public class SysLogProcessor {
    @Autowired
    private DmStandardizedBatchWriter standardizedBatchWriter;

+    @Autowired
+    private SyslogNonNormalMessageService nonNormalMessageService;
+
+    @Autowired
+    @Qualifier("nonNormalWriterExecutor")
+    private Executor nonNormalWriterExecutor;
+
    @Autowired
    private IngressStatsProjectionWriter ingressStatsProjectionWriter;

@@ -83,6 +92,9 @@ public class SysLogProcessor {
    @Autowired private RawInfluxMetrics rawInfluxMetrics;

    private final AtomicInteger totalProcessed = new AtomicInteger(0);
+    private final AtomicLong nonNormalAsyncCompletedTotal = new AtomicLong();
+    private final AtomicLong nonNormalAsyncFailedTotal = new AtomicLong();
+    private final AtomicLong nonNormalAsyncRejectedTotal = new AtomicLong();
    // [SYSLOG-ALARM][KACK][已修复-主链路停用] 历史辅助方法仍保留此计数用于兼容，但唯一活动 listener 不再用全局计数决定 ACK。
    private final AtomicInteger currentBatchCount = new AtomicInteger(0);

@@ -133,8 +145,9 @@ public class SysLogProcessor {
    }

    /**
-     * 统一批处理边界：原文批量写入 InfluxDB 后，有界并行执行标准化；
-     * 只有整批全部到达可靠终态才提交 offset。
+     * 统一批处理边界：原文批量写入 InfluxDB 后，有界并行执行标准化。
+     * 正常消息仍需到达可靠终态才提交 offset；非标消息按现场临时策略异步落库，
+     * 其写库结果不参与 Kafka ACK。
     */
    private void consumeBatch(List<ConsumerRecord<String, String>> records, Acknowledgment ack,
                              boolean enforceRetryDue, String lane) {
@@ -180,7 +193,9 @@ public class SysLogProcessor {
                        + "ruleComputeMs={}, ruleSnapshotMs={}, messageParseMs={}, fieldMappingMs={}, "
                        + "filterCompleteMs={}, enrichmentMs={}, ruleAssembleMs={}, compiledRuleHit={}, "
                        + "compiledRuleMiss={}, messageParseHit={}, messageParseMiss={}, deviceUnique={}, "
-                        + "collectTaskUnique={}, shadowCompared={}, shadowMismatch={}, dmWriteMs={}, totalMs={}",
+                        + "collectTaskUnique={}, shadowCompared={}, shadowMismatch={}, dmWriteMs={}, "
+                        + "nonNormalQueued={}, nonNormalRejected={}, nonNormalAsyncCompletedTotal={}, "
+                        + "nonNormalAsyncFailedTotal={}, nonNormalAsyncRejectedTotal={}, totalMs={}",
                lane, topicPartitions(records), records.size(), preparedMessages.size(), persistedMessages.size(),
                metrics.success.get(), metrics.filtered.get(), metrics.retry.get(), metrics.dlt.get(),
                metrics.normalizedRows.get(), metrics.dmSuccessfulRows.get(), metrics.dmFailedMessages.get(), ingressGroups,
@@ -190,6 +205,9 @@ public class SysLogProcessor {
                metrics.compiledRuleMisses(), metrics.messageParseHits(), metrics.messageParseMisses(),
                metrics.deviceUnique(), metrics.collectTaskUnique(), metrics.shadowCompared(),
                metrics.shadowDifferences(), metrics.dmWriteMs(),
+                metrics.nonNormalQueued(), metrics.nonNormalRejected(),
+                nonNormalAsyncCompletedTotal.get(), nonNormalAsyncFailedTotal.get(),
+                nonNormalAsyncRejectedTotal.get(),
                System.currentTimeMillis() - batchStartedAt);
    }

@@ -492,15 +510,18 @@ public class SysLogProcessor {
            metrics.recordRuleExecution(executionMetrics, enrichmentContext);

            List<DmStandardizedBatchWriter.WriteRequest<PreparedMessage>> writeRequests = new ArrayList<>();
+            List<SyslogNonNormalMessage> nonNormalMessages = new ArrayList<>();
            Set<PreparedMessage> pendingDatabaseMessages = new HashSet<>();
+            int filteredMessages = 0;
            for (CompletableFuture<PreparedNormalization> future : futures) {
                PreparedNormalization prepared = future.join();
                if (prepared.terminalState != null) {
                    metrics.record(prepared.terminalState);
                    continue;
                }
+                nonNormalMessages.addAll(prepared.result.getNonNormalMessages());
                if (prepared.result.getOutcome() == LogNormalProcessor.ProcessingOutcome.FILTERED) {
-                    metrics.record(SyslogFailureRouter.TerminalState.FILTERED);
+                    filteredMessages++;
                    continue;
                }

@@ -528,6 +549,14 @@ public class SysLogProcessor {
                    metrics.record(SyslogFailureRouter.TerminalState.SUCCESS);
                }
            }
+            for (int i = 0; i < filteredMessages; i++) {
+                metrics.record(SyslogFailureRouter.TerminalState.FILTERED);
+            }
+
+            // 正常链路先达到可靠终态，再把非标日志交给独立单线程，避免正常写入失败重试时
+            // 提前制造非标重复数据。这里只提交任务，不等待结果，非标写入不参与 Kafka ACK。
+            // 队列满、写库失败或进程异常退出都可能丢非标数据，这是现场选择的性能优先策略。
+            submitNonNormalMessages(nonNormalMessages, metrics);
        } catch (TimeoutException timeout) {
            cancelOutstanding(futures);
            throw new IllegalStateException("标准化批次处理超时，禁止提交 offset", timeout);
@@ -544,6 +573,48 @@ public class SysLogProcessor {
        }
    }

+    /**
+     * 把非标批次交给独立线程后立即返回。任何提交或执行失败只记录，不传播到 Kafka ACK 边界。
+     */
+    private void submitNonNormalMessages(List<SyslogNonNormalMessage> messages, BatchMetrics metrics) {
+        if (messages.isEmpty()) {
+            return;
+        }
+        List<SyslogNonNormalMessage> batch = new ArrayList<>(messages);
+        try {
+            nonNormalWriterExecutor.execute(() -> {
+                long startedAt = System.nanoTime();
+                try {
+                    int inserted = nonNormalMessageService.saveBatchMessages(batch);
+                    if (inserted != batch.size()) {
+                        nonNormalAsyncFailedTotal.addAndGet(batch.size());
+                        log.error("非标日志异步批写结果异常，Kafka offset 不回退: rows={}, inserted={}, "
+                                        + "durationMs={}, failedTotal={}",
+                                batch.size(), inserted, elapsedMillis(startedAt),
+                                nonNormalAsyncFailedTotal.get());
+                        return;
+                    }
+                    nonNormalAsyncCompletedTotal.addAndGet(inserted);
+                    log.debug("非标日志异步批写完成: rows={}, durationMs={}, completedTotal={}",
+                            inserted, elapsedMillis(startedAt), nonNormalAsyncCompletedTotal.get());
+                } catch (Exception failure) {
+                    nonNormalAsyncFailedTotal.addAndGet(batch.size());
+                    log.error("非标日志异步批写失败，Kafka offset 不回退，当前非标数据可能丢失: "
+                                    + "rows={}, durationMs={}, failedTotal={}",
+                            batch.size(), elapsedMillis(startedAt),
+                            nonNormalAsyncFailedTotal.get(), failure);
+                }
+            });
+            metrics.recordNonNormalSubmission(batch.size(), 0);
+        } catch (RuntimeException rejected) {
+            nonNormalAsyncRejectedTotal.addAndGet(batch.size());
+            metrics.recordNonNormalSubmission(0, batch.size());
+            log.error("非标日志异步队列提交失败，当前非标数据已丢弃且不阻塞 Kafka ACK: "
+                            + "rows={}, rejectedTotal={}",
+                    batch.size(), nonNormalAsyncRejectedTotal.get(), rejected);
+        }
+    }
+
    private static long elapsedMillis(long startedAtNanos) {
        return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedAtNanos);
    }
@@ -681,6 +752,8 @@ public class SysLogProcessor {
        private final AtomicLong ruleComputeMs = new AtomicLong();
        private final AtomicLong ruleSnapshotMs = new AtomicLong();
        private final AtomicLong dmWriteMs = new AtomicLong();
+        private final AtomicInteger nonNormalQueued = new AtomicInteger();
+        private final AtomicInteger nonNormalRejected = new AtomicInteger();
        private volatile RuleExecutionMetrics ruleExecutionMetrics = new RuleExecutionMetrics();
        private volatile int deviceUnique;
        private volatile int collectTaskUnique;
@@ -703,6 +776,10 @@ public class SysLogProcessor {
            return success.get();
        }

+        int filteredCount() {
+            return filtered.get();
+        }
+
        long ruleComputeMs() {
            return ruleComputeMs.get();
        }
@@ -711,6 +788,9 @@ public class SysLogProcessor {
            return dmWriteMs.get();
        }

+        int nonNormalQueued() { return nonNormalQueued.get(); }
+        int nonNormalRejected() { return nonNormalRejected.get(); }
+
        long ruleSnapshotMs() { return ruleSnapshotMs.get(); }
        long messageParseMs() { return ruleExecutionMetrics.getMessageParseMs(); }
        long fieldMappingMs() { return ruleExecutionMetrics.getFieldMappingMs(); }
@@ -747,6 +827,11 @@ public class SysLogProcessor {
            dmWriteMs.set(Math.max(0L, elapsedMs));
        }

+        void recordNonNormalSubmission(int queued, int rejected) {
+            nonNormalQueued.addAndGet(Math.max(0, queued));
+            nonNormalRejected.addAndGet(Math.max(0, rejected));
+        }
+
        /** 仅用于批次汇总观测，不参与 ACK 判断；ACK 仍由每条消息的可靠终态决定。 */
        void recordDmWrite(int rows, int successfulRows, int failedMessages) {
            this.normalizedRows.addAndGet(rows);
@@ -15,21 +15,16 @@ import java.util.Map;
 public interface SyslogNormalAlarmMapper {

    /**
-     * 动态插入数据 - 使用 Map
+     * 使用当前小批次的字段并集模板插入告警日志。调用方保证同批 Map 具有相同键集合，
+     * 缺失值绑定为 NULL，使 MyBatis ExecutorType.BATCH 能复用 PreparedStatement。
     */
-    int insertDynamic(@Param("dataMap") Map<String, Object> dataMap);
+    int insertBatchPlanned(@Param("dataMap") Map<String, Object> dataMap);

    /**
     * 动态插入数据 - 使用实体类
     */
    int insertByEntity(SyslogNormalAlarm entity);

-    /**
-     * 批量插入数据
-     */
-    int batchInsert(@Param("dataList") List<Map<String, Object>> dataList);
-
-
    /**
     * 统计24小时内分组数据量
     */
@@ -12,21 +12,16 @@ import java.util.List;
 public interface SyslogNormalDataMapper {

    /**
-     * 动态插入数据 - 使用 Map
+     * 使用当前小批次的字段并集模板插入普通日志。调用方保证同批 Map 具有相同键集合，
+     * 缺失值绑定为 NULL，使 MyBatis ExecutorType.BATCH 能复用 PreparedStatement。
     */
-    int insertDynamic(@Param("dataMap") Map<String, Object> dataMap);
+    int insertBatchPlanned(@Param("dataMap") Map<String, Object> dataMap);

    /**
     * 动态插入数据 - 使用实体类
     */
    int insertByEntity(SyslogNormalData entity);

-    /**
-     * 批量插入数据
-     */
-    int batchInsert(@Param("dataList") java.util.List<Map<String, Object>> dataList);
-
-
    /**
     * 统计24小时内分组数据量
     */
@@ -13,13 +13,18 @@ import org.springframework.transaction.support.TransactionSynchronizationAdapter
 import org.springframework.transaction.support.TransactionSynchronizationManager;

 import java.util.List;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
 import java.util.Map;
+import java.util.Set;

 /**
 * 在一个达梦小事务中执行真正的 MyBatis JDBC batch。
 *
 * <p>这个类只负责数据库执行，不负责错误分类、拆分或 Kafka 状态。上层写入器保证传入的所有行
- * 属于同一张表和同一个动态列签名，因此 MyBatis BatchExecutor 可以复用同一个 PreparedStatement。
+ * 普通日志和告警日志分别使用当前小批次的非空字段并集模板，因此同一目标表、同一分区日期内
+ * MyBatis BatchExecutor 可以复用同一个 PreparedStatement。
 * 每个 {@link #execute(boolean, List, int)} 调用都会创建独立事务；失败时整个小批次回滚，上层随后
 * 决定二分定位还是按系统故障停止。</p>
 */
@@ -63,7 +68,7 @@ public class DmStandardizedBatchExecutor {
     * 执行一个同构小批次。
     *
     * @param alarmTable               true 写 syslog_normal_alarm，false 写 syslog_normal_data
-     * @param rows                     已按相同动态列集合分组的行
+     * @param rows                     已按相同目标表和分区日期分组的行
     * @param transactionTimeoutSeconds 当前小事务硬超时
     */
    public void execute(boolean alarmTable, List<Map<String, Object>> rows,
@@ -72,10 +77,11 @@ public class DmStandardizedBatchExecutor {
            return;
        }

+        List<Map<String, Object>> alignedRows = alignBatchColumns(rows);
        int attempt = 0;
        while (true) {
            try {
-                executeTransaction(alarmTable, rows, transactionTimeoutSeconds, attempt);
+                executeTransaction(alarmTable, alignedRows, transactionTimeoutSeconds, attempt);
                return;
            } catch (DirectStatsProjectionWriter.ConcurrentStatsWriteException concurrentFailure) {
                if (attempt >= Math.max(0, maxConcurrencyRetries)) {
@@ -89,6 +95,39 @@ public class DmStandardizedBatchExecutor {
        }
    }

+    /**
+     * 计算当前小批次所有非空字段的并集，并为每行建立相同键集合的浅拷贝。
+     *
+     * <p>Mapper XML 只允许声明过的列，额外业务键即使出现在 Map 中也不会进入 SQL。
+     * 同一键在某行不存在时，Map#get 返回 null，最终由 JDBC 按 SQL NULL 绑定。</p>
+     */
+    static List<Map<String, Object>> alignBatchColumns(List<Map<String, Object>> rows) {
+        Set<String> columnUnion = new LinkedHashSet<>();
+        for (Map<String, Object> row : rows) {
+            if (row == null) {
+                throw new IllegalArgumentException("达梦批写行不能为空");
+            }
+            for (Map.Entry<String, Object> entry : row.entrySet()) {
+                if (entry.getKey() != null && entry.getValue() != null) {
+                    columnUnion.add(entry.getKey());
+                }
+            }
+        }
+        if (columnUnion.isEmpty()) {
+            throw new IllegalArgumentException("达梦批写没有可插入的非空字段");
+        }
+
+        List<Map<String, Object>> alignedRows = new ArrayList<>(rows.size());
+        for (Map<String, Object> row : rows) {
+            Map<String, Object> aligned = new LinkedHashMap<>(columnUnion.size());
+            for (String column : columnUnion) {
+                aligned.put(column, row.get(column));
+            }
+            alignedRows.add(aligned);
+        }
+        return alignedRows;
+    }
+
    private void executeTransaction(boolean alarmTable, List<Map<String, Object>> rows,
                                    int transactionTimeoutSeconds, int retries) {

@@ -112,12 +151,12 @@ public class DmStandardizedBatchExecutor {
                SyslogNormalAlarmMapper mapper = batchSqlSession.getMapper(SyslogNormalAlarmMapper.class);
                for (Map<String, Object> row : rows) {
                    // BatchExecutor 下这里不会逐条提交；返回值也不是最终影响行数，不能据此判断成功。
-                    mapper.insertDynamic(row);
+                    mapper.insertBatchPlanned(row);
                }
            } else {
                SyslogNormalDataMapper mapper = batchSqlSession.getMapper(SyslogNormalDataMapper.class);
                for (Map<String, Object> row : rows) {
-                    mapper.insertDynamic(row);
+                    mapper.insertBatchPlanned(row);
                }
            }

@@ -32,8 +32,6 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
-import java.util.Set;
-import java.util.TreeSet;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
@@ -48,7 +46,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 * <p>写入器遵守三条不能破坏的可靠性规则：</p>
 *
 * <ol>
- *   <li>正常行按目标表、分区日期和动态列签名分组后批量提交；</li>
+ *   <li>普通日志和告警日志分别按目标表、分区日期计算批次字段并集后批量提交；</li>
 *   <li>数据类错误回滚当前小批次并有限二分，最终只把坏行交给 Retry/DLT；</li>
 *   <li>连接、超时、提交未知等系统故障绝不降级为大量单条请求，而是立即失败并保留 Kafka offset。</li>
 * </ol>
@@ -67,7 +65,7 @@ public class DmStandardizedBatchWriter {
    @Value("${dm.standardized-write.mode:BATCH}")
    private String mode;

-    @Value("${dm.standardized-write.batch-size:100}")
+    @Value("${dm.standardized-write.batch-size:500}")
    private int batchSize;

    @Value("${dm.standardized-write.transaction-timeout-seconds:45}")
@@ -161,7 +159,7 @@ public class DmStandardizedBatchWriter {
    }

    /**
-     * 相同目标表、log_time 日期和非空列集合才会进入同一批次。
+     * 两张标准化表在每个小批次计算字段并集，只按目标表和 log_time 日期分组。
     * LinkedHashMap 保持稳定遍历顺序，便于测试和现场日志复现。
     */
    private <T> void writeGroupedBatches(List<WriteRequest<T>> requests, WriteReport<T> report,
@@ -173,7 +171,7 @@ public class DmStandardizedBatchWriter {
                groups.computeIfAbsent(key, ignored -> new ArrayList<>()).add(request);
            } catch (RuntimeException invalidGroupingKey) {
                report.recordFailure(request.getSource(), new PermanentRowWriteException(
-                        "标准化结果缺少稳定分区时间或列签名无效", invalidGroupingKey));
+                        "标准化结果缺少稳定分区时间", invalidGroupingKey));
            }
        }

@@ -294,18 +292,7 @@ public class DmStandardizedBatchWriter {
        boolean alarmTable = normalDataService.isAlarmData(data);
        LocalDate partitionDate = partitionDate(data.get("log_time"));

-        // MyBatis XML 的动态列判断只包含非 null 值，所以签名也只记录非 null 列。
-        // TreeSet 提供固定顺序，避免 HashMap 插入顺序不同却被错误拆成多个批次。
-        Set<String> columns = new TreeSet<>();
-        for (Map.Entry<String, Object> entry : data.entrySet()) {
-            if (entry.getValue() != null) {
-                columns.add(entry.getKey());
-            }
-        }
-        if (columns.isEmpty()) {
-            throw new IllegalArgumentException("标准化结果没有可写入列");
-        }
-        return new BatchKey(alarmTable, partitionDate, String.join(",", columns));
+        return new BatchKey(alarmTable, partitionDate);
    }

    private LocalDate partitionDate(Object value) {
@@ -499,12 +486,10 @@ public class DmStandardizedBatchWriter {
    private static final class BatchKey {
        private final boolean alarmTable;
        private final LocalDate partitionDate;
-        private final String columnSignature;

-        private BatchKey(boolean alarmTable, LocalDate partitionDate, String columnSignature) {
+        private BatchKey(boolean alarmTable, LocalDate partitionDate) {
            this.alarmTable = alarmTable;
            this.partitionDate = partitionDate;
-            this.columnSignature = columnSignature;
        }

        @Override
@@ -513,13 +498,12 @@ public class DmStandardizedBatchWriter {
            if (!(other instanceof BatchKey)) return false;
            BatchKey that = (BatchKey) other;
            return alarmTable == that.alarmTable
-                    && Objects.equals(partitionDate, that.partitionDate)
-                    && Objects.equals(columnSignature, that.columnSignature);
+                    && Objects.equals(partitionDate, that.partitionDate);
        }

        @Override
        public int hashCode() {
-            return Objects.hash(alarmTable, partitionDate, columnSignature);
+            return Objects.hash(alarmTable, partitionDate);
        }
    }

@@ -15,6 +15,7 @@ import java.net.InetAddress;
 import java.net.UnknownHostException;

 import java.util.List;
+import java.util.LinkedHashMap;
 import java.util.Locale;
 import java.util.Map;
 import java.time.LocalDateTime;
@@ -35,21 +36,22 @@ public class SyslogNormalDataService {


    /**
-     * 动态插入单条数据 - Map方式
+     * 标准化单条兼容入口：按 event_level 路由后，两张目标表均使用受白名单约束的动态列模板。
     */
    @Transactional
    public int insertDynamic(Map<String, Object> dataMap) {
        // [流程节点7] 根据标准化结果分类落库：安全事件写入 syslog_normal_alarm，访问事件写入 syslog_normal_data。
        // [SYSLOG-ALARM][DBBATCH] 每条标准化结果单独开启事务并执行动态 INSERT，会产生大量数据库往返、SQL 解析和提交开销；
-        // 高峰期 DB 变慢又会放大 Kafka 批次超时。建议按目标表和固定列集合聚合批次写入，并用 event_id + 规则版本唯一键保证重试幂等。
+        // 高峰期 DB 变慢又会放大 Kafka 批次超时。建议按目标表和批次字段并集聚合写入，并用 event_id + 规则版本唯一键保证重试幂等。
        prepareForInsert(dataMap);

        // [流程节点7.2] event_level 非空走安全事件表；为空走访问事件表，这是两条后续分析链路的分叉点。
        boolean alarmData = isAlarmData(dataMap);
+        Map<String, Object> plannedData = nonNullColumnMap(dataMap);
        try {
            int inserted = alarmData
-                    ? syslogNormalAlarmMapper.insertDynamic(dataMap)
-                    : syslogNormalDataMapper.insertDynamic(dataMap);
+                    ? syslogNormalAlarmMapper.insertBatchPlanned(plannedData)
+                    : syslogNormalDataMapper.insertBatchPlanned(plannedData);
            if (inserted > 0) {
                if (alarmData && lateAlarmBucketWriter != null) {
                    lateAlarmBucketWriter.record(java.util.Collections.singletonList(dataMap));
@@ -72,6 +74,17 @@ public class SyslogNormalDataService {
        }
    }

+    /** 单条兼容路径也只把实际非空字段交给列模板，避免原 Map 中的空键误触发列选择。 */
+    private Map<String, Object> nonNullColumnMap(Map<String, Object> dataMap) {
+        Map<String, Object> result = new LinkedHashMap<>();
+        for (Map.Entry<String, Object> entry : dataMap.entrySet()) {
+            if (entry.getKey() != null && entry.getValue() != null) {
+                result.put(entry.getKey(), entry.getValue());
+            }
+        }
+        return result;
+    }
+
    /**
     * 在单条和批量写入之前执行完全相同的数据准备与关键字段校验。
     *
@@ -220,9 +233,8 @@ public class SyslogNormalDataService {
                data.put("log_time", new java.util.Date());
            }
        }
-        // 旧 mapper 的 batchInsert 只写固定的少数字段，DIRECT 下会造成“统计有 device/rule，
-        // 事实行却丢了 device/rule”的永久差异。主 Kafka 链路使用 DmStandardizedBatchExecutor；
-        // 这个低频兼容 API 逐条复用完整动态 INSERT，但仍处于本方法的同一个 Spring 事务中。
+        // 主 Kafka 链路使用 DmStandardizedBatchExecutor 执行真正的 JDBC batch；这个低频兼容
+        // API 在同一个 Spring 事务内复用受白名单约束的列模板，不能再调用旧的五字段 batch SQL。
        int inserted = 0;
        for (Map<String, Object> data : dataList) {
            inserted += insertDynamic(data);
@@ -11,12 +11,14 @@ import com.common.service.StatsChangeCaptureService;
 import com.common.service.DirectStatsProjectionWriter;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;

 import java.time.LocalDateTime;
 import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;

@@ -38,6 +40,9 @@ public class SyslogNonNormalMessageServiceImpl
    @Autowired
    private DirectStatsProjectionWriter directStatsProjectionWriter;

+    @Value("${dm.non-normal-write.batch-size:500}")
+    private int nonNormalBatchSize;
+
    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean saveMessage(SyslogNonNormalMessage message) {
@@ -125,8 +130,20 @@ public class SyslogNonNormalMessageServiceImpl
                }
            }

-            // 批量插入
-            int inserted = messageMapper.batchInsert(messages);
+            // 同一个服务事务内分块执行固定列多行 INSERT，避免单条提交，也避免一个 Kafka poll
+            // 直接生成过大的 SQL。所有分块成功后才统一写 DIRECT 统计并提交事务。
+            int inserted = 0;
+            int effectiveBatchSize = Math.max(1, nonNormalBatchSize);
+            for (int start = 0; start < messages.size(); start += effectiveBatchSize) {
+                int end = Math.min(start + effectiveBatchSize, messages.size());
+                int affected = messageMapper.batchInsert(
+                        new ArrayList<>(messages.subList(start, end)));
+                if (affected <= 0) {
+                    throw new IllegalStateException("非标日志批量 INSERT 未写入记录: start="
+                            + start + ", end=" + end);
+                }
+                inserted += affected;
+            }
            if (inserted > 0) {
                if (statsChangeCaptureService.isEnabled()) {
                    for (SyslogNonNormalMessage message : messages) {
@@ -31,6 +31,9 @@ public class ThreadPoolConfig {
    @Value("${dm.standardized-write.writer-queue-capacity:100}")
    private int dmWriterQueueCapacity;

+    @Value("${dm.non-normal-write.queue-capacity:500}")
+    private int nonNormalWriterQueueCapacity;
+
    @Bean("logNormalProcessorExecutor")
    @Primary
    public Executor logNormalProcessorExecutor() {
@@ -71,4 +74,22 @@ public class ThreadPoolConfig {
        executor.initialize();
        return executor;
    }
+
+    /**
+     * 非标日志使用单独的串行写线程，与正常事实批写解耦。
+     * listener 不等待该线程，队列溢出或异步写入失败只记录错误，不阻塞 Kafka ACK。
+     */
+    @Bean("nonNormalWriterExecutor")
+    public Executor nonNormalWriterExecutor() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(1);
+        executor.setMaxPoolSize(1);
+        executor.setQueueCapacity(Math.max(1, nonNormalWriterQueueCapacity));
+        executor.setThreadNamePrefix("dm-non-normal-writer-");
+        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
+        executor.setWaitForTasksToCompleteOnShutdown(true);
+        executor.setAwaitTerminationSeconds(60);
+        executor.initialize();
+        return executor;
+    }
 }
@@ -56,8 +56,11 @@ mybatis.type-aliases-package=com.common.entity
 mybatis.configuration.default-statement-timeout=30
 mybatis.configuration.default-fetch-size=1000
 mybatis.configuration.map-underscore-to-camel-case=true
+# 批次列并集写入会为部分行绑定空字段；达梦对未指定类型的 NULL 参数兼容性较差，统一使用 JDBC NULL。
+mybatis.configuration.jdbc-type-for-null=NULL
 mybatis.type-handlers-package=com.Modules.etl.handler
 mybatis-plus.configuration.map-underscore-to-camel-case=true
+mybatis-plus.configuration.jdbc-type-for-null=NULL
 mybatis-plus.type-handlers-package=com.Modules.etl.handler

 # kafka Configuration
@@ -142,10 +145,11 @@ app.processor.batch-size=100
 app.processor.process-timeout-ms=120000

 # 达梦标准化表默认启用批量写入；现场需要紧急回退时可显式切换为 SINGLE。
-# BATCH 模式按目标表、log_time 日期和动态列集合分组，失败时只对数据类错误有限二分；
+# BATCH 模式按目标表和 log_time 日期分组，按小批次实际非空字段并集执行 JDBC Batch；
+# 失败时只对数据类错误有限二分；
 # 连接、超时或提交结果未知会立即停止并保留 Kafka offset，禁止放大为逐条请求风暴。
 dm.standardized-write.mode=BATCH
-dm.standardized-write.batch-size=100
+dm.standardized-write.batch-size=${DM_STANDARDIZED_WRITE_BATCH_SIZE:500}
 dm.standardized-write.writer-concurrency=2
 dm.standardized-write.writer-queue-capacity=100
 dm.standardized-write.transaction-timeout-seconds=45
@@ -154,6 +158,8 @@ dm.standardized-write.max-split-depth=10
 dm.standardized-write.max-fallback-operations=100
 dm.standardized-write.circuit-breaker-failure-threshold=3
 dm.standardized-write.circuit-breaker-backoff-ms=30000
+dm.non-normal-write.batch-size=${DM_NON_NORMAL_WRITE_BATCH_SIZE:500}
+dm.non-normal-write.queue-capacity=${DM_NON_NORMAL_WRITE_QUEUE_CAPACITY:500}

 # ETL配置
 etl.batch.page-size=1000
@@ -4,6 +4,7 @@ import com.Modules.NormalData.compiled.BatchEnrichmentContext;
 import com.Modules.NormalData.compiled.BatchRuleSnapshot;
 import com.Modules.NormalData.compiled.CompiledRuleCache;
 import com.Modules.NormalData.compiled.CompiledRuleCompiler;
+import com.Modules.NormalData.compiled.PreparedNormalizationInput;
 import com.Modules.NormalData.compiled.RuleExecutionMetrics;
 import com.Modules.NormalData.compiled.RuleSource;
 import com.common.entity.SyslogNonNormalMessage;
@@ -90,6 +91,20 @@ class LogNormalProcessorRuleSelectionTest {
        verify(fixture.nonNormalService).saveMessage(any(SyslogNonNormalMessage.class));
    }

+    @Test
+    void activeKafkaPreparationDefersNonNormalPersistence() {
+        Fixture fixture = fixture(Arrays.asList(
+                regexRule(10L, "first", "first=(\\S+)"),
+                regexRule(20L, "second", "second=(\\S+)")), config(true));
+
+        LogNormalProcessor.NormalizationResult result = fixture.prepareActive("nothing matches");
+
+        assertEquals(LogNormalProcessor.ProcessingOutcome.FILTERED, result.getOutcome());
+        assertEquals(1, result.getNonNormalMessages().size());
+        assertEquals("未命中规则", result.getNonNormalMessages().get(0).getReason());
+        verify(fixture.nonNormalService, never()).saveMessage(any());
+    }
+
    @Test
    void nonNormalSaveFailureRemainsRetryable() {
        Fixture fixture = fixture(Arrays.asList(
@@ -232,5 +247,23 @@ class LogNormalProcessorRuleSelectionTest {
                    new BatchEnrichmentContext(false, taskService, deviceService),
                    new RuleExecutionMetrics());
        }
+
+        private LogNormalProcessor.NormalizationResult prepareActive(String body) {
+            LocalDateTime receivedAt = LocalDateTime.of(2026, 8, 21, 15, 0);
+            Map<String, String> deviceInfo = new LinkedHashMap<>();
+            deviceInfo.put("receive_time", "20260821150000000");
+            deviceInfo.put("event_id", "event-selection");
+            deviceInfo.put("device_id", "7");
+            deviceInfo.put("device_name", "fw");
+            deviceInfo.put("vendor", "vendor");
+            deviceInfo.put("data_type", "json");
+            deviceInfo.put("device_collect_id", "3");
+            PreparedNormalizationInput input = new PreparedNormalizationInput(
+                    "event-selection", "raw", receivedAt, receivedAt,
+                    7L, 3L, "json", body, body, "", deviceInfo);
+            return processor.prepare(input, snapshot,
+                    new BatchEnrichmentContext(false, taskService, deviceService),
+                    new RuleExecutionMetrics());
+        }
    }
 }
@@ -59,7 +59,7 @@ class QuanxiNormalizationTest {
                assertTrue(f.store.isAlarmData(row));
                assertEquals(1, f.store.insertDynamic(row));
            }
-            verify(f.alarm, times(20)).insertDynamic(anyMap());
+            verify(f.alarm, times(20)).insertBatchPlanned(anyMap());
            verifyNoInteractions(f.normal);
        }
    }
@@ -107,7 +107,9 @@ class QuanxiNormalizationTest {
    @Test void ruleDestinationsAreActualAlarmSqlColumnsAndHttpPathsAreExact() throws Exception {
        JsonNode rule = JSON.readTree(file("normalization-rule.json").toFile());
        String sql = new String(Files.readAllBytes(file("../../../syslog-consumer-rule/src/main/resources/mapper/SyslogNormalAlarmMapper.xml")), "UTF-8");
-        for (JsonNode m : rule.path("mappers")) assertTrue(sql.contains("dataMap." + m.path("dest_field").asText() + " "));
+        for (JsonNode m : rule.path("mappers")) {
+            assertTrue(sql.contains("dataMap." + m.path("dest_field").asText()));
+        }
        for (String[] field : HTTP_FIELDS) {
            boolean found = false;
            for (JsonNode m : rule.path("mappers")) if (field[1].equals(m.path("dest_field").asText())) {
@@ -138,7 +140,7 @@ class QuanxiNormalizationTest {
            ReflectionTestUtils.setField(store, "syslogNormalDataMapper", normal);
            ReflectionTestUtils.setField(store, "statsChangeCaptureService", mock(StatsChangeCaptureService.class));
            ReflectionTestUtils.setField(store, "directStatsProjectionWriter", mock(DirectStatsProjectionWriter.class));
-            when(alarm.insertDynamic(anyMap())).thenReturn(1);
+            when(alarm.insertBatchPlanned(anyMap())).thenReturn(1);
            NormalizationOptimizationProperties p = new NormalizationOptimizationProperties();
            p.getCompiledRule().setEnabled(compiled); p.getCompiledRule().setShadowCompareEnabled(false);
            p.getRuleSelection().setEnabled(true); p.getRuleSelection().setShadowEnabled(false);
@@ -3,9 +3,11 @@ package com.Modules.NormalData;
 import com.Modules.NormalData.compiled.BatchEnrichmentContext;
 import com.Modules.NormalData.compiled.BatchRuleSnapshot;
 import com.Modules.NormalData.compiled.PreparedNormalizationInput;
+import com.common.entity.SyslogNonNormalMessage;
 import com.common.service.DeviceCollectTaskService;
 import com.common.service.DeviceDeviceService;
 import com.common.service.DmStandardizedBatchWriter;
+import com.common.service.SyslogNonNormalMessageService;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -19,12 +21,18 @@ import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;

+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;

@@ -32,23 +40,29 @@ import static org.mockito.Mockito.when;
 class SysLogProcessorParallelTest {

    private ExecutorService executor;
+    private ExecutorService nonNormalExecutor;
    private SysLogProcessor processor;
    private LogNormalProcessor normalProcessor;
    private SyslogFailureRouter failureRouter;
    private DmStandardizedBatchWriter batchWriter;
+    private SyslogNonNormalMessageService nonNormalMessageService;

    @BeforeEach
    void setUp() {
        executor = Executors.newFixedThreadPool(4);
+        nonNormalExecutor = Executors.newSingleThreadExecutor();
        processor = new SysLogProcessor();
        normalProcessor = mock(LogNormalProcessor.class);
        failureRouter = mock(SyslogFailureRouter.class);
        batchWriter = mock(DmStandardizedBatchWriter.class);
+        nonNormalMessageService = mock(SyslogNonNormalMessageService.class);
        ReflectionTestUtils.setField(processor, "taskExecutor", executor);
        ReflectionTestUtils.setField(processor, "processTimeoutMs", 5000L);
        ReflectionTestUtils.setField(processor, "logNormalProcessor", normalProcessor);
        ReflectionTestUtils.setField(processor, "failureRouter", failureRouter);
        ReflectionTestUtils.setField(processor, "standardizedBatchWriter", batchWriter);
+        ReflectionTestUtils.setField(processor, "nonNormalMessageService", nonNormalMessageService);
+        ReflectionTestUtils.setField(processor, "nonNormalWriterExecutor", nonNormalExecutor);
        when(batchWriter.write(any())).thenReturn(new DmStandardizedBatchWriter.WriteReport<>());
        when(normalProcessor.captureRuleSnapshot(any())).thenReturn(
                new BatchRuleSnapshot(Collections.emptyMap(), System.currentTimeMillis()));
@@ -59,7 +73,9 @@ class SysLogProcessorParallelTest {
    @AfterEach
    void tearDown() throws InterruptedException {
        executor.shutdownNow();
+        nonNormalExecutor.shutdownNow();
        executor.awaitTermination(5, TimeUnit.SECONDS);
+        nonNormalExecutor.awaitTermination(5, TimeUnit.SECONDS);
    }

    @Test
@@ -121,6 +137,73 @@ class SysLogProcessorParallelTest {
                "达梦写入耗时应独立覆盖批写调用");
    }

+    @Test
+    void deferredNonNormalMessagesAreSavedOncePerKafkaBatch() {
+        when(normalProcessor.prepare(any(PreparedNormalizationInput.class),
+                any(), any(), any())).thenAnswer(invocation -> {
+            SyslogNonNormalMessage message = new SyslogNonNormalMessage();
+            message.setId(java.util.UUID.randomUUID().toString());
+            return LogNormalProcessor.NormalizationResult.filtered(
+                    Collections.singletonList(message));
+        });
+        when(nonNormalMessageService.saveBatchMessages(anyList()))
+                .thenAnswer(invocation -> ((List<?>) invocation.getArgument(0)).size());
+
+        SysLogProcessor.BatchMetrics metrics = new SysLogProcessor.BatchMetrics();
+        processor.normalizePreparedMessages(messages(8), metrics);
+
+        @SuppressWarnings("unchecked")
+        org.mockito.ArgumentCaptor<List<SyslogNonNormalMessage>> captor =
+                org.mockito.ArgumentCaptor.forClass(List.class);
+        verify(nonNormalMessageService, timeout(1000)).saveBatchMessages(captor.capture());
+        assertEquals(8, captor.getValue().size());
+        assertEquals(8, metrics.nonNormalQueued());
+        assertEquals(0, metrics.nonNormalRejected());
+        assertEquals(0, metrics.successCount());
+    }
+
+    @Test
+    void asynchronousNonNormalFailureDoesNotFailKafkaBatch() {
+        when(normalProcessor.prepare(any(PreparedNormalizationInput.class),
+                any(), any(), any())).thenAnswer(invocation -> {
+            SyslogNonNormalMessage message = new SyslogNonNormalMessage();
+            message.setId(java.util.UUID.randomUUID().toString());
+            return LogNormalProcessor.NormalizationResult.filtered(
+                    Collections.singletonList(message));
+        });
+        when(nonNormalMessageService.saveBatchMessages(anyList()))
+                .thenThrow(new IllegalStateException("dm unavailable"));
+
+        SysLogProcessor.BatchMetrics metrics = new SysLogProcessor.BatchMetrics();
+        processor.normalizePreparedMessages(messages(2), metrics);
+
+        verify(nonNormalMessageService, timeout(1000)).saveBatchMessages(anyList());
+        assertEquals(2, metrics.nonNormalQueued());
+        assertEquals(2, metrics.filteredCount());
+    }
+
+    @Test
+    void rejectedNonNormalTaskDoesNotFailKafkaBatch() {
+        when(normalProcessor.prepare(any(PreparedNormalizationInput.class),
+                any(), any(), any())).thenAnswer(invocation -> {
+            SyslogNonNormalMessage message = new SyslogNonNormalMessage();
+            return LogNormalProcessor.NormalizationResult.filtered(
+                    Collections.singletonList(message));
+        });
+        ReflectionTestUtils.setField(processor, "nonNormalWriterExecutor",
+                (java.util.concurrent.Executor) command -> {
+                    throw new RejectedExecutionException("queue full");
+                });
+
+        SysLogProcessor.BatchMetrics metrics = new SysLogProcessor.BatchMetrics();
+        processor.normalizePreparedMessages(messages(3), metrics);
+
+        verify(nonNormalMessageService, never()).saveBatchMessages(anyList());
+        assertEquals(0, metrics.nonNormalQueued());
+        assertEquals(3, metrics.nonNormalRejected());
+        assertEquals(3, metrics.filteredCount());
+    }
+
    private List<SysLogProcessor.PreparedMessage> messages(int count) {
        List<SysLogProcessor.PreparedMessage> messages = new ArrayList<>();
        for (int i = 0; i < count; i++) {
@@ -0,0 +1,127 @@
+package com.common.mapper;
+
+import org.apache.ibatis.builder.xml.XMLMapperBuilder;
+import org.apache.ibatis.mapping.MappedStatement;
+import org.apache.ibatis.session.Configuration;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Scanner;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class FixedColumnBatchMapperTest {
+
+    @Test
+    void normalDataUsesWhitelistedBatchUnionTemplate() {
+        assertBatchUnionMapper(
+                "/mapper/SyslogNormalDataMapper.xml", "syslog_normal_data");
+    }
+
+    @Test
+    void normalAlarmUsesWhitelistedBatchUnionTemplate() {
+        assertBatchUnionMapper(
+                "/mapper/SyslogNormalAlarmMapper.xml", "syslog_normal_alarm");
+    }
+
+    @Test
+    void alignedRowsProduceIdenticalMyBatisSqlEvenWhenValuesAreNull() {
+        assertSameBoundSql(
+                "/mapper/SyslogNormalDataMapper.xml",
+                "com.common.mapper.SyslogNormalDataMapper.insertBatchPlanned");
+        assertSameBoundSql(
+                "/mapper/SyslogNormalAlarmMapper.xml",
+                "com.common.mapper.SyslogNormalAlarmMapper.insertBatchPlanned");
+    }
+
+    private void assertBatchUnionMapper(String resourcePath, String table) {
+        String xml = resource(resourcePath);
+        Matcher statement = Pattern.compile(
+                "<insert id=\"insertBatchPlanned\".*?INSERT INTO " + table
+                        + "\\s*<trim.*?</trim>\\s*VALUES\\s*<trim.*?</trim>\\s*</insert>",
+                Pattern.DOTALL).matcher(xml);
+
+        assertTrue(statement.find(), "缺少 " + table + " 批次列并集 INSERT");
+        String insert = statement.group(0);
+        assertTrue(insert.contains("dataMap.containsKey('src_ip')"));
+        assertTrue(insert.contains("dataMap.containsKey('dest_ip')"));
+        assertTrue(insert.contains("dataMap.containsKey('rule_version')"));
+        assertFalse(insert.contains("dataMap.src_ip != null"),
+                "列选择必须依据调用方补齐后的统一键集合，不能再按各行值判断");
+
+        Matcher conditions = Pattern.compile("dataMap\\.containsKey\\('([a-zA-Z0-9_]+)'\\)")
+                .matcher(insert);
+        java.util.Map<String, Integer> occurrences = new java.util.HashMap<>();
+        while (conditions.find()) {
+            occurrences.merge(conditions.group(1), 1, Integer::sum);
+        }
+        assertTrue(occurrences.size() > 300, "白名单必须保留历史标准化候选字段");
+        assertTrue(occurrences.values().stream().allMatch(count -> count == 2),
+                "每个允许字段必须同时出现在列清单和值清单中");
+    }
+
+    private void assertSameBoundSql(String resourcePath, String statementId) {
+        String xml = resource(resourcePath);
+        Configuration configuration = new Configuration();
+        try (InputStream stream = new ByteArrayInputStream(xml.getBytes(StandardCharsets.UTF_8))) {
+            new XMLMapperBuilder(stream, configuration, resourcePath,
+                    configuration.getSqlFragments()).parse();
+        } catch (Exception parseFailure) {
+            throw new AssertionError("Mapper XML 解析失败: " + resourcePath, parseFailure);
+        }
+        MappedStatement statement = configuration.getMappedStatement(statementId);
+
+        Map<String, Object> first = new HashMap<>();
+        first.put("id", "row-1");
+        first.put("src_ip", "10.0.0.1");
+        first.put("dest_ip", null);
+        first.put("rule_version", "v1");
+        Map<String, Object> second = new HashMap<>();
+        second.put("id", "row-2");
+        second.put("src_ip", null);
+        second.put("dest_ip", "10.0.0.2");
+        second.put("rule_version", "v1");
+
+        String firstSql = normalizedSql(statement, first);
+        String secondSql = normalizedSql(statement, second);
+        assertEquals(firstSql, secondSql, "同一字段并集的各行必须生成完全相同的 JDBC Batch SQL");
+        assertTrue(firstSql.contains("src_ip"));
+        assertTrue(firstSql.contains("dest_ip"));
+        assertFalse(firstSql.contains("webshell_type"));
+    }
+
+    private String normalizedSql(MappedStatement statement, Map<String, Object> row) {
+        Map<String, Object> parameter = new HashMap<>();
+        parameter.put("dataMap", row);
+        return statement.getBoundSql(parameter).getSql().replaceAll("\\s+", " ").trim();
+    }
+
+    @Test
+    void nonNormalDataUsesFixedColumnsAndMultiRowValues() {
+        String xml = resource("/mapper/SyslogNonNormalMessageMapper.xml");
+        Matcher statement = Pattern.compile(
+                "<insert id=\"batchInsert\".*?</insert>", Pattern.DOTALL).matcher(xml);
+
+        assertTrue(statement.find(), "缺少 syslog_non_normal_message 批量 INSERT");
+        assertFalse(statement.group(0).contains("<if"), "非标批量 INSERT 不能动态裁剪列");
+        assertTrue(statement.group(0).contains("<foreach collection=\"list\""));
+        assertTrue(statement.group(0).contains("reason_detail"));
+    }
+
+    private String resource(String path) {
+        InputStream stream = FixedColumnBatchMapperTest.class.getResourceAsStream(path);
+        assertNotNull(stream, "找不到测试资源: " + path);
+        try (Scanner scanner = new Scanner(stream, "UTF-8").useDelimiter("\\A")) {
+            return scanner.hasNext() ? scanner.next() : "";
+        }
+    }
+}
@@ -0,0 +1,41 @@
+package com.common.service;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+class DmStandardizedBatchExecutorTest {
+
+    @Test
+    void alignsEveryRowToTheNonNullColumnUnionWithoutMutatingSources() {
+        Map<String, Object> first = new HashMap<>();
+        first.put("id", "row-1");
+        first.put("src_ip", "10.0.0.1");
+        first.put("unused_null", null);
+
+        Map<String, Object> second = new HashMap<>();
+        second.put("id", "row-2");
+        second.put("dest_ip", "10.0.0.2");
+
+        List<Map<String, Object>> aligned =
+                DmStandardizedBatchExecutor.alignBatchColumns(Arrays.asList(first, second));
+
+        assertEquals(aligned.get(0).keySet(), aligned.get(1).keySet());
+        assertEquals(Arrays.asList("dest_ip", "id", "src_ip"),
+                aligned.get(0).keySet().stream().sorted().collect(java.util.stream.Collectors.toList()));
+        assertEquals("10.0.0.1", aligned.get(0).get("src_ip"));
+        assertNull(aligned.get(0).get("dest_ip"));
+        assertNull(aligned.get(1).get("src_ip"));
+        assertEquals("10.0.0.2", aligned.get(1).get("dest_ip"));
+        assertFalse(aligned.get(0).containsKey("unused_null"));
+        assertFalse(first.containsKey("dest_ip"), "不得修改上游标准化结果 Map");
+        assertFalse(second.containsKey("src_ip"), "不得修改上游标准化结果 Map");
+    }
+}
@@ -100,6 +100,31 @@ class DmStandardizedBatchWriterTest {
        verify(normalDataService, never()).insertDynamic(any());
    }

+    @Test
+    void normalDataWithDifferentPopulatedColumnsSharesOneUnionPlannedBatch() {
+        List<DmStandardizedBatchWriter.WriteRequest<String>> requests = requests(2, -1);
+        requests.get(0).getData().put("src_ip", "10.0.0.1");
+        requests.get(1).getData().put("dest_ip", "10.0.0.2");
+
+        DmStandardizedBatchWriter.WriteReport<String> report = writer.write(requests);
+
+        assertEquals(2, report.getSuccessfulRows());
+        verify(batchExecutor, times(1)).execute(anyBoolean(), anyList(), anyInt());
+    }
+
+    @Test
+    void alarmDataWithDifferentPopulatedColumnsSharesOneUnionPlannedBatch() {
+        when(normalDataService.isAlarmData(any())).thenReturn(true);
+        List<DmStandardizedBatchWriter.WriteRequest<String>> requests = requests(2, -1);
+        requests.get(0).getData().put("src_ip", "10.0.0.1");
+        requests.get(1).getData().put("dest_ip", "10.0.0.2");
+
+        DmStandardizedBatchWriter.WriteReport<String> report = writer.write(requests);
+
+        assertEquals(2, report.getSuccessfulRows());
+        verify(batchExecutor, times(1)).execute(anyBoolean(), anyList(), anyInt());
+    }
+
    private List<DmStandardizedBatchWriter.WriteRequest<String>> requests(int count, int badIndex) {
        List<DmStandardizedBatchWriter.WriteRequest<String>> requests = new ArrayList<>();
        for (int i = 0; i < count; i++) {
@@ -108,7 +133,7 @@ class DmStandardizedBatchWriterTest {
            row.put("rule_version", "legacy");
            row.put("created_at", LocalDateTime.of(2026, 8, 11, 3, 55));
            row.put("log_time", LocalDateTime.of(2026, 8, 11, 3, 55));
-            // 所有行都保留相同列签名；坏行只改变值，确保它真的混在同一个 JDBC batch 中。
+            // 普通日志按当前小批次字段并集生成模板；额外键不会再造成批次拆分。
            row.put("bad", i == badIndex);
            requests.add(new DmStandardizedBatchWriter.WriteRequest<>("source-" + i, row));
        }
@@ -0,0 +1,61 @@
+package com.common.service;
+
+import com.common.entity.SyslogNonNormalMessage;
+import com.common.mapper.SyslogNonNormalMessageMapper;
+import com.common.service.impl.SyslogNonNormalMessageServiceImpl;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class SyslogNonNormalMessageServiceBatchTest {
+
+    private SyslogNonNormalMessageServiceImpl service;
+    private SyslogNonNormalMessageMapper mapper;
+    private DirectStatsProjectionWriter directStatsProjectionWriter;
+
+    @BeforeEach
+    void setUp() {
+        service = new SyslogNonNormalMessageServiceImpl();
+        mapper = mock(SyslogNonNormalMessageMapper.class);
+        directStatsProjectionWriter = mock(DirectStatsProjectionWriter.class);
+        ReflectionTestUtils.setField(service, "messageMapper", mapper);
+        ReflectionTestUtils.setField(service, "statsChangeCaptureService",
+                mock(StatsChangeCaptureService.class));
+        ReflectionTestUtils.setField(service, "directStatsProjectionWriter",
+                directStatsProjectionWriter);
+        ReflectionTestUtils.setField(service, "nonNormalBatchSize", 100);
+        when(mapper.batchInsert(anyList()))
+                .thenAnswer(invocation -> ((List<?>) invocation.getArgument(0)).size());
+    }
+
+    @Test
+    void chunksLargePollInsideOneServiceTransactionAndAggregatesStatsOnce() {
+        List<SyslogNonNormalMessage> messages = new ArrayList<>();
+        for (int i = 0; i < 250; i++) {
+            SyslogNonNormalMessage message = new SyslogNonNormalMessage();
+            message.setId("non-normal-" + i);
+            messages.add(message);
+        }
+
+        assertEquals(250, service.saveBatchMessages(messages));
+
+        @SuppressWarnings("unchecked")
+        ArgumentCaptor<List<SyslogNonNormalMessage>> captor = ArgumentCaptor.forClass(List.class);
+        verify(mapper, times(3)).batchInsert(captor.capture());
+        assertEquals(100, captor.getAllValues().get(0).size());
+        assertEquals(100, captor.getAllValues().get(1).size());
+        assertEquals(50, captor.getAllValues().get(2).size());
+        verify(directStatsProjectionWriter).writeNonNormalRows(messages);
+    }
+}
@@ -38,7 +38,7 @@ class SyslogNormalDataServiceIdempotencyTest {
    @Test
    void dmUniqueViolationWithExistingNormalResultIsIdempotentSuccess() {
        Map<String, Object> data = idempotencyData(false);
-        when(normalDataMapper.insertDynamic(anyMap())).thenThrow(dmUniqueViolation());
+        when(normalDataMapper.insertBatchPlanned(anyMap())).thenThrow(dmUniqueViolation());
        when(normalDataMapper.countByLogTimeAndId(
                (LocalDateTime) data.get("log_time"), (String) data.get("id")))
                .thenReturn(1);
@@ -53,7 +53,7 @@ class SyslogNormalDataServiceIdempotencyTest {
    @Test
    void dmUniqueViolationWithExistingAlarmResultIsIdempotentSuccess() {
        Map<String, Object> data = idempotencyData(true);
-        when(normalAlarmMapper.insertDynamic(anyMap())).thenThrow(dmUniqueViolation());
+        when(normalAlarmMapper.insertBatchPlanned(anyMap())).thenThrow(dmUniqueViolation());
        when(normalAlarmMapper.countByLogTimeAndId(
                (LocalDateTime) data.get("log_time"), (String) data.get("id")))
                .thenReturn(1);
@@ -70,7 +70,7 @@ class SyslogNormalDataServiceIdempotencyTest {
        Map<String, Object> data = idempotencyData(false);
        DataIntegrityViolationException invalidData = new DataIntegrityViolationException(
                "value too long", new SQLException("data too long"));
-        when(normalDataMapper.insertDynamic(anyMap())).thenThrow(invalidData);
+        when(normalDataMapper.insertBatchPlanned(anyMap())).thenThrow(invalidData);

        assertThrows(DataIntegrityViolationException.class, () -> service.insertDynamic(data));

@@ -23,9 +23,9 @@ class KafkaProducerEventIdTest {
    void maxRequestSizeIsPassedToTheManuallyConstructedKafkaProducer() {
        Properties properties = new Properties();

-        kafkaProducer.applyMaxRequestSize(properties, 4 * 1024 * 1024);
+        kafkaProducer.applyMaxRequestSize(properties, 8 * 1024 * 1024);

-        assertEquals(4 * 1024 * 1024, properties.get(ProducerConfig.MAX_REQUEST_SIZE_CONFIG));
+        assertEquals(8 * 1024 * 1024, properties.get(ProducerConfig.MAX_REQUEST_SIZE_CONFIG));
    }

    @Test
@@ -0,0 +1,195 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+# 将指定消费组在原始 Topic 上的 offset 重置到指定 Kafka 消息时间。
+# 适用于“丢弃截止时间之前的历史积压，保留截止时间之后消息”的现场操作。
+#
+# 本脚本不会停止/启动应用，也不会删除 Topic 或消息：
+# - 执行前必须由运维停止该消费组的全部 syslog-consumer-rule 实例；
+# - 自动导出原 offset，Kafka 数据仍在 retention 内时可用于人工回退；
+# - 先执行 dry-run，只有输入精确确认串才执行真正的 offset 重置。
+
+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}"
+GROUP_ID="${GROUP_ID:-agent-01-syslog-group}"
+RAW_TOPIC="${RAW_TOPIC:-agent-01-syslog-topic}"
+CUTOFF_DATETIME="${CUTOFF_DATETIME:-2026-09-11T00:00:00.000}"
+KAFKA_TIMEZONE="${KAFKA_TIMEZONE:-Asia/Shanghai}"
+
+# 优先使用调用方显式指定的目录；否则自动识别脚本同目录、现场 Kafka 安装目录或 PATH。
+if [[ -z "${KAFKA_BIN:-}" ]]; then
+  for CANDIDATE_BIN in \
+    "${SCRIPT_DIR}" \
+    "/opt/kafka/kafka_2.13-3.6.1/bin"; do
+    if [[ -x "${CANDIDATE_BIN}/kafka-consumer-groups.sh" ]] \
+        && [[ -x "${CANDIDATE_BIN}/kafka-topics.sh" ]]; then
+      KAFKA_BIN="${CANDIDATE_BIN}"
+      break
+    fi
+  done
+fi
+
+if [[ -z "${KAFKA_BIN:-}" ]] && command -v kafka-consumer-groups.sh >/dev/null 2>&1; then
+  KAFKA_BIN=$(dirname -- "$(command -v kafka-consumer-groups.sh)")
+fi
+
+if [[ -z "${KAFKA_BIN:-}" ]]; then
+  echo "ERROR: 未找到 Kafka CLI；现场预期路径为 /opt/kafka/kafka_2.13-3.6.1/bin" >&2
+  exit 1
+fi
+
+CONSUMER_GROUPS_CMD="${KAFKA_BIN}/kafka-consumer-groups.sh"
+TOPICS_CMD="${KAFKA_BIN}/kafka-topics.sh"
+
+if [[ ! -x "${CONSUMER_GROUPS_CMD}" ]] || [[ ! -x "${TOPICS_CMD}" ]]; then
+  echo "ERROR: ${KAFKA_BIN} 下缺少可执行的 kafka-consumer-groups.sh 或 kafka-topics.sh" >&2
+  exit 1
+fi
+
+if [[ ! "${CUTOFF_DATETIME}" =~ ^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3}$ ]]; then
+  echo "ERROR: CUTOFF_DATETIME 格式必须是 YYYY-MM-DDTHH:mm:ss.SSS" >&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
+
+# Kafka 3.6 的 --to-datetime 不携带时区，显式固定 JVM 时区，避免宿主机差异造成 8 小时偏移。
+export KAFKA_OPTS="${KAFKA_OPTS:-} -Duser.timezone=${KAFKA_TIMEZONE}"
+
+RUN_ID=$(date '+%Y%m%d_%H%M%S')
+AUDIT_DIR="${OFFSET_BACKUP_DIR:-./kafka-offset-reset-${RUN_ID}}"
+mkdir -p "${AUDIT_DIR}"
+
+BEFORE_FILE="${AUDIT_DIR}/before-describe.txt"
+RESTORE_FILE="${AUDIT_DIR}/restore-current-offsets.csv"
+DRY_RUN_FILE="${AUDIT_DIR}/reset-dry-run.txt"
+EXECUTE_FILE="${AUDIT_DIR}/reset-execute.txt"
+AFTER_FILE="${AUDIT_DIR}/after-describe.txt"
+
+echo "目标 Kafka：${BOOTSTRAP}"
+echo "目标消费组：${GROUP_ID}"
+echo "目标 Topic：${RAW_TOPIC}"
+echo "保留起点：${CUTOFF_DATETIME} (${KAFKA_TIMEZONE})"
+echo "审计目录：${AUDIT_DIR}"
+
+echo "[1/6] 检查 Topic"
+TOPIC_INFO=$("${TOPICS_CMD}" "${COMMON_ARGS[@]}" --describe --topic "${RAW_TOPIC}")
+printf '%s\n' "${TOPIC_INFO}"
+
+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: 无法确认 ${RAW_TOPIC} 的分区数" >&2
+  exit 1
+fi
+
+echo "[2/6] 检查消费组是否已停止，并保存执行前快照"
+"${CONSUMER_GROUPS_CMD}" "${COMMON_ARGS[@]}" \
+  --describe --group "${GROUP_ID}" 2>&1 | tee "${BEFORE_FILE}"
+
+MATCHED_ROWS=$(awk -v topic="${RAW_TOPIC}" '
+  $2==topic && $3~/^[0-9]+$/ && $4~/^[0-9]+$/ && $5~/^[0-9]+$/ {count++}
+  END {print count+0}
+' "${BEFORE_FILE}")
+if [[ "${MATCHED_ROWS}" -ne "${PARTITION_COUNT}" ]]; then
+  echo "ERROR: 快照只识别到 ${MATCHED_ROWS}/${PARTITION_COUNT} 个目标分区，拒绝继续" >&2
+  exit 1
+fi
+
+if awk -v topic="${RAW_TOPIC}" '
+  $2==topic && $3~/^[0-9]+$/ && $7!="-" {active=1}
+  END {exit(active ? 0 : 1)}
+' "${BEFORE_FILE}"; then
+  echo "ERROR: 消费组 ${GROUP_ID} 仍有在线成员。请先停止全部 syslog-consumer-rule 实例。" >&2
+  exit 1
+fi
+
+CURRENT_LAG=$(awk -v topic="${RAW_TOPIC}" '
+  $2==topic && $6~/^[0-9]+$/ {sum+=$6}
+  END {printf "%.0f", sum+0}
+' "${BEFORE_FILE}")
+echo "当前积压：${CURRENT_LAG} 条"
+
+echo "[3/6] 导出原 offset 回退文件"
+"${CONSUMER_GROUPS_CMD}" "${COMMON_ARGS[@]}" \
+  --group "${GROUP_ID}" \
+  --topic "${RAW_TOPIC}" \
+  --reset-offsets \
+  --to-current \
+  --dry-run \
+  --export > "${RESTORE_FILE}"
+
+RESTORE_ROWS=$(awk -F, -v topic="${RAW_TOPIC}" '
+  $1==topic && $2~/^[0-9]+$/ && $3~/^[0-9]+$/ {count++}
+  END {print count+0}
+' "${RESTORE_FILE}")
+if [[ "${RESTORE_ROWS}" -ne "${PARTITION_COUNT}" ]]; then
+  echo "ERROR: 回退文件只包含 ${RESTORE_ROWS}/${PARTITION_COUNT} 个分区，拒绝继续" >&2
+  exit 1
+fi
+echo "原 offset 已保存：${RESTORE_FILE}"
+
+echo "[4/6] 预演按时间重置"
+"${CONSUMER_GROUPS_CMD}" "${COMMON_ARGS[@]}" \
+  --group "${GROUP_ID}" \
+  --topic "${RAW_TOPIC}" \
+  --reset-offsets \
+  --to-datetime "${CUTOFF_DATETIME}" \
+  --dry-run | tee "${DRY_RUN_FILE}"
+
+PLAN_ROWS=$(awk -v topic="${RAW_TOPIC}" '
+  $1==topic && $2~/^[0-9]+$/ && $3~/^[0-9]+$/ {count++; next}
+  $2==topic && $3~/^[0-9]+$/ && $4~/^[0-9]+$/ {count++}
+  END {print count+0}
+' "${DRY_RUN_FILE}")
+if [[ "${PLAN_ROWS}" -ne "${PARTITION_COUNT}" ]]; then
+  echo "ERROR: dry-run 只生成 ${PLAN_ROWS}/${PARTITION_COUNT} 个分区，拒绝执行" >&2
+  exit 1
+fi
+
+EXPECTED_CONFIRM="RESET:${GROUP_ID}:${RAW_TOPIC}:${CUTOFF_DATETIME}"
+echo "[5/6] 等待人工确认"
+echo "即将永久跳过 ${CUTOFF_DATETIME} (${KAFKA_TIMEZONE}) 之前尚未消费的消息。"
+echo "原 offset 回退文件：${RESTORE_FILE}"
+echo "如确认执行，请完整输入：${EXPECTED_CONFIRM}"
+
+if [[ -n "${CONFIRM_RESET:-}" ]]; then
+  CONFIRM_INPUT="${CONFIRM_RESET}"
+elif [[ -t 0 ]]; then
+  read -r CONFIRM_INPUT
+else
+  echo "ERROR: 当前不是交互终端；必须通过 CONFIRM_RESET 提供精确确认串" >&2
+  exit 1
+fi
+
+if [[ "${CONFIRM_INPUT}" != "${EXPECTED_CONFIRM}" ]]; then
+  echo "CANCELLED: 确认串不匹配，未修改任何 offset。" >&2
+  exit 1
+fi
+
+echo "[6/6] 执行 offset 重置"
+"${CONSUMER_GROUPS_CMD}" "${COMMON_ARGS[@]}" \
+  --group "${GROUP_ID}" \
+  --topic "${RAW_TOPIC}" \
+  --reset-offsets \
+  --to-datetime "${CUTOFF_DATETIME}" \
+  --execute | tee "${EXECUTE_FILE}"
+
+echo "保存执行后快照"
+"${CONSUMER_GROUPS_CMD}" "${COMMON_ARGS[@]}" \
+  --describe --group "${GROUP_ID}" 2>&1 | tee "${AFTER_FILE}"
+
+echo "PASS: 已将 ${GROUP_ID}/${RAW_TOPIC} 重置到 ${CUTOFF_DATETIME} (${KAFKA_TIMEZONE})。"
+echo "未删除 Topic；现在可以启动 syslog-consumer-rule 并检查 LAG。"
+echo "回退文件：${RESTORE_FILE}（仅在 Kafka 原消息仍处于 retention 内时有效）。"
@@ -0,0 +1,88 @@
+#!/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}"
+
+SUFFIXES=("RETRY.1M" "RETRY.10M" "RETRY.1H" "DLT" "REPLAY")
+RETENTIONS=(259200000 604800000 604800000 7776000000 604800000)
+
+echo "[3/4] 创建或更新升级 Topic"
+for i in 0 1 2 3 4; do
+  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。"
@@ -7,7 +7,8 @@ set -euo pipefail

 INFLUX_CONFIG_PATH="${INFLUX_CONFIG_PATH:-/etc/influxdb2/influx-configs}"
 V2_BUCKET="${V2_BUCKET:-yelangbucket_v2}"
-RETENTION="${RETENTION:-720h}"
+# 现场要求原始日志保留 180 天；仍可通过 RETENTION 环境变量覆盖。
+RETENTION="${RETENTION:-180d}"
 SHARD_GROUP_DURATION="${SHARD_GROUP_DURATION:-24h}"

 command -v docker >/dev/null 2>&1 || {