feat(syslog): 优化大日志通道并增加异常连续消费兜底

- 128 KiB 及以上日志默认进入大日志 Topic
- InfluxDB 原文及达梦大字段超过 256 KiB 时截断保存
- 截断内容追加 HB_XDR_TRUNCATED 标记
- InfluxDB 写入失败时继续执行规则解析和达梦写入
- 未知或系统性异常优先写入普通/大日志 DLT
- DLT 写入失败时记录特殊告警并提交 offset，避免阻塞后续消息
- DLT 失败等待时间默认限制为 1 秒
- 增加连续性放行、DLT 成功及失败监控指标
- 更新 Docker Compose、v3 升级文档及相关测试

haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java

@@ -67,6 +67,9 @@ public class SysLogProcessor {
     @Value("${syslog.large-listener.processing-batch-max-bytes:8388608}")
     private long largeProcessingBatchMaxBytes;
 
+    @Value("${syslog.failure-continuity.enabled:true}")
+    private boolean failureContinuityEnabled;
+
     @Autowired
     private AppProperties appProperties;
 
@@ -178,7 +181,7 @@ public class SysLogProcessor {
         }
         for (List<ConsumerRecord<String, String>> chunk : partitionRecordsByBytes(
                 records, Math.max(1L, largeProcessingBatchMaxBytes))) {
-            processBatch(chunk, enforceRetryDue, lane, largeTaskExecutor);
+            processBatchWithContinuity(chunk, enforceRetryDue, lane, largeTaskExecutor);
         }
         ack.acknowledge();
     }
