fix(syslog-consumer-rule): 解耦统计物化并解决多实例达梦死锁

- 将设备、规则和采集任务业务统计改为专用 Kafka Topic 异步物化 - 使用维度 ID 作为 Kafka key，保证同一统计维度串行更新 - 物化过程读取小型统计表精确值，确保重复消费不会重复计数 - 增加提交后发布、失败重试及定时小表对账补偿 - 固定日统计聚合键写入顺序，完善达梦死锁识别和事务重试 - JSON 解析失败直接转为非标日志，避免进入 DLT 重试链路 - 非标日志使用独立异步队列，不再阻塞正常消息 Kafka ACK - Kafka 升级脚本新增 STATS.MATERIALIZE Topic - 增加统计物化状态指标、配置示例及单元测试
2026-09-11 20:42:22 +08:00
parent 2def445537
commit c1472b9a95
24 changed files with 1123 additions and 55 deletions
@@ -9,6 +9,7 @@ import com.config.AppProperties;
 import com.config.KafkaConsumerProperties;
 import com.common.service.DmStandardizedBatchWriter;
 import com.common.service.IngressStatsProjectionWriter;
+import com.common.service.DirectStatsProjectionWriter;
 import com.common.service.SyslogNonNormalMessageService;
 import com.common.entity.IngressStatsEvent;
 import com.common.entity.SyslogNonNormalMessage;
@@ -80,6 +81,12 @@ public class SysLogProcessor {
    @Qualifier("nonNormalWriterExecutor")
    private Executor nonNormalWriterExecutor;

+    @Value("${dm.non-normal-write.max-concurrency-retries:3}")
+    private int nonNormalMaxConcurrencyRetries;
+
+    @Value("${dm.non-normal-write.concurrency-retry-backoff-ms:50}")
+    private long nonNormalConcurrencyRetryBackoffMs;
+
    @Autowired
    private IngressStatsProjectionWriter ingressStatsProjectionWriter;

@@ -585,7 +592,7 @@ public class SysLogProcessor {
            nonNormalWriterExecutor.execute(() -> {
                long startedAt = System.nanoTime();
                try {
-                    int inserted = nonNormalMessageService.saveBatchMessages(batch);
+                    int inserted = saveNonNormalBatchWithConcurrencyRetry(batch);
                    if (inserted != batch.size()) {
                        nonNormalAsyncFailedTotal.addAndGet(batch.size());
                        log.error("非标日志异步批写结果异常，Kafka offset 不回退: rows={}, inserted={}, "
@@ -615,6 +622,36 @@ public class SysLogProcessor {
        }
    }

+    /**
+     * 每次 service 调用都经过 Spring 事务代理；达梦死锁会先整体回滚，随后用同一批 ID
+     * 开启新事务有限重试。该等待只占用独立非标线程，不进入 Kafka ACK 边界。
+     */
+    private int saveNonNormalBatchWithConcurrencyRetry(List<SyslogNonNormalMessage> batch) {
+        int attempt = 0;
+        while (true) {
+            try {
+                return nonNormalMessageService.saveBatchMessages(batch);
+            } catch (RuntimeException failure) {
+                if (!DirectStatsProjectionWriter.isConcurrentFailure(failure)
+                        || attempt >= Math.max(0, nonNormalMaxConcurrencyRetries)) {
+                    throw failure;
+                }
+                attempt++;
+                long delayMs = Math.max(0L, nonNormalConcurrencyRetryBackoffMs) * attempt;
+                log.warn("非标日志批写遇到达梦并发冲突，准备重试: rows={}, attempt={}/{}, backoffMs={}",
+                        batch.size(), attempt, Math.max(0, nonNormalMaxConcurrencyRetries), delayMs);
+                if (delayMs > 0L) {
+                    try {
+                        Thread.sleep(delayMs);
+                    } catch (InterruptedException interrupted) {
+                        Thread.currentThread().interrupt();
+                        throw new IllegalStateException("非标日志死锁重试等待被中断", interrupted);
+                    }
+                }
+            }
+        }
+    }
+
    private static long elapsedMillis(long startedAtNanos) {
        return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedAtNanos);
    }
@@ -69,8 +69,19 @@ public final class CompiledRule {
    }

    public ParsedMessage parse(String message) {
-        String cropped = logNormalData.ParserMessageJsonType(message,
+        String cropped;
+        try {
+            cropped = logNormalData.ParserMessageJsonType(message,
                    cropper == null ? null : cropper.toLegacyParams());
+        } catch (IllegalArgumentException invalidJsonContent) {
+            // JSON 截取器在消息中没有完整 {...} 时会直接抛异常。对 JSON 规则而言，
+            // 这与 Jackson 解析失败属于同一个确定性业务结果，由上层直接转非标，
+            // 不能冒泡成系统异常后进入 Retry/DLT。
+            if ("json".equals(dataType)) {
+                return null;
+            }
+            throw invalidJsonContent;
+        }
        LinkedHashMap<String, Object> parsed;
        switch (dataType) {
            case "json":
@@ -11,6 +11,7 @@ import org.springframework.stereotype.Service;
 import org.springframework.transaction.support.TransactionSynchronizationManager;

 import java.sql.Date;
+import java.sql.SQLException;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.ZoneId;
@@ -61,6 +62,7 @@ public class DirectStatsProjectionWriter {
    private final JdbcTemplate jdbc;
    private final StatsWriteModeProvider modeProvider;
    private final DirectStatsWriteMetrics metrics;
+    private final StatsMaterializationPublisher materializationPublisher;

    @Value("${stats.incremental.read-enabled:false}")
    private boolean readEnabled;
@@ -88,10 +90,12 @@ public class DirectStatsProjectionWriter {

    public DirectStatsProjectionWriter(JdbcTemplate jdbc,
                                       StatsWriteModeProvider modeProvider,
-                                       DirectStatsWriteMetrics metrics) {
+                                       DirectStatsWriteMetrics metrics,
+                                       StatsMaterializationPublisher materializationPublisher) {
        this.jdbc = jdbc;
        this.modeProvider = modeProvider;
        this.metrics = metrics;
+        this.materializationPublisher = materializationPublisher;
    }

    public WriteResult writeStandardizedRows(List<Map<String, Object>> rows) {
@@ -154,6 +158,8 @@ public class DirectStatsProjectionWriter {
        status.put("device_table", selectedDeviceTable());
        status.put("rule_table", selectedRuleTable());
        status.put("today_stats_date_enabled", todayStatsDateEnabled);
+        status.put("business_materialization_mode", asynchronousMaterialization()
+                ? "KAFKA_DIRTY_KEY" : "SYNCHRONOUS");
        return status;
    }

@@ -202,8 +208,12 @@ public class DirectStatsProjectionWriter {

            if (modeProvider.materializesBusinessFields() && readEnabled) {
                long materializeStarted = System.nanoTime();
+                if (asynchronousMaterialization()) {
+                    materializationPublisher.enqueueAfterCommit(materializationCommands(devices, rules));
+                } else {
                    materializeDevices(devices);
                    materializeRules(rules);
+                }
                materializeMs = elapsedMs(materializeStarted);
            }
            metrics.recordSuccess(accumulator.getInputRows(), devices.size(), rules.size(),
@@ -216,7 +226,7 @@ public class DirectStatsProjectionWriter {
                    aggregateMs, mergeMs, materializeMs);
        } catch (RuntimeException failure) {
            metrics.recordFailure();
-            if (isConcurrentWrite(failure)) {
+            if (isConcurrentFailure(failure)) {
                throw new ConcurrentStatsWriteException("直接统计聚合键并发写入失败，可安全回滚后有限重试", failure);
            }
            throw new DirectStatsWriteException("直接统计写入失败，事实事务必须整体回滚", failure);
@@ -261,7 +271,11 @@ public class DirectStatsProjectionWriter {
            long materializeMs = 0L;
            if (modeProvider.materializesBusinessFields() && readEnabled) {
                long materializeStarted = System.nanoTime();
+                if (asynchronousMaterialization()) {
+                    materializationPublisher.enqueueAfterCommit(deviceMaterializationCommands(keys));
+                } else {
                    materializeDeviceKeys(keys);
+                }
                materializeMs = elapsedMs(materializeStarted);
            }
            metrics.recordSuccess(accumulator.getInputRows(), keys.size(), 0,
@@ -270,7 +284,7 @@ public class DirectStatsProjectionWriter {
                    aggregateMs, mergeMs, materializeMs);
        } catch (RuntimeException failure) {
            metrics.recordFailure();
-            if (isConcurrentWrite(failure)) {
+            if (isConcurrentFailure(failure)) {
                throw new ConcurrentStatsWriteException("直接统计局部分组并发写入失败，可安全回滚后有限重试", failure);
            }
            throw new DirectStatsWriteException("直接统计局部分组重算失败，事实事务必须整体回滚", failure);
@@ -389,6 +403,41 @@ public class DirectStatsProjectionWriter {
                        "WHERE id=? AND del_flag='0'", parameters);
    }

+    private boolean asynchronousMaterialization() {
+        return modeProvider.materializesBusinessFields() && readEnabled
+                && materializationPublisher.isEnabled();
+    }
+
+    private List<StatsMaterializationCommand> materializationCommands(
+            List<Map.Entry<DirectStatsAccumulator.DailyKey, DirectStatsAccumulator.DeviceDelta>> devices,
+            List<Map.Entry<DirectStatsAccumulator.DailyKey, DirectStatsAccumulator.RuleDelta>> rules) {
+        List<StatsMaterializationCommand> commands = new ArrayList<>(devices.size() + rules.size());
+        LocalDate today = LocalDate.now(ZoneId.of(businessZone));
+        for (Map.Entry<DirectStatsAccumulator.DailyKey, DirectStatsAccumulator.DeviceDelta> entry : devices) {
+            DirectStatsAccumulator.DailyKey key = entry.getKey();
+            if (today.equals(key.getDate())) {
+                commands.add(StatsMaterializationCommand.device(key.getDate(), key.getDimensionId()));
+            }
+        }
+        for (Map.Entry<DirectStatsAccumulator.DailyKey, DirectStatsAccumulator.RuleDelta> entry : rules) {
+            DirectStatsAccumulator.DailyKey key = entry.getKey();
+            commands.add(StatsMaterializationCommand.rule(key.getDate(), key.getDimensionId()));
+        }
+        return commands;
+    }
+
+    private List<StatsMaterializationCommand> deviceMaterializationCommands(
+            List<DirectStatsAccumulator.DailyKey> keys) {
+        List<StatsMaterializationCommand> commands = new ArrayList<>(keys.size());
+        LocalDate today = LocalDate.now(ZoneId.of(businessZone));
+        for (DirectStatsAccumulator.DailyKey key : keys) {
+            if (today.equals(key.getDate())) {
+                commands.add(StatsMaterializationCommand.device(key.getDate(), key.getDimensionId()));
+            }
+        }
+        return commands;
+    }
+
    private String selectedDeviceTable() {
        return modeProvider.writesShadowProjection() ? shadowDeviceTable : directDeviceTable;
    }
@@ -410,7 +459,12 @@ public class DirectStatsProjectionWriter {
        }
    }

-    private static boolean isConcurrentWrite(Throwable failure) {
+    /**
+     * 识别 Spring 与不同 JDBC 驱动对并发冲突的各种包装形式。
+     * 达梦可能把死锁翻译成 DataIntegrityViolationException，并仅在内层
+     * BatchUpdateException 的 SQLState/消息中保留 Deadlock 信息。
+     */
+    public static boolean isConcurrentFailure(Throwable failure) {
        Throwable current = failure;
        while (current != null) {
            if (current instanceof DuplicateKeyException
@@ -418,6 +472,18 @@ public class DirectStatsProjectionWriter {
                    || current instanceof DeadlockLoserDataAccessException) {
                return true;
            }
+            if (current instanceof SQLException) {
+                String sqlState = ((SQLException) current).getSQLState();
+                if (sqlState != null && sqlState.startsWith("40")) {
+                    return true;
+                }
+            }
+            String message = String.valueOf(current.getMessage())
+                    .toLowerCase(java.util.Locale.ROOT);
+            if (message.contains("deadlock") || message.contains("死锁")
+                    || message.contains("unique constraint") || message.contains("duplicate key")) {
+                return true;
+            }
            current = current.getCause();
        }
        return false;
@@ -2,9 +2,6 @@ package com.common.service;

 import com.common.entity.IngressStatsEvent;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.dao.DuplicateKeyException;
-import org.springframework.dao.CannotAcquireLockException;
-import org.springframework.dao.DeadlockLoserDataAccessException;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.PlatformTransactionManager;
@@ -15,6 +12,7 @@ import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.util.Comparator;
+import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -30,6 +28,7 @@ public class IngressStatsProjectionWriter {
    private final JdbcTemplate jdbc;
    private final PlatformTransactionManager transactionManager;
    private final StatsWriteModeProvider modeProvider;
+    private final StatsMaterializationPublisher materializationPublisher;

    @Value("${stats.ingress-metadata.enabled:false}")
    private boolean enabled;
@@ -50,10 +49,12 @@ public class IngressStatsProjectionWriter {

    public IngressStatsProjectionWriter(JdbcTemplate jdbc,
                                        PlatformTransactionManager transactionManager,
-                                        StatsWriteModeProvider modeProvider) {
+                                        StatsWriteModeProvider modeProvider,
+                                        StatsMaterializationPublisher materializationPublisher) {
        this.jdbc = jdbc;
        this.transactionManager = transactionManager;
        this.modeProvider = modeProvider;
+        this.materializationPublisher = materializationPublisher;
    }

    public int writeBatch(List<IngressStatsEvent> events) {
@@ -89,6 +90,8 @@ public class IngressStatsProjectionWriter {
        status.put("read_enabled", readEnabled);
        status.put("target_device_table", modeProvider.writesShadowProjection()
                ? "device_daily_stats+" + shadowDeviceTable : "device_daily_stats");
+        status.put("business_materialization_mode", asynchronousMaterialization()
+                ? "KAFKA_DIRTY_KEY" : "SYNCHRONOUS");
        status.put("successful_batches", successfulBatches.get());
        status.put("received_events", receivedEvents.get());
        status.put("updated_groups", updatedGroups.get());
@@ -99,9 +102,7 @@ public class IngressStatsProjectionWriter {
    }

    private static boolean retryableConcurrentWrite(RuntimeException failure) {
-        return failure instanceof DuplicateKeyException
-                || failure instanceof CannotAcquireLockException
-                || failure instanceof DeadlockLoserDataAccessException;
+        return DirectStatsProjectionWriter.isConcurrentFailure(failure);
    }

    private int writeAggregation(Aggregation aggregation) {
@@ -109,6 +110,11 @@ public class IngressStatsProjectionWriter {
        transaction.setTimeout(Math.max(1, transactionTimeoutSeconds));
        Integer updated = transaction.execute(status -> {
            int groups = 0;
+            boolean asyncMaterialization = asynchronousMaterialization();
+            List<StatsMaterializationCommand> commands = asyncMaterialization
+                    ? new ArrayList<>(aggregation.devices.size() + aggregation.collects.size())
+                    : null;
+            LocalDate today = LocalDate.now(ZoneId.of(businessZone));
            for (Map.Entry<DeviceKey, LocalDateTime> entry : aggregation.devices.entrySet()) {
                DeviceKey key = entry.getKey();
                LocalDateTime receiveTime = entry.getValue();
@@ -118,12 +124,16 @@ public class IngressStatsProjectionWriter {
                if (modeProvider.writesShadowProjection()) {
                    mergeDeviceReceive(safeTable(shadowDeviceTable), key, receiveTime);
                }
-                if (readEnabled && LocalDate.now(ZoneId.of(businessZone)).equals(key.date)) {
+                if (readEnabled && today.equals(key.date)) {
+                    if (asyncMaterialization) {
+                        commands.add(StatsMaterializationCommand.device(key.date, key.deviceId));
+                    } else {
                        jdbc.update("UPDATE device_device SET last_receive_time=CASE " +
                                        "WHEN last_receive_time IS NULL OR last_receive_time<? THEN ? ELSE last_receive_time END," +
                                        "updated_at=SYSDATE WHERE id=?",
                                receiveTime, receiveTime, key.deviceId);
                    }
+                }
                groups++;
            }
            for (Map.Entry<Long, CollectTimes> entry : aggregation.collects.entrySet()) {
@@ -142,14 +152,21 @@ public class IngressStatsProjectionWriter {
                                "(s.device_collect_id,s.first_success_time,s.last_success_time,NULL,0,SYSDATE)",
                        entry.getKey(), times.min, times.max);
                if (readEnabled) {
+                    if (asyncMaterialization) {
+                        commands.add(StatsMaterializationCommand.collect(entry.getKey()));
+                    } else {
                        jdbc.update("UPDATE device_collect_task SET " +
                                        "first_time=CASE WHEN first_time IS NULL OR first_time>? THEN ? ELSE first_time END," +
                                        "last_success_time=CASE WHEN last_success_time IS NULL OR last_success_time<? " +
                                        "THEN ? ELSE last_success_time END,updated_at=SYSDATE WHERE id=?",
                                times.min, times.min, times.max, times.max, entry.getKey());
                    }
+                }
                groups++;
            }
+            if (asyncMaterialization) {
+                materializationPublisher.enqueueAfterCommit(commands);
+            }
            return groups;
        });
        return updated == null ? 0 : updated;
@@ -215,6 +232,11 @@ public class IngressStatsProjectionWriter {
                Date.valueOf(key.date), key.deviceId, receiveTime);
    }

+    private boolean asynchronousMaterialization() {
+        return modeProvider.materializesBusinessFields() && readEnabled
+                && materializationPublisher.isEnabled();
+    }
+
    private static String safeTable(String table) {
        if (table == null || !table.matches("[A-Za-z][A-Za-z0-9_]*")) {
            throw new IllegalStateException("非法入口设备统计表名: " + table);
@@ -10,9 +10,6 @@ import java.util.Objects;
 import java.util.Set;
 import java.util.UUID;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.dao.CannotAcquireLockException;
-import org.springframework.dao.DeadlockLoserDataAccessException;
-import org.springframework.dao.DuplicateKeyException;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Component;

@@ -116,20 +113,7 @@ public class LateAlarmBucketWriter {
  }

  private boolean isConcurrentWrite(Throwable failure) {
-    Throwable current = failure;
-    while (current != null) {
-      if (current instanceof DuplicateKeyException
-          || current instanceof CannotAcquireLockException
-          || current instanceof DeadlockLoserDataAccessException) {
-        return true;
-      }
-      String message = String.valueOf(current.getMessage()).toLowerCase(java.util.Locale.ROOT);
-      if (message.contains("unique constraint") || message.contains("duplicate key")) {
-        return true;
-      }
-      current = current.getCause();
-    }
-    return false;
+    return DirectStatsProjectionWriter.isConcurrentFailure(failure);
  }

  private static final class BucketKey {
@@ -0,0 +1,63 @@
+package com.common.service;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+
+import java.time.LocalDate;
+import java.util.Objects;
+
+/** 统计物化脏键；通知只要求复制当前精确值，不携带会被重复累加的 delta。 */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class StatsMaterializationCommand {
+    public static final String SCHEMA = "SYSLOG_STATS_MATERIALIZE_V1";
+    public enum Type { DEVICE, RULE, COLLECT }
+
+    private String schema = SCHEMA;
+    private Type type;
+    private long dimensionId;
+    private String statDate;
+
+    public StatsMaterializationCommand() {}
+
+    private StatsMaterializationCommand(Type type, long dimensionId, LocalDate statDate) {
+        this.type = Objects.requireNonNull(type, "type");
+        this.dimensionId = dimensionId;
+        this.statDate = statDate == null ? null : statDate.toString();
+    }
+
+    public static StatsMaterializationCommand device(LocalDate date, long id) {
+        return new StatsMaterializationCommand(Type.DEVICE, id, date);
+    }
+    public static StatsMaterializationCommand rule(LocalDate date, long id) {
+        return new StatsMaterializationCommand(Type.RULE, id, date);
+    }
+    public static StatsMaterializationCommand collect(long id) {
+        return new StatsMaterializationCommand(Type.COLLECT, id, null);
+    }
+
+    public void validate() {
+        if (!SCHEMA.equals(schema)) throw new IllegalArgumentException("不支持的统计物化协议: " + schema);
+        if (type == null || dimensionId <= 0L) throw new IllegalArgumentException("统计物化命令缺少合法维度");
+        if (type != Type.COLLECT) date();
+    }
+
+    public String key() {
+        validate();
+        return type.name() + ":" + dimensionId;
+    }
+
+    public LocalDate date() {
+        if (statDate == null || statDate.trim().isEmpty()) {
+            throw new IllegalArgumentException("统计物化命令缺少 statDate");
+        }
+        return LocalDate.parse(statDate);
+    }
+
+    public String getSchema() { return schema; }
+    public void setSchema(String schema) { this.schema = schema; }
+    public Type getType() { return type; }
+    public void setType(Type type) { this.type = type; }
+    public long getDimensionId() { return dimensionId; }
+    public void setDimensionId(long dimensionId) { this.dimensionId = dimensionId; }
+    public String getStatDate() { return statDate; }
+    public void setStatDate(String statDate) { this.statDate = statDate; }
+}
@@ -0,0 +1,255 @@
+package com.common.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import java.sql.Date;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * 消费统计脏键并从小型聚合表复制精确值到业务表。
+ *
+ * <p>Topic key 是维度类型和 ID；Kafka 保证同一维度只落在一个分区，消费组保证同一时刻只由
+ * 一个实例处理该分区。命令不携带 delta，因此重复投递和故障重放不会造成重复计数。</p>
+ */
+@Slf4j
+@Component
+public class StatsMaterializationConsumer {
+    private static final String DEVICE_MATERIALIZE_WITH_DATE =
+            "MERGE INTO device_device d USING (SELECT stat_date,device_id,parse_count,non_log_count," +
+            "last_receive_time FROM %s WHERE stat_date=? AND device_id=?) s " +
+            "ON (d.id=s.device_id) WHEN MATCHED THEN UPDATE SET " +
+            "d.today_parse_count=s.parse_count,d.today_non_log_count=s.non_log_count," +
+            "d.today_stats_date=s.stat_date,d.last_receive_time=CASE " +
+            "WHEN s.last_receive_time IS NULL THEN d.last_receive_time " +
+            "WHEN d.last_receive_time IS NULL OR d.last_receive_time<s.last_receive_time " +
+            "THEN s.last_receive_time ELSE d.last_receive_time END,d.updated_at=SYSDATE";
+
+    private static final String DEVICE_MATERIALIZE_WITHOUT_DATE =
+            "MERGE INTO device_device d USING (SELECT stat_date,device_id,parse_count,non_log_count," +
+            "last_receive_time FROM %s WHERE stat_date=? AND device_id=?) s " +
+            "ON (d.id=s.device_id) WHEN MATCHED THEN UPDATE SET " +
+            "d.today_parse_count=s.parse_count,d.today_non_log_count=s.non_log_count," +
+            "d.last_receive_time=CASE WHEN s.last_receive_time IS NULL THEN d.last_receive_time " +
+            "WHEN d.last_receive_time IS NULL OR d.last_receive_time<s.last_receive_time " +
+            "THEN s.last_receive_time ELSE d.last_receive_time END,d.updated_at=SYSDATE";
+
+    private static final String RULE_MATERIALIZE =
+            "MERGE INTO dm_normalize_rule r USING (SELECT normalize_rule_id," +
+            "MAX(max_log_time) max_log_time FROM %s WHERE normalize_rule_id=? " +
+            "GROUP BY normalize_rule_id) s ON (r.id=s.normalize_rule_id AND r.del_flag='0') " +
+            "WHEN MATCHED THEN UPDATE SET r.first_data_saved_at=CASE " +
+            "WHEN s.max_log_time IS NULL THEN r.first_data_saved_at " +
+            "WHEN r.first_data_saved_at IS NULL OR r.first_data_saved_at<s.max_log_time " +
+            "THEN s.max_log_time ELSE r.first_data_saved_at END,r.updated_at=SYSDATE";
+
+    private static final String COLLECT_MATERIALIZE =
+            "MERGE INTO device_collect_task c USING (SELECT device_collect_id,first_success_time," +
+            "last_success_time FROM device_collect_stats WHERE device_collect_id=?) s " +
+            "ON (c.id=s.device_collect_id) WHEN MATCHED THEN UPDATE SET " +
+            "c.first_time=CASE WHEN s.first_success_time IS NULL THEN c.first_time " +
+            "WHEN c.first_time IS NULL OR c.first_time>s.first_success_time " +
+            "THEN s.first_success_time ELSE c.first_time END," +
+            "c.last_success_time=CASE WHEN s.last_success_time IS NULL THEN c.last_success_time " +
+            "WHEN c.last_success_time IS NULL OR c.last_success_time<s.last_success_time " +
+            "THEN s.last_success_time ELSE c.last_success_time END,c.updated_at=SYSDATE";
+
+    private final JdbcTemplate jdbc;
+    private final PlatformTransactionManager transactionManager;
+    private final ObjectMapper objectMapper;
+    private final StatsWriteModeProvider modeProvider;
+
+    @Value("${stats.incremental.read-enabled:false}") private boolean readEnabled;
+    @Value("${stats.direct.device-table:device_daily_stats}") private String deviceTable;
+    @Value("${stats.direct.rule-table:normalize_rule_daily_stats}") private String ruleTable;
+    @Value("${stats.direct.today-stats-date-enabled:true}") private boolean todayStatsDateEnabled;
+    @Value("${stats.business-zone:Asia/Shanghai}") private String businessZone;
+    @Value("${stats.incremental.transaction-timeout-seconds:45}") private int transactionTimeoutSeconds;
+    @Value("${stats.materialization.max-concurrency-retries:3}") private int maxConcurrencyRetries;
+    @Value("${stats.materialization.concurrency-retry-backoff-ms:50}") private long retryBackoffMs;
+
+    private final AtomicLong consumed = new AtomicLong();
+    private final AtomicLong invalid = new AtomicLong();
+    private final AtomicLong applied = new AtomicLong();
+    private final AtomicLong successfulBatches = new AtomicLong();
+    private final AtomicLong failures = new AtomicLong();
+    private final AtomicLong retries = new AtomicLong();
+    private volatile LocalDateTime lastSuccessTime;
+    private volatile LocalDateTime lastFailureTime;
+
+    public StatsMaterializationConsumer(JdbcTemplate jdbc,
+                                        PlatformTransactionManager transactionManager,
+                                        ObjectMapper objectMapper,
+                                        StatsWriteModeProvider modeProvider) {
+        this.jdbc = jdbc;
+        this.transactionManager = transactionManager;
+        this.objectMapper = objectMapper.copy();
+        this.objectMapper.deactivateDefaultTyping();
+        this.modeProvider = modeProvider;
+    }
+
+    @KafkaListener(
+            id = "syslog-stats-materializer",
+            groupId = "${stats.materialization.group-id:${spring.kafka.consumer.group-id}-stats-materialize}",
+            topics = "${stats.materialization.topic:${spring.kafka.consumer.topic}.STATS.MATERIALIZE}",
+            autoStartup = "${stats.materialization.enabled:true}",
+            containerFactory = "statsMaterializationKafkaListenerContainerFactory"
+    )
+    public void consume(List<ConsumerRecord<String, String>> records, Acknowledgment acknowledgment) {
+        if (records == null || records.isEmpty()) {
+            acknowledgment.acknowledge();
+            return;
+        }
+        consumed.addAndGet(records.size());
+        Map<String, StatsMaterializationCommand> commands = parseAndCoalesce(records);
+        if (commands.isEmpty()) {
+            acknowledgment.acknowledge();
+            return;
+        }
+        try {
+            int count = materializeWithRetry(commands.values());
+            acknowledgment.acknowledge();
+            applied.addAndGet(count);
+            successfulBatches.incrementAndGet();
+            lastSuccessTime = LocalDateTime.now();
+            log.info("统计物化批次完成: received={}, coalesced={}, applied={}, invalidTotal={}",
+                    records.size(), commands.size(), count, invalid.get());
+        } catch (RuntimeException failure) {
+            failures.incrementAndGet();
+            lastFailureTime = LocalDateTime.now();
+            throw failure;
+        }
+    }
+
+    private Map<String, StatsMaterializationCommand> parseAndCoalesce(
+            List<ConsumerRecord<String, String>> records) {
+        Map<String, StatsMaterializationCommand> result = new TreeMap<>();
+        for (ConsumerRecord<String, String> record : records) {
+            try {
+                StatsMaterializationCommand command = objectMapper.readValue(
+                        record.value(), StatsMaterializationCommand.class);
+                command.validate();
+                String key = command.key();
+                if (record.key() == null || !key.equals(record.key())) {
+                    throw new IllegalArgumentException("Kafka key 与统计物化命令不一致");
+                }
+                result.put(key, command);
+            } catch (Exception badCommand) {
+                invalid.incrementAndGet();
+                log.error("忽略非法内部统计物化命令: topic={}, partition={}, offset={}, key={}",
+                        record.topic(), record.partition(), record.offset(), record.key(), badCommand);
+            }
+        }
+        return result;
+    }
+
+    int materializeWithRetry(Collection<StatsMaterializationCommand> commands) {
+        int attempt = 0;
+        while (true) {
+            try {
+                return materializeOnce(commands);
+            } catch (RuntimeException failure) {
+                if (!DirectStatsProjectionWriter.isConcurrentFailure(failure)
+                        || attempt >= Math.max(0, maxConcurrencyRetries)) {
+                    throw failure;
+                }
+                attempt++;
+                retries.incrementAndGet();
+                sleepBeforeRetry(attempt);
+            }
+        }
+    }
+
+    private int materializeOnce(Collection<StatsMaterializationCommand> commands) {
+        if (!modeProvider.materializesBusinessFields() || !readEnabled) {
+            throw new IllegalStateException("统计物化消费者只允许在 DIRECT 且 read-enabled 模式运行");
+        }
+        final LocalDate today = LocalDate.now(ZoneId.of(businessZone));
+        final List<Object[]> deviceParams = new ArrayList<>();
+        final List<Object[]> ruleParams = new ArrayList<>();
+        final List<Object[]> collectParams = new ArrayList<>();
+        for (StatsMaterializationCommand command : commands) {
+            command.validate();
+            switch (command.getType()) {
+                case DEVICE:
+                    // yesterday 的迟到事实只保存在日统计表，不允许覆盖今日展示字段。
+                    if (today.equals(command.date())) {
+                        deviceParams.add(new Object[]{Date.valueOf(today), command.getDimensionId()});
+                    }
+                    break;
+                case RULE:
+                    // 按 rule id 从所有日统计取 MAX，命令即使跨日期合并也不会丢失较大值。
+                    ruleParams.add(new Object[]{command.getDimensionId()});
+                    break;
+                case COLLECT:
+                    collectParams.add(new Object[]{command.getDimensionId()});
+                    break;
+                default:
+                    throw new IllegalArgumentException("未知统计物化命令类型: " + command.getType());
+            }
+        }
+        TransactionTemplate transaction = new TransactionTemplate(transactionManager);
+        transaction.setTimeout(Math.max(1, transactionTimeoutSeconds));
+        Integer count = transaction.execute(status -> {
+            if (!deviceParams.isEmpty()) {
+                String template = todayStatsDateEnabled
+                        ? DEVICE_MATERIALIZE_WITH_DATE : DEVICE_MATERIALIZE_WITHOUT_DATE;
+                jdbc.batchUpdate(String.format(template, safeTable(deviceTable)), deviceParams);
+            }
+            if (!ruleParams.isEmpty()) {
+                jdbc.batchUpdate(String.format(RULE_MATERIALIZE, safeTable(ruleTable)), ruleParams);
+            }
+            if (!collectParams.isEmpty()) {
+                jdbc.batchUpdate(COLLECT_MATERIALIZE, collectParams);
+            }
+            return deviceParams.size() + ruleParams.size() + collectParams.size();
+        });
+        return count == null ? 0 : count;
+    }
+
+    private void sleepBeforeRetry(int attempt) {
+        try {
+            Thread.sleep(Math.max(0L, retryBackoffMs) * attempt);
+        } catch (InterruptedException interrupted) {
+            Thread.currentThread().interrupt();
+            throw new IllegalStateException("统计物化死锁重试等待被中断", interrupted);
+        }
+    }
+
+    public Map<String, Object> status() {
+        Map<String, Object> result = new LinkedHashMap<>();
+        result.put("consumed_total", consumed.get());
+        result.put("invalid_total", invalid.get());
+        result.put("applied_total", applied.get());
+        result.put("successful_batches", successfulBatches.get());
+        result.put("failures_total", failures.get());
+        result.put("concurrency_retries_total", retries.get());
+        result.put("last_success_time", lastSuccessTime);
+        result.put("last_failure_time", lastFailureTime);
+        return result;
+    }
+
+    private static String safeTable(String table) {
+        if (table == null || !table.matches("[A-Za-z][A-Za-z0-9_]*")) {
+            throw new IllegalStateException("非法统计物化源表名: " + table);
+        }
+        return table;
+    }
+}
@@ -0,0 +1,128 @@
+package com.common.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.support.TransactionSynchronizationAdapter;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+import org.springframework.util.concurrent.ListenableFuture;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/** 事务提交后合并统计脏键并批量发布；失败的 key 留在内存等待下轮，小表校准负责崩溃窗口。 */
+@Slf4j
+@Component
+public class StatsMaterializationPublisher {
+    private final KafkaTemplate<String, String> kafkaTemplate;
+    private final ObjectMapper objectMapper;
+    private final ConcurrentMap<String, StatsMaterializationCommand> pending = new ConcurrentHashMap<>();
+    private final AtomicBoolean flushing = new AtomicBoolean();
+    private final AtomicLong enqueued = new AtomicLong();
+    private final AtomicLong published = new AtomicLong();
+    private final AtomicLong failures = new AtomicLong();
+    private volatile long lastSuccessAt;
+    private volatile long lastFailureAt;
+
+    @Value("${stats.materialization.enabled:true}") private boolean enabled;
+    @Value("${stats.materialization.topic:${spring.kafka.consumer.topic}.STATS.MATERIALIZE}") private String topic;
+    @Value("${stats.materialization.publish-timeout-ms:10000}") private long publishTimeoutMs;
+    @Value("${stats.materialization.publish-max-batch-size:1000}") private int maxBatchSize;
+
+    public StatsMaterializationPublisher(KafkaTemplate<String, String> kafkaTemplate, ObjectMapper objectMapper) {
+        this.kafkaTemplate = kafkaTemplate;
+        this.objectMapper = objectMapper.copy();
+        this.objectMapper.deactivateDefaultTyping();
+    }
+
+    public boolean isEnabled() { return enabled; }
+
+    public void enqueueAfterCommit(Collection<StatsMaterializationCommand> commands) {
+        if (!enabled || commands == null || commands.isEmpty()) return;
+        if (!TransactionSynchronizationManager.isSynchronizationActive()) {
+            throw new IllegalStateException("统计物化通知必须在 Spring 事务内登记");
+        }
+        List<StatsMaterializationCommand> snapshot = new ArrayList<>(commands);
+        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
+            @Override public void afterCommit() { enqueueNow(snapshot); }
+        });
+    }
+
+    public void enqueueNow(Collection<StatsMaterializationCommand> commands) {
+        if (!enabled || commands == null) return;
+        for (StatsMaterializationCommand command : commands) {
+            if (command == null) continue;
+            pending.put(command.key(), command);
+            enqueued.incrementAndGet();
+        }
+    }
+
+    @Scheduled(fixedDelayString = "${stats.materialization.publish-interval-ms:200}")
+    public void flush() {
+        if (!enabled || pending.isEmpty() || !flushing.compareAndSet(false, true)) return;
+        try {
+            List<StatsMaterializationCommand> snapshot = new ArrayList<>(pending.values());
+            snapshot.sort(Comparator.comparing(StatsMaterializationCommand::key));
+            int size = Math.min(snapshot.size(), Math.max(1, maxBatchSize));
+            List<PendingSend> sends = new ArrayList<>(size);
+            for (int i = 0; i < size; i++) {
+                StatsMaterializationCommand command = snapshot.get(i);
+                sends.add(new PendingSend(command, kafkaTemplate.send(
+                        topic, command.key(), objectMapper.writeValueAsString(command))));
+            }
+            long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(Math.max(1L, publishTimeoutMs));
+            for (PendingSend send : sends) {
+                long remaining = deadline - System.nanoTime();
+                if (remaining <= 0L) throw new java.util.concurrent.TimeoutException("统计物化通知发布超时");
+                send.future.get(remaining, TimeUnit.NANOSECONDS);
+                if (pending.remove(send.command.key(), send.command)) published.incrementAndGet();
+            }
+            lastSuccessAt = System.currentTimeMillis();
+        } catch (InterruptedException interrupted) {
+            Thread.currentThread().interrupt();
+            recordFailure(interrupted);
+        } catch (Exception failure) {
+            recordFailure(failure);
+        } finally {
+            flushing.set(false);
+        }
+    }
+
+    private void recordFailure(Exception failure) {
+        failures.incrementAndGet();
+        lastFailureAt = System.currentTimeMillis();
+        log.error("统计物化通知发布失败，脏键保留等待重试: pending={}", pending.size(), failure);
+    }
+
+    public Map<String, Object> status() {
+        Map<String, Object> result = new LinkedHashMap<>();
+        result.put("enabled", enabled); result.put("topic", topic);
+        result.put("pending_keys", pending.size()); result.put("enqueued_total", enqueued.get());
+        result.put("published_total", published.get()); result.put("publish_failures_total", failures.get());
+        result.put("last_success_epoch_ms", lastSuccessAt == 0L ? null : lastSuccessAt);
+        result.put("last_failure_epoch_ms", lastFailureAt == 0L ? null : lastFailureAt);
+        return result;
+    }
+
+    private static final class PendingSend {
+        private final StatsMaterializationCommand command;
+        private final ListenableFuture<SendResult<String, String>> future;
+        private PendingSend(StatsMaterializationCommand command,
+                            ListenableFuture<SendResult<String, String>> future) {
+            this.command = command; this.future = future;
+        }
+    }
+}
@@ -0,0 +1,154 @@
+package com.common.service;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import java.sql.Date;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * 扫描三个小型聚合表与业务展示字段的差异，补偿“数据库已提交、进程却在发布通知前退出”的窗口。
+ * 不扫描 syslog 事实宽表，不参与事实消费事务。
+ */
+@Slf4j
+@Component
+public class StatsMaterializationReconciler {
+    private final JdbcTemplate jdbc;
+    private final StatsWriteModeProvider modeProvider;
+    private final StatsMaterializationPublisher publisher;
+    private final AtomicBoolean running = new AtomicBoolean();
+    private final AtomicLong runs = new AtomicLong();
+    private final AtomicLong detected = new AtomicLong();
+    private final AtomicLong failures = new AtomicLong();
+    private volatile LocalDateTime lastSuccessTime;
+    private volatile LocalDateTime lastFailureTime;
+
+    @Value("${stats.materialization.enabled:true}") private boolean enabled;
+    @Value("${stats.incremental.read-enabled:false}") private boolean readEnabled;
+    @Value("${stats.direct.device-table:device_daily_stats}") private String deviceTable;
+    @Value("${stats.direct.rule-table:normalize_rule_daily_stats}") private String ruleTable;
+    @Value("${stats.business-zone:Asia/Shanghai}") private String businessZone;
+
+    public StatsMaterializationReconciler(JdbcTemplate jdbc,
+                                          StatsWriteModeProvider modeProvider,
+                                          StatsMaterializationPublisher publisher) {
+        this.jdbc = jdbc;
+        this.modeProvider = modeProvider;
+        this.publisher = publisher;
+    }
+
+    @Scheduled(initialDelayString = "${stats.materialization.reconcile-initial-delay-ms:30000}",
+            fixedDelayString = "${stats.materialization.reconcile-interval-ms:60000}")
+    public void reconcile() {
+        if (!enabled || !readEnabled || !modeProvider.materializesBusinessFields()
+                || !running.compareAndSet(false, true)) {
+            return;
+        }
+        try {
+            List<StatsMaterializationCommand> commands = findDifferences();
+            publisher.enqueueNow(commands);
+            runs.incrementAndGet();
+            detected.addAndGet(commands.size());
+            lastSuccessTime = LocalDateTime.now();
+            if (!commands.isEmpty()) {
+                log.warn("统计物化对账发现差异并已重新投递: dirtyKeys={}", commands.size());
+            }
+        } catch (RuntimeException failure) {
+            failures.incrementAndGet();
+            lastFailureTime = LocalDateTime.now();
+            log.error("统计物化小表对账失败，下个周期自动重试", failure);
+        } finally {
+            running.set(false);
+        }
+    }
+
+    List<StatsMaterializationCommand> findDifferences() {
+        LocalDate today = LocalDate.now(ZoneId.of(businessZone));
+        List<StatsMaterializationCommand> commands = new ArrayList<>();
+        String safeDeviceTable = safeTable(deviceTable);
+        String safeRuleTable = safeTable(ruleTable);
+
+        List<Map<String, Object>> devices = jdbc.queryForList(
+                "SELECT s.device_id dimension_id FROM " + safeDeviceTable + " s " +
+                        "JOIN device_device d ON d.id=s.device_id WHERE s.stat_date=? AND (" +
+                        "NVL(d.today_parse_count,0)<>NVL(s.parse_count,0) OR " +
+                        "NVL(d.today_non_log_count,0)<>NVL(s.non_log_count,0) OR " +
+                        "d.today_stats_date IS NULL OR d.today_stats_date<>s.stat_date OR " +
+                        "(s.last_receive_time IS NOT NULL AND (d.last_receive_time IS NULL " +
+                        "OR d.last_receive_time<s.last_receive_time)))",
+                Date.valueOf(today));
+        for (Map<String, Object> row : devices) {
+            commands.add(StatsMaterializationCommand.device(today, longValue(row, "dimension_id")));
+        }
+
+        List<Map<String, Object>> rules = jdbc.queryForList(
+                "SELECT s.normalize_rule_id dimension_id FROM (SELECT normalize_rule_id," +
+                        "MAX(max_log_time) max_log_time FROM " + safeRuleTable +
+                        " WHERE max_log_time IS NOT NULL GROUP BY normalize_rule_id) s " +
+                        "JOIN dm_normalize_rule r ON r.id=s.normalize_rule_id AND r.del_flag='0' " +
+                        "WHERE r.first_data_saved_at IS NULL OR r.first_data_saved_at<s.max_log_time");
+        for (Map<String, Object> row : rules) {
+            commands.add(StatsMaterializationCommand.rule(today, longValue(row, "dimension_id")));
+        }
+
+        List<Map<String, Object>> collects = jdbc.queryForList(
+                "SELECT s.device_collect_id dimension_id FROM device_collect_stats s " +
+                        "JOIN device_collect_task c ON c.id=s.device_collect_id WHERE " +
+                        "(s.first_success_time IS NOT NULL AND (c.first_time IS NULL " +
+                        "OR c.first_time>s.first_success_time)) OR " +
+                        "(s.last_success_time IS NOT NULL AND (c.last_success_time IS NULL " +
+                        "OR c.last_success_time<s.last_success_time))");
+        for (Map<String, Object> row : collects) {
+            commands.add(StatsMaterializationCommand.collect(longValue(row, "dimension_id")));
+        }
+        return commands;
+    }
+
+    public Map<String, Object> status() {
+        Map<String, Object> result = new LinkedHashMap<>();
+        result.put("enabled", enabled);
+        result.put("runs_total", runs.get());
+        result.put("dirty_keys_detected_total", detected.get());
+        result.put("failures_total", failures.get());
+        result.put("running", running.get());
+        result.put("last_success_time", lastSuccessTime);
+        result.put("last_failure_time", lastFailureTime);
+        return result;
+    }
+
+    private static long longValue(Map<String, Object> row, String name) {
+        Object value = row.get(name);
+        if (value == null) value = row.get(name.toUpperCase(Locale.ROOT));
+        if (value == null) {
+            for (Map.Entry<String, Object> entry : row.entrySet()) {
+                if (entry.getKey().equalsIgnoreCase(name)) {
+                    value = entry.getValue();
+                    break;
+                }
+            }
+        }
+        if (!(value instanceof Number)) {
+            throw new IllegalStateException("统计物化对账结果缺少数字列 " + name + ": " + value);
+        }
+        return ((Number) value).longValue();
+    }
+
+    private static String safeTable(String table) {
+        if (table == null || !table.matches("[A-Za-z][A-Za-z0-9_]*")) {
+            throw new IllegalStateException("非法统计物化对账表名: " + table);
+        }
+        return table;
+    }
+}
@@ -50,6 +50,22 @@ public class KafkaReliabilityConfig {
        return factory;
    }

+    /** 统计脏键与原始日志隔离；同一维度 key 由 Kafka 分区保证只被一个组成员处理。 */
+    @Bean("statsMaterializationKafkaListenerContainerFactory")
+    public ConcurrentKafkaListenerContainerFactory<Object, Object> statsMaterializationKafkaListenerContainerFactory(
+            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
+            ConsumerFactory<Object, Object> consumerFactory,
+            DefaultErrorHandler syslogDefaultErrorHandler,
+            @Value("${stats.materialization.listener-concurrency:1}") int concurrency) {
+        if (concurrency < 1) {
+            throw new IllegalArgumentException("stats.materialization.listener-concurrency 必须大于 0");
+        }
+        ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
+                configuredFactory(configurer, consumerFactory, syslogDefaultErrorHandler);
+        factory.setConcurrency(concurrency);
+        return factory;
+    }
+
    private ConcurrentKafkaListenerContainerFactory<Object, Object> configuredFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory<Object, Object> consumerFactory,
@@ -4,6 +4,9 @@ import com.common.service.DirectStatsProjectionWriter;
 import com.common.service.DirectStatsDayRolloverService;
 import com.common.service.IngressStatsProjectionWriter;
 import com.common.service.DmStandardizedBatchExecutor;
+import com.common.service.StatsMaterializationConsumer;
+import com.common.service.StatsMaterializationPublisher;
+import com.common.service.StatsMaterializationReconciler;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -19,15 +22,24 @@ public class DirectStatsController {
    private final IngressStatsProjectionWriter ingressWriter;
    private final DirectStatsDayRolloverService dayRolloverService;
    private final DmStandardizedBatchExecutor batchExecutor;
+    private final StatsMaterializationPublisher materializationPublisher;
+    private final StatsMaterializationConsumer materializationConsumer;
+    private final StatsMaterializationReconciler materializationReconciler;

    public DirectStatsController(DirectStatsProjectionWriter writer,
                                 IngressStatsProjectionWriter ingressWriter,
                                 DirectStatsDayRolloverService dayRolloverService,
-                                 DmStandardizedBatchExecutor batchExecutor) {
+                                 DmStandardizedBatchExecutor batchExecutor,
+                                 StatsMaterializationPublisher materializationPublisher,
+                                 StatsMaterializationConsumer materializationConsumer,
+                                 StatsMaterializationReconciler materializationReconciler) {
        this.writer = writer;
        this.ingressWriter = ingressWriter;
        this.dayRolloverService = dayRolloverService;
        this.batchExecutor = batchExecutor;
+        this.materializationPublisher = materializationPublisher;
+        this.materializationConsumer = materializationConsumer;
+        this.materializationReconciler = materializationReconciler;
    }

    @GetMapping("/status")
@@ -36,6 +48,9 @@ public class DirectStatsController {
        status.put("ingress_metadata", ingressWriter.status());
        status.put("direct_day_rollover", dayRolloverService.status());
        status.put("dm_transaction", batchExecutor.status());
+        status.put("materialization_publisher", materializationPublisher.status());
+        status.put("materialization_consumer", materializationConsumer.status());
+        status.put("materialization_reconciler", materializationReconciler.status());
        return ResponseEntity.ok(status);
    }
 }
@@ -160,6 +160,8 @@ 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}
+dm.non-normal-write.max-concurrency-retries=${DM_NON_NORMAL_WRITE_MAX_CONCURRENCY_RETRIES:3}
+dm.non-normal-write.concurrency-retry-backoff-ms=${DM_NON_NORMAL_WRITE_CONCURRENCY_RETRY_BACKOFF_MS:50}

 # ETL配置
 etl.batch.page-size=1000
@@ -193,12 +195,26 @@ stats.direct.shadow-device-table=device_daily_stats_direct_shadow
 stats.direct.shadow-rule-table=normalize_rule_daily_stats_direct_shadow
 stats.direct.control-table=stats_direct_control
 stats.direct.today-stats-date-enabled=true
-stats.direct.max-concurrency-retries=1
-stats.direct.concurrency-retry-backoff-ms=20
+stats.direct.max-concurrency-retries=${STATS_DIRECT_MAX_CONCURRENCY_RETRIES:3}
+stats.direct.concurrency-retry-backoff-ms=${STATS_DIRECT_CONCURRENCY_RETRY_BACKOFF_MS:50}
 stats.direct.transaction-log-enabled=false
 stats.direct.day-rollover-check-cron=5 * * * * ?
 stats.business-zone=Asia/Shanghai
 stats.fact-only.confirm-rebuild-required=false
+# 事实事务只维护小型精确统计表；提交后按维度脏键异步刷新业务展示字段。
+# 三个实例必须配置相同 topic/group-id，Topic 分区数建议与原始 Topic 相同。
+stats.materialization.enabled=${STATS_MATERIALIZATION_ENABLED:true}
+stats.materialization.topic=${STATS_MATERIALIZATION_TOPIC:${spring.kafka.consumer.topic}.STATS.MATERIALIZE}
+stats.materialization.group-id=${STATS_MATERIALIZATION_GROUP_ID:${spring.kafka.consumer.group-id}-stats-materialize}
+# 每个实例一个物化 consumer；现场三个实例合计三个，足以处理低基数脏键并减少达梦并发。
+stats.materialization.listener-concurrency=${STATS_MATERIALIZATION_LISTENER_CONCURRENCY:1}
+stats.materialization.publish-interval-ms=${STATS_MATERIALIZATION_PUBLISH_INTERVAL_MS:200}
+stats.materialization.publish-timeout-ms=${STATS_MATERIALIZATION_PUBLISH_TIMEOUT_MS:10000}
+stats.materialization.publish-max-batch-size=${STATS_MATERIALIZATION_PUBLISH_MAX_BATCH_SIZE:1000}
+stats.materialization.max-concurrency-retries=${STATS_MATERIALIZATION_MAX_CONCURRENCY_RETRIES:3}
+stats.materialization.concurrency-retry-backoff-ms=${STATS_MATERIALIZATION_CONCURRENCY_RETRY_BACKOFF_MS:50}
+stats.materialization.reconcile-initial-delay-ms=${STATS_MATERIALIZATION_RECONCILE_INITIAL_DELAY_MS:30000}
+stats.materialization.reconcile-interval-ms=${STATS_MATERIALIZATION_RECONCILE_INTERVAL_MS:60000}

 # 标准化规则编译与批次缓存灰度开关
 normalization.jetcache.enabled=true
@@ -122,6 +122,23 @@ class LogNormalProcessorRuleSelectionTest {
        verify(fixture.nonNormalService, never()).saveMessage(any());
    }

+    @Test
+    void jsonWithoutObjectBoundaryAlsoGoesDirectlyToDeferredNonNormal() {
+        RuleSource invalidJson = new RuleSource(10L, "json-no-cropper", "json-no-cropper",
+                "json", "{\"mappers\":[" + equalMapping()
+                + "],\"complete_columns\":[]}", true, "0");
+        RuleSource regexWouldMatch = regexRule(20L, "regex-second", "value=(\\S+)");
+        Fixture fixture = fixture(Arrays.asList(invalidJson, regexWouldMatch), config(true));
+
+        LogNormalProcessor.NormalizationResult result = fixture.prepareActive("value=ok");
+
+        assertEquals(LogNormalProcessor.ProcessingOutcome.FILTERED, result.getOutcome());
+        assertTrue(result.getRows().isEmpty());
+        assertEquals(1, result.getNonNormalMessages().size());
+        assertEquals("JSON解析异常", result.getNonNormalMessages().get(0).getReason());
+        verify(fixture.nonNormalService, never()).saveMessage(any());
+    }
+
    @Test
    void legacyInvalidJsonIsNonNormalTerminalInsteadOfException() {
        Fixture fixture = fixture(Arrays.asList(
@@ -13,6 +13,7 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.dao.DataIntegrityViolationException;

 import java.time.Instant;
 import java.time.LocalDateTime;
@@ -63,6 +64,8 @@ class SysLogProcessorParallelTest {
        ReflectionTestUtils.setField(processor, "standardizedBatchWriter", batchWriter);
        ReflectionTestUtils.setField(processor, "nonNormalMessageService", nonNormalMessageService);
        ReflectionTestUtils.setField(processor, "nonNormalWriterExecutor", nonNormalExecutor);
+        ReflectionTestUtils.setField(processor, "nonNormalMaxConcurrencyRetries", 3);
+        ReflectionTestUtils.setField(processor, "nonNormalConcurrencyRetryBackoffMs", 0L);
        when(batchWriter.write(any())).thenReturn(new DmStandardizedBatchWriter.WriteReport<>());
        when(normalProcessor.captureRuleSnapshot(any())).thenReturn(
                new BatchRuleSnapshot(Collections.emptyMap(), System.currentTimeMillis()));
@@ -182,6 +185,31 @@ class SysLogProcessorParallelTest {
        assertEquals(2, metrics.filteredCount());
    }

+    @Test
+    void asynchronousNonNormalDeadlockRetriesInANewServiceTransaction() {
+        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));
+        });
+        AtomicInteger attempts = new AtomicInteger();
+        when(nonNormalMessageService.saveBatchMessages(anyList())).thenAnswer(invocation -> {
+            if (attempts.incrementAndGet() == 1) {
+                throw new RuntimeException(new DataIntegrityViolationException("Deadlock"));
+            }
+            return ((List<?>) invocation.getArgument(0)).size();
+        });
+
+        SysLogProcessor.BatchMetrics metrics = new SysLogProcessor.BatchMetrics();
+        processor.normalizePreparedMessages(messages(2), metrics);
+
+        verify(nonNormalMessageService, timeout(1000).times(2)).saveBatchMessages(anyList());
+        assertEquals(2, attempts.get());
+        assertEquals(2, metrics.nonNormalQueued());
+    }
+
    @Test
    void rejectedNonNormalTaskDoesNotFailKafkaBatch() {
        when(normalProcessor.prepare(any(PreparedNormalizationInput.class),
@@ -9,6 +9,7 @@ import java.util.List;
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import com.common.service.StatsMaterializationConsumer;

 /** 防止多节点 consumer-rule 的监听器 ID 或显式消费组在后续合并中回退。 */
 class SyslogKafkaTopologyTest {
@@ -48,4 +49,20 @@ class SyslogKafkaTopologyTest {
                "${syslog.reliability.replay-topic:${spring.kafka.consumer.topic}.REPLAY}"
        }, listener.topics());
    }
+
+    @Test
+    void statsMaterializerUsesDedicatedTopicAndGroup() throws Exception {
+        Method method = StatsMaterializationConsumer.class.getMethod(
+                "consume", List.class, org.springframework.kafka.support.Acknowledgment.class);
+        KafkaListener listener = method.getAnnotation(KafkaListener.class);
+
+        assertNotNull(listener);
+        assertEquals("syslog-stats-materializer", listener.id());
+        assertEquals("${stats.materialization.group-id:${spring.kafka.consumer.group-id}-stats-materialize}",
+                listener.groupId());
+        assertEquals("statsMaterializationKafkaListenerContainerFactory", listener.containerFactory());
+        assertArrayEquals(new String[]{
+                "${stats.materialization.topic:${spring.kafka.consumer.topic}.STATS.MATERIALIZE}"
+        }, listener.topics());
+    }
 }
@@ -4,6 +4,7 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Answers;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.test.util.ReflectionTestUtils;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
@@ -11,6 +12,7 @@ import org.springframework.transaction.support.TransactionSynchronizationManager
 import java.time.LocalDateTime;
 import java.time.LocalDate;
 import java.time.ZoneId;
+import java.sql.BatchUpdateException;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -19,7 +21,9 @@ import java.util.concurrent.atomic.AtomicInteger;

 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyCollection;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoInteractions;
@@ -105,10 +109,51 @@ class DirectStatsProjectionWriterTest {
        assertTrue(sql.getAllValues().get(1).contains("UPDATE device_device"));
    }

+    @Test
+    void damengBatchDeadlockWrappedAsIntegrityViolationIsRetryable() {
+        BatchUpdateException deadlock = new BatchUpdateException("Deadlock", null, 0, new int[0]);
+        DataIntegrityViolationException wrapped =
+                new DataIntegrityViolationException("PreparedStatementCallback", deadlock);
+
+        assertTrue(DirectStatsProjectionWriter.isConcurrentFailure(wrapped));
+    }
+
+    @Test
+    void kafkaMaterializationRemovesBusinessTableUpdatesFromFactTransaction() {
+        JdbcTemplate jdbc = mock(JdbcTemplate.class);
+        StatsMaterializationPublisher publisher = mock(StatsMaterializationPublisher.class);
+        when(publisher.isEnabled()).thenReturn(true);
+        DirectStatsProjectionWriter writer = new DirectStatsProjectionWriter(
+                jdbc, new StatsWriteModeProvider("DIRECT"), new DirectStatsWriteMetrics(), publisher);
+        configure(writer, true);
+        TransactionSynchronizationManager.setActualTransactionActive(true);
+
+        Map<String, Object> row = new HashMap<>();
+        row.put("created_at", LocalDateTime.now());
+        row.put("log_time", LocalDateTime.now());
+        row.put("device_id", 7L);
+        row.put("normalize_rule_id", 11L);
+        writer.writeStandardizedRow(row);
+
+        ArgumentCaptor<String> sql = ArgumentCaptor.forClass(String.class);
+        verify(jdbc, times(2)).batchUpdate(sql.capture(), anyList());
+        assertTrue(sql.getAllValues().get(0).contains("MERGE INTO device_daily_stats"));
+        assertTrue(sql.getAllValues().get(1).contains("MERGE INTO normalize_rule_daily_stats"));
+        verify(publisher).enqueueAfterCommit(anyCollection());
+        assertTrue("KAFKA_DIRTY_KEY".equals(writer.status().get("business_materialization_mode")));
+    }
+
    private DirectStatsProjectionWriter writer(JdbcTemplate jdbc, String mode, boolean readEnabled) {
        StatsWriteModeProvider modeProvider = new StatsWriteModeProvider(mode);
+        StatsMaterializationPublisher publisher = mock(StatsMaterializationPublisher.class);
+        when(publisher.isEnabled()).thenReturn(false);
        DirectStatsProjectionWriter writer = new DirectStatsProjectionWriter(
-                jdbc, modeProvider, new DirectStatsWriteMetrics());
+                jdbc, modeProvider, new DirectStatsWriteMetrics(), publisher);
+        configure(writer, readEnabled);
+        return writer;
+    }
+
+    private void configure(DirectStatsProjectionWriter writer, boolean readEnabled) {
        ReflectionTestUtils.setField(writer, "readEnabled", readEnabled);
        ReflectionTestUtils.setField(writer, "directDeviceTable", "device_daily_stats");
        ReflectionTestUtils.setField(writer, "directRuleTable", "normalize_rule_daily_stats");
@@ -117,6 +162,5 @@ class DirectStatsProjectionWriterTest {
        ReflectionTestUtils.setField(writer, "controlTable", "stats_direct_control");
        ReflectionTestUtils.setField(writer, "todayStatsDateEnabled", true);
        ReflectionTestUtils.setField(writer, "businessZone", "Asia/Shanghai");
-        return writer;
    }
 }
@@ -0,0 +1,28 @@
+package com.common.service;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDate;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class StatsMaterializationCommandTest {
+    @Test
+    void usesStableDimensionKeyAndValidatesDate() {
+        StatsMaterializationCommand command = StatsMaterializationCommand.device(
+                LocalDate.of(2026, 9, 11), 7L);
+
+        assertEquals("DEVICE:7", command.key());
+        assertEquals(LocalDate.of(2026, 9, 11), command.date());
+    }
+
+    @Test
+    void rejectsMissingDateForDailyDimension() {
+        StatsMaterializationCommand command = new StatsMaterializationCommand();
+        command.setType(StatsMaterializationCommand.Type.RULE);
+        command.setDimensionId(8L);
+
+        assertThrows(IllegalArgumentException.class, command::validate);
+    }
+}
@@ -0,0 +1,59 @@
+package com.common.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.support.SimpleTransactionStatus;
+
+import java.time.LocalDate;
+import java.time.ZoneId;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+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.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class StatsMaterializationConsumerTest {
+    @Test
+    void materializesExactValuesInOneTransaction() {
+        JdbcTemplate jdbc = mock(JdbcTemplate.class);
+        PlatformTransactionManager manager = mock(PlatformTransactionManager.class);
+        when(manager.getTransaction(any(TransactionDefinition.class)))
+                .thenReturn(new SimpleTransactionStatus());
+        StatsMaterializationConsumer consumer = new StatsMaterializationConsumer(
+                jdbc, manager, new ObjectMapper(), new StatsWriteModeProvider("DIRECT"));
+        ReflectionTestUtils.setField(consumer, "readEnabled", true);
+        ReflectionTestUtils.setField(consumer, "deviceTable", "device_daily_stats");
+        ReflectionTestUtils.setField(consumer, "ruleTable", "normalize_rule_daily_stats");
+        ReflectionTestUtils.setField(consumer, "todayStatsDateEnabled", true);
+        ReflectionTestUtils.setField(consumer, "businessZone", "Asia/Shanghai");
+        ReflectionTestUtils.setField(consumer, "transactionTimeoutSeconds", 45);
+        ReflectionTestUtils.setField(consumer, "maxConcurrencyRetries", 3);
+        ReflectionTestUtils.setField(consumer, "retryBackoffMs", 0L);
+        LocalDate today = LocalDate.now(ZoneId.of("Asia/Shanghai"));
+
+        int count = consumer.materializeWithRetry(Arrays.asList(
+                StatsMaterializationCommand.device(today, 7L),
+                StatsMaterializationCommand.rule(today, 8L),
+                StatsMaterializationCommand.collect(2L)));
+
+        assertEquals(3, count);
+        ArgumentCaptor<String> sql = ArgumentCaptor.forClass(String.class);
+        verify(jdbc, times(3)).batchUpdate(sql.capture(), anyList());
+        List<String> statements = sql.getAllValues();
+        assertTrue(statements.get(0).contains("MERGE INTO device_device"));
+        assertTrue(statements.get(1).contains("MAX(max_log_time)"));
+        assertTrue(statements.get(2).contains("MERGE INTO device_collect_task"));
+        verify(manager).commit(any());
+    }
+}
@@ -0,0 +1,41 @@
+package com.common.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+
+import java.time.LocalDate;
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+
+class StatsMaterializationPublisherTest {
+    @AfterEach
+    void clearTransactionState() {
+        TransactionSynchronizationManager.clear();
+    }
+
+    @Test
+    void dirtyKeyBecomesPublishableOnlyAfterDatabaseCommit() {
+        @SuppressWarnings("unchecked")
+        KafkaTemplate<String, String> kafka = mock(KafkaTemplate.class);
+        StatsMaterializationPublisher publisher = new StatsMaterializationPublisher(kafka, new ObjectMapper());
+        ReflectionTestUtils.setField(publisher, "enabled", true);
+        TransactionSynchronizationManager.initSynchronization();
+
+        publisher.enqueueAfterCommit(Collections.singletonList(
+                StatsMaterializationCommand.device(LocalDate.of(2026, 9, 11), 7L)));
+        assertEquals(0, publisher.status().get("pending_keys"));
+
+        for (TransactionSynchronization synchronization
+                : TransactionSynchronizationManager.getSynchronizations()) {
+            synchronization.afterCommit();
+        }
+        assertEquals(1, publisher.status().get("pending_keys"));
+    }
+}
@@ -0,0 +1,43 @@
+package com.common.service;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+
+class StatsMaterializationReconcilerTest {
+    @Test
+    void scansOnlyAggregateTablesAndBuildsExactDirtyKeys() {
+        JdbcTemplate jdbc = mock(JdbcTemplate.class, invocation -> {
+            if (!"queryForList".equals(invocation.getMethod().getName())) {
+                return org.mockito.Answers.RETURNS_DEFAULTS.answer(invocation);
+            }
+            String sql = (String) invocation.getArguments()[0];
+            Map<String, Object> row = new HashMap<>();
+            if (sql.contains("device_daily_stats")) row.put("DIMENSION_ID", 7L);
+            else if (sql.contains("normalize_rule_daily_stats")) row.put("dimension_id", 8L);
+            else if (sql.contains("device_collect_stats")) row.put("DIMENSION_ID", 2L);
+            else return Collections.emptyList();
+            return Collections.singletonList(row);
+        });
+        StatsMaterializationReconciler reconciler = new StatsMaterializationReconciler(
+                jdbc, new StatsWriteModeProvider("DIRECT"), mock(StatsMaterializationPublisher.class));
+        ReflectionTestUtils.setField(reconciler, "deviceTable", "device_daily_stats");
+        ReflectionTestUtils.setField(reconciler, "ruleTable", "normalize_rule_daily_stats");
+        ReflectionTestUtils.setField(reconciler, "businessZone", "Asia/Shanghai");
+
+        List<StatsMaterializationCommand> commands = reconciler.findDifferences();
+
+        assertEquals(3, commands.size());
+        assertEquals("DEVICE:7", commands.get(0).key());
+        assertEquals("RULE:8", commands.get(1).key());
+        assertEquals("COLLECT:2", commands.get(2).key());
+    }
+}
@@ -57,11 +57,11 @@ fi

 echo "新 Topic 使用分区数=${PARTITIONS}，副本数=${REPLICAS}"

-SUFFIXES=("RETRY.1M" "RETRY.10M" "RETRY.1H" "DLT" "REPLAY")
-RETENTIONS=(259200000 604800000 604800000 7776000000 604800000)
+SUFFIXES=("RETRY.1M" "RETRY.10M" "RETRY.1H" "DLT" "REPLAY" "STATS.MATERIALIZE")
+RETENTIONS=(259200000 604800000 604800000 7776000000 604800000 604800000)

 echo "[3/4] 创建或更新升级 Topic"
-for i in 0 1 2 3 4; do
+for i in 0 1 2 3 4 5; do
  TOPIC="${RAW_TOPIC}.${SUFFIXES[$i]}"
  RETENTION_MS="${RETENTIONS[$i]}"

@@ -39,10 +39,10 @@ if [[ ! "${PARTITIONS}" =~ ^[1-9][0-9]*$ ]] || [[ ! "${REPLICAS}" =~ ^[1-9][0-9]
  exit 1
 fi

-SUFFIXES=("RETRY.1M" "RETRY.10M" "RETRY.1H" "DLT" "REPLAY")
-RETENTIONS=(259200000 604800000 604800000 7776000000 604800000)
+SUFFIXES=("RETRY.1M" "RETRY.10M" "RETRY.1H" "DLT" "REPLAY" "STATS.MATERIALIZE")
+RETENTIONS=(259200000 604800000 604800000 7776000000 604800000 604800000)

-for i in 0 1 2 3 4; do
+for i in 0 1 2 3 4 5; do
  TOPIC="${RAW_TOPIC}.${SUFFIXES[$i]}"
  echo "创建或确认 Topic：${TOPIC}"
  "${TOPICS_CMD}" "${COMMON_ARGS[@]}" \
@@ -162,6 +162,17 @@ services:

 ## 发布顺序

+升级 `syslog-consumer-rule` 前，先在 Kafka 集群执行一次
+`upgrade-doc/v1/kafka/kafka-cluster-upgrade.sh`，确认已经创建：
+
+```text
+agent-01-syslog-topic.STATS.MATERIALIZE
+```
+
+实际前缀以现场 `spring.kafka.consumer.topic` 为准。统计 Topic 的分区数、副本数会沿用原始 Topic；
+三个 `syslog-consumer-rule` 实例必须使用相同的 `STATS_MATERIALIZATION_TOPIC` 和
+`STATS_MATERIALIZATION_GROUP_ID`。
+
 三个 JAR 一起升级：

 1. `syslog-consumer`
@@ -178,6 +189,8 @@ API 任务先执行 validate。先启用告警任务，确认数量、时间和

 - 三个服务启动正常；
 - Kafka lag 能回到 0，Retry/DLT 没有持续增长；
+- 统计物化 Topic 没有持续积压，日志持续出现 `统计物化批次完成`；
+- `/api/syslog/stats/direct/status` 中物化发布、消费、对账均无持续失败；
 - WAL checkpoint 能追上 durable sequence；
 - API validate 的配置、DNS、TLS、认证和最小查询全部通过；
 - API run 中 `recordsFetched` 与 `recordsDurable` 一致，`recordsQuarantined=0`；
@@ -17,3 +17,14 @@ NORMALIZATION_RULE_SELECTION_AMBIGUOUS_FALLBACK=LOWEST_RULE_ID
 NORMALIZATION_RULE_SELECTION_DETAIL_LOG_ENABLED=false
 NORMALIZATION_RULE_SELECTION_EXPLICIT_ENABLED=true
 NORMALIZATION_RULE_SELECTION_AUTOMATIC_FALLBACK_ENABLED=true
+
+# 统计投影解耦：三个 syslog-consumer-rule 实例保持完全相同的 Topic 与 Group ID。
+STATS_MATERIALIZATION_ENABLED=true
+STATS_MATERIALIZATION_TOPIC=agent-01-syslog-topic.STATS.MATERIALIZE
+STATS_MATERIALIZATION_GROUP_ID=agent-01-syslog-group-stats-materialize
+STATS_MATERIALIZATION_LISTENER_CONCURRENCY=1
+STATS_MATERIALIZATION_PUBLISH_INTERVAL_MS=200
+STATS_MATERIALIZATION_PUBLISH_MAX_BATCH_SIZE=1000
+STATS_MATERIALIZATION_MAX_CONCURRENCY_RETRIES=3
+STATS_MATERIALIZATION_CONCURRENCY_RETRY_BACKOFF_MS=50
+STATS_MATERIALIZATION_RECONCILE_INTERVAL_MS=60000