@@ -201,8 +204,8 @@ public class SysLogProcessor {
 
     /**
      * 统一批处理边界：原文批量写入 InfluxDB 后，有界并行执行标准化。
-     * 正常消息仍需到达可靠终态才提交 offset；非标消息按现场临时策略异步落库，
-     * 其写库结果不参与 Kafka ACK。
+     * 失败时优先进入 Retry/DLT；最终兜底按连续性策略记录并推进 offset。非标消息按现场临时策略
+     * 异步落库，其写库结果不参与 Kafka ACK。
      */
     private void consumeBatch(List<ConsumerRecord<String, String>> records, Acknowledgment ack,
                               boolean enforceRetryDue, String lane, Executor normalizationExecutor) {
@@ -210,12 +213,29 @@ public class SysLogProcessor {
             ack.acknowledge();
             return;
         }
-        processBatch(records, enforceRetryDue, lane, normalizationExecutor);
+        processBatchWithContinuity(records, enforceRetryDue, lane, normalizationExecutor);
         ack.acknowledge();
     }
 
-    private void processBatch(List<ConsumerRecord<String, String>> records,
-                              boolean enforceRetryDue, String lane, Executor normalizationExecutor) {
+    private void processBatchWithContinuity(List<ConsumerRecord<String, String>> records,
+                                            boolean enforceRetryDue, String lane,
+                                            Executor normalizationExecutor) {
+        try {
+            processBatch(records, enforceRetryDue, lane, normalizationExecutor);
+        } catch (SyslogFailureRouter.RetryNotDueException notDue) {
+            // Retry Topic 的计划时间尚未到达，不属于处理异常，仍按原延时语义保留 offset。
+            throw notDue;
+        } catch (RuntimeException failure) {
+            if (!failureContinuityEnabled) throw failure;
+            routeContinuityBatch(records, failure, "BATCH_" + lane);
+            log.error("[SYSLOG_CONTINUITY] 批次特殊异常已记录，继续消费后续消息: lane={}, "
+                            + "records={}, errorType={}",
+                    lane, records.size(), failure.getClass().getName(), failure);
+        }
+    }
+
+    void processBatch(List<ConsumerRecord<String, String>> records,
+                      boolean enforceRetryDue, String lane, Executor normalizationExecutor) {
         long batchStartedAt = System.currentTimeMillis();
         long prepareStartedAt = batchStartedAt;
         BatchMetrics metrics = new BatchMetrics();
@@ -239,7 +259,21 @@ public class SysLogProcessor {
         int ingressGroups = ingressStatsProjectionWriter.writeBatch(ingressEvents(preparedMessages));
 
         long influxStartedAt = System.currentTimeMillis();
-        List<PreparedMessage> persistedMessages = persistRawBatch(preparedMessages, metrics);
+        List<PreparedMessage> persistedMessages;
+        try {
+            persistedMessages = persistRawBatch(preparedMessages, metrics);
+        } catch (RuntimeException rawStorageFailure) {
+            if (!failureContinuityEnabled) throw rawStorageFailure;
+            routeContinuityPrepared(preparedMessages, rawStorageFailure, "RAW_STORAGE");
+            rawInfluxMetrics.rawWriteBypassed(preparedMessages.size());
+            metrics.rawStorageBypassed.addAndGet(preparedMessages.size());
+            // 无论 DLT 是否可用，都继续使用内存中的完整内容执行规则和达梦写入。
+            persistedMessages = preparedMessages;
+            log.error("[SYSLOG_CONTINUITY] InfluxDB 原文写入失败，已尝试记录 DLT 并继续标准化: "
+                            + "lane={}, records={}, errorType={}",
+                    lane, preparedMessages.size(), rawStorageFailure.getClass().getName(),
+                    rawStorageFailure);
+        }
         long influxMs = System.currentTimeMillis() - influxStartedAt;
 
         long normalizeStartedAt = System.currentTimeMillis();
@@ -249,7 +283,7 @@ public class SysLogProcessor {
         totalProcessed.addAndGet(records.size());
         log.info("Syslog批次完成: lane={}, topicPartitions={}, records={}, prepared={}, persisted={}, "
                         + "success={}, filtered={}, retry={}, dlt={}, normalizedRows={}, dmSuccessfulRows={}, "
-                        + "dmFailedMessages={}, ingressGroups={}, prepareMs={}, influxMs={}, normalizeMs={}, "
+                        + "dmFailedMessages={}, ingressGroups={}, rawStorageBypassed={}, prepareMs={}, influxMs={}, normalizeMs={}, "
                         + "ruleComputeMs={}, ruleSnapshotMs={}, messageParseMs={}, fieldMappingMs={}, "
                         + "filterCompleteMs={}, enrichmentMs={}, ruleAssembleMs={}, compiledRuleHit={}, "
                         + "compiledRuleMiss={}, messageParseHit={}, messageParseMiss={}, deviceUnique={}, "
@@ -259,6 +293,7 @@ public class SysLogProcessor {
                 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,
+                metrics.rawStorageBypassed.get(),
                 prepareMs, influxMs, normalizeMs, metrics.ruleComputeMs(), metrics.ruleSnapshotMs(),
                 metrics.messageParseMs(), metrics.fieldMappingMs(), metrics.filterCompleteMs(),
                 metrics.enrichmentMs(), metrics.ruleAssembleMs(), metrics.compiledRuleHits(),
@@ -271,6 +306,73 @@ public class SysLogProcessor {
                 System.currentTimeMillis() - batchStartedAt);
     }
 
+    private void routeContinuityBatch(List<ConsumerRecord<String, String>> records,
+                                      RuntimeException failure,
+                                      String stage) {
+        int published = 0;
+        for (ConsumerRecord<String, String> record : records) {
+            RuntimeException dltFailure = tryRouteContinuityRecord(record, null, failure, stage);
+            if (dltFailure != null) {
+                recordContinuityLoss(records.size(), published, record, failure, dltFailure, stage);
+                return;
+            }
+            published++;
+        }
+        rawInfluxMetrics.continuityDltPublished(published);
+    }
+
+    private void routeContinuityPrepared(List<PreparedMessage> messages,
+                                         RuntimeException failure,
+                                         String stage) {
+        int published = 0;
+        for (PreparedMessage message : messages) {
+            RuntimeException dltFailure = tryRouteContinuityRecord(
+                    message.record, message.context, failure, stage);
+            if (dltFailure != null) {
+                recordContinuityLoss(messages.size(), published, message.record,
+                        failure, dltFailure, stage);
+                return;
+            }
+            published++;
+        }
+        rawInfluxMetrics.continuityDltPublished(published);
+    }
+
+    /**
+     * 连续性优先的最后边界。优先把原消息写入 DLT；即使 DLT 不可用，也只记录特殊告警和计数，
+     * 不再把异常抛回 Kafka listener，避免一个坏批次永久阻塞整个分区。
+     */
+    private RuntimeException tryRouteContinuityRecord(ConsumerRecord<String, String> record,
+                                                      SyslogFailureRouter.MessageContext context,
+                                                      RuntimeException sourceFailure,
+                                                      String stage) {
+        try {
+            SyslogFailureRouter.MessageContext resolved = context == null
+                    ? failureRouter.resolve(record) : context;
+            failureRouter.routeContinuityDlt(record, resolved, sourceFailure, stage);
+            return null;
+        } catch (RuntimeException dltFailure) {
+            return dltFailure;
+        }
+    }
+
+    private void recordContinuityLoss(int total, int published,
+                                      ConsumerRecord<String, String> firstFailedRecord,
+                                      RuntimeException sourceFailure,
+                                      RuntimeException dltFailure,
+                                      String stage) {
+        if (published > 0) rawInfluxMetrics.continuityDltPublished(published);
+        int releasedWithoutDlt = total - published;
+        rawInfluxMetrics.continuityDltPublishFailed(releasedWithoutDlt);
+        // 首次失败后不再逐条等待或打印，剩余消息用批次计数记录，避免 Kafka 故障时放大阻塞和日志量。
+        log.error("[SYSLOG_CONTINUITY_LOSS] 特殊异常写入 DLT 失败，但按连续性策略仍将提交原 offset: "
+                        + "topic={}, partition={}, firstFailedOffset={}, stage={}, batchRecords={}, "
+                        + "dltPublished={}, releasedWithoutDlt={}, sourceErrorType={}, dltErrorType={}",
+                firstFailedRecord.topic(), firstFailedRecord.partition(), firstFailedRecord.offset(), stage,
+                total, published, releasedWithoutDlt, sourceFailure.getClass().getName(),
+                dltFailure.getClass().getName(), dltFailure);
+    }
+
     static List<List<ConsumerRecord<String, String>>> partitionRecordsByBytes(
             List<ConsumerRecord<String, String>> records, long maxBytes) {
         if (maxBytes <= 0L) throw new IllegalArgumentException("maxBytes 必须大于 0");
@@ -495,7 +597,7 @@ public class SysLogProcessor {
     /**
      * 一次 HTTP 请求持久化当前 Kafka 批次的原文。
      * 批量请求如果因单条数据格式被 InfluxDB 拒绝，降级为逐条写入以定位坏消息；
-     * 连接、超时或服务端故障则直接保留原 offset，不制造 DLT 风暴。
+     * 连接、超时或服务端故障会上抛到连续性边界，由其尝试 DLT 后继续标准化。
      */
     private List<PreparedMessage> persistRawBatch(List<PreparedMessage> messages, BatchMetrics metrics) {
         if (messages.isEmpty()) {
@@ -536,7 +638,7 @@ public class SysLogProcessor {
             return persisted;
         } catch (Exception dependencyFailure) {
             if (messages.get(0).v2) rawInfluxMetrics.v2FailedBatch();
-            // 整批原文的持久化结果不可确认，必须让 Kafka 从原 offset 重投。
+            // 整批原文的持久化结果不可确认，交给外层连续性策略统一处理。
             throw new IllegalStateException("InfluxDB 批量原文写入未取得成功确认，禁止提交 offset", dependencyFailure);
         }
     }
@@ -595,7 +697,7 @@ public class SysLogProcessor {
     }
 
     /**
-     * 批内有界并行标准化。任务异常必须汇总回 listener；超时或系统性失败时整批不 ACK。
+     * 批内有界并行标准化。任务异常汇总回 listener 外层的连续性边界。
      */
     void normalizePreparedMessages(List<PreparedMessage> messages, BatchMetrics metrics) {
         normalizePreparedMessages(messages, metrics, taskExecutor);
@@ -783,7 +885,7 @@ public class SysLogProcessor {
         }
     }
 
-    /** Retry/DLT 取得 Broker ACK 才返回；系统性故障抛出以保留原 offset。 */
+    /** 常规 Retry/DLT 路由；未到终态的异常交给外层连续性边界记录并放行。 */
     private SyslogFailureRouter.TerminalState routeFailure(ConsumerRecord<String, String> record,
                                                            SyslogFailureRouter.MessageContext context,
                                                            Exception failure) {
@@ -905,6 +1007,7 @@ public class SysLogProcessor {
         private final AtomicInteger normalizedRows = new AtomicInteger();
         private final AtomicInteger dmSuccessfulRows = new AtomicInteger();
         private final AtomicInteger dmFailedMessages = new AtomicInteger();
+        private final AtomicInteger rawStorageBypassed = new AtomicInteger();
         private final AtomicLong ruleComputeMs = new AtomicLong();
         private final AtomicLong ruleSnapshotMs = new AtomicLong();
         private final AtomicLong dmWriteMs = new AtomicLong();

haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SyslogFailureRouter.java

@@ -30,9 +30,8 @@ import java.util.concurrent.TimeoutException;
 /**
  * Kafka 异常消息可靠路由器。
  *
- * <p>该组件把“业务处理失败”与“原 offset 是否允许推进”放在同一个同步边界内：
- * Retry/DLT 的 send Future 只有取得 Broker ACK 后才返回终态；发布失败或结果未知一律抛出，
- * 让原消息保持未提交。这样不会用一次异步方法调用冒充消息已经被补偿链路接管。</p>
+ * <p>常规 Retry/DLT 的 send Future 只有取得 Broker ACK 后才返回终态；发布失败或结果未知会抛给
+ * 调用方。活动 listener 外层另有连续性优先边界，可按现场策略记录后放行。</p>
  */
 @Slf4j
 @Component
@@ -65,6 +64,8 @@ public class SyslogFailureRouter {
     private String largeDltTopic;
     @Value("${syslog.reliability.publish-timeout-ms:30000}")
     private long publishTimeoutMs;
+    @Value("${syslog.failure-continuity.dlt-publish-timeout-ms:1000}")
+    private long continuityDltPublishTimeoutMs;
     @Value("${syslog.reliability.consumer-version:unknown}")
     private String consumerVersion;
 
@@ -132,7 +133,7 @@ public class SyslogFailureRouter {
     public TerminalState route(ConsumerRecord<String, String> record, MessageContext context, Throwable failure) {
         FailureCategory category = classify(failure);
         if (category == FailureCategory.SYSTEMIC) {
-            // [SYSLOG-ALARM][KACK][已修复] 系统性依赖故障不倾倒 DLT，也不提交原 offset，由容器背压并等待恢复。
+            // 常规路由把系统性故障上抛；活动 listener 的连续性边界再决定是否转 DLT 并放行。
             return TerminalState.PAUSED;
         }
 
@@ -179,11 +180,59 @@ public class SyslogFailureRouter {
         return state;
     }
 
+    /**
+     * 连续性优先兜底：把无法安全完成当前处理阶段的原消息直接写入 DLT。
+     * 本方法仍显式报告发布失败；是否继续提交原 offset 由调用方的连续性策略决定。
+     */
+    public TerminalState routeContinuityDlt(ConsumerRecord<String, String> record,
+                                             MessageContext context,
+                                             Throwable failure,
+                                             String failedStage) {
+        int nextAttempt = Math.max(0, context.getAttemptCount()) + 1;
+        boolean largeLane = largeOriginalTopic != null
+                && largeOriginalTopic.equals(context.getOriginalTopic());
+        String targetTopic = largeLane ? largeDltTopic : dltTopic;
+        FailureEnvelope source = readEnvelope(record.value());
+        FailureEnvelope envelope = new FailureEnvelope();
+        envelope.setEnvelopeType(ENVELOPE_TYPE);
+        envelope.setEventId(context.getEventId());
+        envelope.setDeviceId(context.getDeviceId() != null ? context.getDeviceId()
+                : (source == null ? null : source.getDeviceId()));
+        envelope.setOriginalTopic(source == null ? record.topic() : source.getOriginalTopic());
+        envelope.setOriginalPartition(source == null ? record.partition() : source.getOriginalPartition());
+        envelope.setOriginalOffset(source == null ? record.offset() : source.getOriginalOffset());
+        envelope.setCurrentTopic(record.topic());
+        envelope.setFirstReceivedAt(context.getFirstReceivedAt());
+        envelope.setFailedStage(failedStage == null ? "UNKNOWN" : failedStage);
+        envelope.setErrorCode("SYSLOG_CONTINUITY_BYPASS");
+        Throwable root = rootCause(failure);
+        envelope.setErrorType(root.getClass().getName());
+        envelope.setErrorMessage(sanitize(root.getMessage()));
+        envelope.setAttemptCount(nextAttempt);
+        envelope.setNextRetryAt(null);
+        envelope.setRuleId(source == null ? null : source.getRuleId());
+        envelope.setRuleVersion(source == null ? null : source.getRuleVersion());
+        envelope.setConsumerVersion(consumerVersion);
+        envelope.setOriginalPayload(context.getPayload());
+        envelope.setPointTimeNs(context.getPointTimeNs());
+
+        publishAndAwait(targetTopic, context.getEventId(), envelope,
+                Math.max(1L, continuityDltPublishTimeoutMs));
+        log.debug("[SYSLOG_CONTINUITY] 特殊异常已可靠写入 DLT: eventId={}, "
+                        + "stage={}, target={}, errorType={}",
+                context.getEventId(), envelope.getFailedStage(), targetTopic, envelope.getErrorType());
+        return TerminalState.DLT_PUBLISHED;
+    }
+
     private void publishAndAwait(String topic, String key, FailureEnvelope envelope) {
+        publishAndAwait(topic, key, envelope, publishTimeoutMs);
+    }
+
+    private void publishAndAwait(String topic, String key, FailureEnvelope envelope, long timeoutMs) {
         try {
             String body = objectMapper.writeValueAsString(envelope);
             // [流程节点4.3] 必须等待 Broker ACK；仅调用 send() 不构成 RETRY_PUBLISHED/DLT_PUBLISHED。
-            kafkaTemplate.send(topic, key, body).get(publishTimeoutMs, TimeUnit.MILLISECONDS);
+            kafkaTemplate.send(topic, key, body).get(timeoutMs, TimeUnit.MILLISECONDS);
         } catch (InterruptedException interrupted) {
             Thread.currentThread().interrupt();
             throw new FailurePublishException("发布异常消息时线程被中断，原 offset 不得提交", interrupted);

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmLargeTextLimiter.java

@@ -18,13 +18,13 @@ import java.util.concurrent.atomic.LongAdder;
 public class DmLargeTextLimiter {
     private static final long MIN_ENABLED_LIMIT = 64L;
 
-    @Value("${dm.normal.large-field-max-utf8-bytes:0}")
+    @Value("${dm.normal.large-field-max-utf8-bytes:262144}")
     private long normalMaxBytes;
 
     @Value("${dm.normal.large-fields:http_req_body,http_resp_body,http_req_header,http_resp_header,detail,description,cmdline,shell_cmdline,parent_cmdline,mail_message}")
     private String normalFieldsValue;
 
-    @Value("${dm.non-normal.large-field-max-utf8-bytes:0}")
+    @Value("${dm.non-normal.large-field-max-utf8-bytes:262144}")
     private long nonNormalMaxBytes;
 
     private volatile Set<String> normalFields = Collections.emptySet();

haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/raw/RawInfluxMetrics.java

@@ -19,6 +19,10 @@ public class RawInfluxMetrics {
     private final LongAdder messagesTruncated = new LongAdder();
     private final LongAdder originalBytesTruncated = new LongAdder();
     private final LongAdder storedBytesAfterTruncation = new LongAdder();
+    private final LongAdder rawWriteBypassedBatches = new LongAdder();
+    private final LongAdder rawWriteBypassedMessages = new LongAdder();
+    private final LongAdder continuityDltPublished = new LongAdder();
+    private final LongAdder continuityDltPublishFailed = new LongAdder();
 
     public void legacyMissing() { legacyMissing.increment(); }
     public void legacyAssigned() { legacyAssigned.increment(); }
@@ -32,6 +36,12 @@ public class RawInfluxMetrics {
         originalBytesTruncated.add(originalBytes);
         storedBytesAfterTruncation.add(storedBytes);
     }
+    public void rawWriteBypassed(long messages) {
+        rawWriteBypassedBatches.increment();
+        rawWriteBypassedMessages.add(messages);
+    }
+    public void continuityDltPublished(long messages) { continuityDltPublished.add(messages); }
+    public void continuityDltPublishFailed(long messages) { continuityDltPublishFailed.add(messages); }
 
     public Map<String, Object> snapshot() {
         Map<String, Object> values = new LinkedHashMap<>();
@@ -47,6 +57,10 @@ public class RawInfluxMetrics {
         values.put("raw_message_truncated_total", messagesTruncated.sum());
         values.put("raw_message_truncated_original_bytes_total", originalBytesTruncated.sum());
         values.put("raw_message_truncated_stored_bytes_total", storedBytesAfterTruncation.sum());
+        values.put("raw_write_bypassed_batches_total", rawWriteBypassedBatches.sum());
+        values.put("raw_write_bypassed_messages_total", rawWriteBypassedMessages.sum());
+        values.put("continuity_dlt_published_total", continuityDltPublished.sum());
+        values.put("continuity_dlt_publish_failed_total", continuityDltPublishFailed.sum());
         return values;
     }
 }

haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/raw/RawStorageProperties.java

@@ -20,7 +20,7 @@ public class RawStorageProperties {
     @Value("${influxdb.raw.query-max-rows:10000}") private int queryMaxRows;
     @Value("${influxdb.raw.query-max-response-bytes:10485760}") private long queryMaxResponseBytes;
     /** 原始 message 单字段的 UTF-8 存储上限；0 表示不限制。 */
-    @Value("${influxdb.raw.message-max-utf8-bytes:0}") private long messageMaxUtf8Bytes;
+    @Value("${influxdb.raw.message-max-utf8-bytes:262144}") private long messageMaxUtf8Bytes;
     /** 一次同步写请求的估算 Line Protocol 上限；按字节切批，不能只按条数。 */
     @Value("${influxdb.raw.write-batch-max-bytes:8388608}") private long writeBatchMaxBytes;
 

haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties

@@ -35,7 +35,7 @@ influxdb.raw.legacy-point-time-fallback-enabled=${INFLUXDB_RAW_LEGACY_POINT_TIME
 influxdb.raw.query-max-rows=${INFLUXDB_RAW_QUERY_MAX_ROWS:10000}
 influxdb.raw.query-max-response-bytes=${INFLUXDB_RAW_QUERY_MAX_RESPONSE_BYTES:10485760}
 # 0 表示不限制；启用后最终 message（含截断标记）不超过该 UTF-8 字节数。
-influxdb.raw.message-max-utf8-bytes=${INFLUXDB_RAW_MESSAGE_MAX_UTF8_BYTES:1048576}
+influxdb.raw.message-max-utf8-bytes=${INFLUXDB_RAW_MESSAGE_MAX_UTF8_BYTES:262144}
 # Influx 同步写按 Line Protocol 估算字节切批，避免少量大日志形成超大 HTTP 请求。
 influxdb.raw.write-batch-max-bytes=${INFLUXDB_RAW_WRITE_BATCH_MAX_BYTES:8388608}
 # InfluxDB 2.7 连接超时配置
@@ -79,6 +79,10 @@ spring.kafka.consumer.large-topic=${KAFKA_CONSUMER_LARGE_TOPIC:${spring.kafka.co
 syslog.raw-consumer.enabled=true
 # Raw Topic 当前有 3 个分区，使用独立工厂显式创建 3 个 consumer；可通过环境变量灰度调整。
 syslog.raw-listener.concurrency=${SYSLOG_RAW_LISTENER_CONCURRENCY:3}
+# 连续性优先：特殊/系统性异常优先写 DLT；即使 DLT 失败也记录告警并推进原 offset，避免阻塞分区。
+syslog.failure-continuity.enabled=${SYSLOG_FAILURE_CONTINUITY_ENABLED:true}
+# DLT 故障时最多快速等待 1 秒；首次失败后整批直接记录并放行。
+syslog.failure-continuity.dlt-publish-timeout-ms=${SYSLOG_FAILURE_CONTINUITY_DLT_PUBLISH_TIMEOUT_MS:1000}
 # 大日志使用独立 Topic、消费组和 ConsumerFactory；建议生产环境独立 JVM 部署。
 syslog.large-listener.enabled=${SYSLOG_LARGE_LISTENER_ENABLED:true}
 syslog.large-listener.group-id=${SYSLOG_LARGE_LISTENER_GROUP_ID:${spring.kafka.consumer.group-id}-large}
@@ -180,7 +184,7 @@ dm.standardized-write.batch-size=${DM_STANDARDIZED_WRITE_BATCH_SIZE:500}
 dm.standardized-write.writer-concurrency=2
 dm.standardized-write.writer-queue-capacity=100
 # 标准化大字段和非标原文均在达梦持久化边界截断；0 表示不限制。
-dm.normal.large-field-max-utf8-bytes=${DM_NORMAL_LARGE_FIELD_MAX_UTF8_BYTES:65536}
+dm.normal.large-field-max-utf8-bytes=${DM_NORMAL_LARGE_FIELD_MAX_UTF8_BYTES:262144}
 dm.normal.large-fields=${DM_NORMAL_LARGE_FIELDS:http_req_body,http_resp_body,http_req_header,http_resp_header,detail,description,cmdline,shell_cmdline,parent_cmdline,mail_message}
 dm.non-normal.large-field-max-utf8-bytes=${DM_NON_NORMAL_LARGE_FIELD_MAX_UTF8_BYTES:262144}
 dm.standardized-write.transaction-timeout-seconds=45

haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogProcessorParallelTest.java

@@ -8,12 +8,14 @@ import com.common.service.DeviceCollectTaskService;
 import com.common.service.DeviceDeviceService;
 import com.common.service.DmStandardizedBatchWriter;
 import com.common.service.SyslogNonNormalMessageService;
+import com.influx.raw.RawInfluxMetrics;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 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 org.springframework.kafka.support.Acknowledgment;
 
 import java.time.Instant;
 import java.time.LocalDateTime;
@@ -21,6 +23,7 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -31,13 +34,17 @@ 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.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.when;
 
-/** 验证恢复并行后仍保留“全部可靠终态才允许返回”的处理边界。 */
+/** 验证批内并行、常规可靠路由及连续性优先的最终放行边界。 */
 class SysLogProcessorParallelTest {
 
     private ExecutorService executor;
@@ -47,6 +54,7 @@ class SysLogProcessorParallelTest {
     private SyslogFailureRouter failureRouter;
     private DmStandardizedBatchWriter batchWriter;
     private SyslogNonNormalMessageService nonNormalMessageService;
+    private RawInfluxMetrics rawInfluxMetrics;
 
     @BeforeEach
     void setUp() {
@@ -57,6 +65,7 @@ class SysLogProcessorParallelTest {
         failureRouter = mock(SyslogFailureRouter.class);
         batchWriter = mock(DmStandardizedBatchWriter.class);
         nonNormalMessageService = mock(SyslogNonNormalMessageService.class);
+        rawInfluxMetrics = mock(RawInfluxMetrics.class);
         ReflectionTestUtils.setField(processor, "taskExecutor", executor);
         ReflectionTestUtils.setField(processor, "processTimeoutMs", 5000L);
         ReflectionTestUtils.setField(processor, "logNormalProcessor", normalProcessor);
@@ -66,6 +75,8 @@ class SysLogProcessorParallelTest {
         ReflectionTestUtils.setField(processor, "nonNormalWriterExecutor", nonNormalExecutor);
         ReflectionTestUtils.setField(processor, "nonNormalMaxConcurrencyRetries", 3);
         ReflectionTestUtils.setField(processor, "nonNormalConcurrencyRetryBackoffMs", 0L);
+        ReflectionTestUtils.setField(processor, "failureContinuityEnabled", true);
+        ReflectionTestUtils.setField(processor, "rawInfluxMetrics", rawInfluxMetrics);
         when(batchWriter.write(any())).thenReturn(new DmStandardizedBatchWriter.WriteReport<>());
         when(normalProcessor.captureRuleSnapshot(any())).thenReturn(
                 new BatchRuleSnapshot(Collections.emptyMap(), System.currentTimeMillis()));
@@ -232,6 +243,61 @@ class SysLogProcessorParallelTest {
         assertEquals(3, metrics.filteredCount());
     }
 
+    @Test
+    void unexpectedBatchFailureIsRecordedAndOffsetIsAcknowledged() {
+        SysLogProcessor spyProcessor = spy(processor);
+        RuntimeException sourceFailure = new IllegalStateException("unknown processing failure");
+        doThrow(sourceFailure).when(spyProcessor)
+                .processBatch(anyList(), eq(false), eq("RAW"), any(Executor.class));
+        when(failureRouter.resolve(any())).thenAnswer(invocation -> {
+            ConsumerRecord<String, String> record = invocation.getArgument(0);
+            return new SyslogFailureRouter.MessageContext(record.value(), "event-" + record.offset(),
+                    0, null, Instant.now(), record.topic(), null);
+        });
+        when(failureRouter.routeContinuityDlt(any(), any(), eq(sourceFailure), eq("BATCH_RAW")))
+                .thenReturn(SyslogFailureRouter.TerminalState.DLT_PUBLISHED);
+        Acknowledgment ack = mock(Acknowledgment.class);
+
+        spyProcessor.consumeAndParallelProcess(records(2), ack);
+
+        verify(failureRouter, times(2))
+                .routeContinuityDlt(any(), any(), eq(sourceFailure), eq("BATCH_RAW"));
+        verify(rawInfluxMetrics).continuityDltPublished(2L);
+        verify(ack).acknowledge();
+    }
+
+    @Test
+    void dltPublishFailureIsLoggedAndOffsetIsStillAcknowledged() {
+        SysLogProcessor spyProcessor = spy(processor);
+        RuntimeException sourceFailure = new IllegalStateException("unknown processing failure");
+        doThrow(sourceFailure).when(spyProcessor)
+                .processBatch(anyList(), eq(false), eq("RAW"), any(Executor.class));
+        when(failureRouter.resolve(any())).thenAnswer(invocation -> {
+            ConsumerRecord<String, String> record = invocation.getArgument(0);
+            return new SyslogFailureRouter.MessageContext(record.value(), "event-" + record.offset(),
+                    0, null, Instant.now(), record.topic(), null);
+        });
+        when(failureRouter.routeContinuityDlt(any(), any(), eq(sourceFailure), eq("BATCH_RAW")))
+                .thenThrow(new SyslogFailureRouter.FailurePublishException(
+                        "broker unavailable", new IllegalStateException("send failed")));
+        Acknowledgment ack = mock(Acknowledgment.class);
+
+        spyProcessor.consumeAndParallelProcess(records(2), ack);
+
+        verify(failureRouter)
+                .routeContinuityDlt(any(), any(), eq(sourceFailure), eq("BATCH_RAW"));
+        verify(rawInfluxMetrics).continuityDltPublishFailed(2L);
+        verify(ack).acknowledge();
+    }
+
+    private List<ConsumerRecord<String, String>> records(int count) {
+        List<ConsumerRecord<String, String>> records = new ArrayList<>();
+        for (int i = 0; i < count; i++) {
+            records.add(new ConsumerRecord<>("raw", 0, i, "key-" + i, "ciphertext-" + i));
+        }
+        return records;
+    }
+
     private List<SysLogProcessor.PreparedMessage> messages(int count) {
         List<SysLogProcessor.PreparedMessage> messages = new ArrayList<>();
         for (int i = 0; i < count; i++) {

haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SyslogFailureRouterReliabilityTest.java

@@ -117,6 +117,41 @@ class SyslogFailureRouterReliabilityTest {
                 () -> router.route(record, router.resolve(record), new SocketException("reset")));
     }
 
+    @Test
+    void continuityFailureIsWrittenDirectlyToDltWithOriginalPayload() throws Exception {
+        SettableListenableFuture<SendResult<String, String>> acknowledged = new SettableListenableFuture<>();
+        acknowledged.set(null);
+        when(template.send(eq("raw.DLT"), anyString(), anyString())).thenReturn(acknowledged);
+        ArgumentCaptor<String> body = ArgumentCaptor.forClass(String.class);
+
+        SyslogFailureRouter.TerminalState state = router.routeContinuityDlt(
+                record, router.resolve(record), new InfluxException("influx unavailable"), "RAW_STORAGE");
+
+        assertEquals(SyslogFailureRouter.TerminalState.DLT_PUBLISHED, state);
+        verify(template).send(eq("raw.DLT"), anyString(), body.capture());
+        SyslogFailureRouter.FailureEnvelope envelope = new ObjectMapper().findAndRegisterModules()
+                .readValue(body.getValue(), SyslogFailureRouter.FailureEnvelope.class);
+        assertEquals("SYSLOG_CONTINUITY_BYPASS", envelope.getErrorCode());
+        assertEquals("RAW_STORAGE", envelope.getFailedStage());
+        assertEquals("ciphertext", envelope.getOriginalPayload());
+        assertEquals("raw", envelope.getOriginalTopic());
+        assertEquals(42L, envelope.getOriginalOffset());
+    }
+
+    @Test
+    void continuityFailureFromLargeLaneUsesLargeDlt() {
+        ConsumerRecord<String, String> largeRecord = new ConsumerRecord<>(
+                "raw.large", 1, 7L, "event", "large-ciphertext");
+        SettableListenableFuture<SendResult<String, String>> acknowledged = new SettableListenableFuture<>();
+        acknowledged.set(null);
+        when(template.send(eq("raw.large.DLT"), anyString(), anyString())).thenReturn(acknowledged);
+
+        router.routeContinuityDlt(largeRecord, router.resolve(largeRecord),
+                new IllegalStateException("unknown"), "BATCH_RAW_LARGE");
+
+        verify(template).send(eq("raw.large.DLT"), anyString(), anyString());
+    }
+
     @Test
     void influxDependencyFailurePausesWithoutDltStorm() {
         SyslogFailureRouter.TerminalState state = router.route(