feat(syslog-consumer-rule): 增加慢批次全链路耗时诊断

- 批次端到端耗时达到500ms时输出SLOW_SYSLOG_BATCH日志
- 记录入口统计、规则计算、InfluxDB和达梦写入的分阶段耗时
- 使用MANUAL_IMMEDIATE模式准确统计Kafka offset同步提交耗时
- 记录Kafka提交状态、异常类型和端到端总耗时
- 在常规批次日志中补充ingressMs
- 增加Kafka监听器ACK模式回归测试

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

@@ -74,6 +74,12 @@ public class SysLogProcessor {
     @Value("${syslog.failure-continuity.enabled:true}")
     private boolean failureContinuityEnabled;
 
+    @Value("${syslog.slow-batch.threshold-ms:500}")
+    private long slowBatchThresholdMs;
+
+    /** listener、业务处理和 Kafka commit 均在同一个 consumer 线程，ThreadLocal 用于拼接端到端慢批次。 */
+    private final ThreadLocal<SlowBatchTrace> slowBatchTrace = new ThreadLocal<>();
+
     @Autowired
     private AppProperties appProperties;
 
@@ -186,11 +192,18 @@ public class SysLogProcessor {
             ack.acknowledge();
             return;
         }
+        beginSlowBatchTrace(lane, records);
+        long processingStartedAt = System.nanoTime();
+        try {
             for (List<ConsumerRecord<String, String>> chunk : partitionRecordsByBytes(
                     records, Math.max(1L, largeProcessingBatchMaxBytes))) {
                 processBatchWithContinuity(chunk, enforceRetryDue, lane, largeTaskExecutor);
             }
-        ack.acknowledge();
+            long processingMs = elapsedMillis(processingStartedAt);
+            acknowledgeAndCompleteSlowBatchTrace(ack, processingMs);
+        } finally {
+            slowBatchTrace.remove();
+        }
     }
 
     @KafkaListener(
@@ -220,8 +233,15 @@ public class SysLogProcessor {
             ack.acknowledge();
             return;
         }
+        beginSlowBatchTrace(lane, records);
+        long processingStartedAt = System.nanoTime();
+        try {
             processBatchWithContinuity(records, enforceRetryDue, lane, normalizationExecutor);
-        ack.acknowledge();
+            long processingMs = elapsedMillis(processingStartedAt);
+            acknowledgeAndCompleteSlowBatchTrace(ack, processingMs);
+        } finally {
+            slowBatchTrace.remove();
+        }
     }
 
     private void processBatchWithContinuity(List<ConsumerRecord<String, String>> records,
@@ -263,7 +283,9 @@ public class SysLogProcessor {
         long prepareMs = System.currentTimeMillis() - prepareStartedAt;
 
         // 入口状态是 Kafka 已可靠接收的事实。批内按设备/探针聚合，失败时禁止 ACK；重投只重复 MIN/MAX。
+        long ingressStartedAt = System.nanoTime();
         int ingressGroups = ingressStatsProjectionWriter.writeBatch(ingressEvents(preparedMessages));
+        long ingressMs = elapsedMillis(ingressStartedAt);
 
         // 必须先完成规则计算，才能确保告警全量保留，并让未保留消息同时跳过 InfluxDB 和达梦。
         long normalizeStartedAt = System.currentTimeMillis();
@@ -283,8 +305,10 @@ public class SysLogProcessor {
         long normalizeMs = System.currentTimeMillis() - normalizeStartedAt;
 
         // 告警是最高优先级：先独立提交告警表，不能让 InfluxDB 或普通表的慢写阻塞告警入库。
+        long persistencePlanStartedAt = System.nanoTime();
         StandardizedPersistencePlan standardizedPlan = prepareStandardizedPersistence(
                 samplingPlan.normalizations, metrics);
+        long persistencePlanMs = elapsedMillis(persistencePlanStartedAt);
         long alarmWriteStartedAt = System.nanoTime();
         DmStandardizedBatchWriter.WriteReport<PreparedMessage> alarmReport =
                 standardizedBatchWriter.write(standardizedPlan.alarmWriteRequests);
@@ -295,14 +319,20 @@ public class SysLogProcessor {
                 samplingPlan.rawMessages, metrics, lane);
         long influxMs = System.currentTimeMillis() - influxStartedAt;
 
+        long remainingDmStartedAt = System.nanoTime();
         persistRemainingStandardizedData(standardizedPlan, alarmReport, alarmWriteMs, metrics);
+        long remainingDmMs = elapsedMillis(remainingDmStartedAt);
 
         totalProcessed.addAndGet(records.size());
+        long businessTotalMs = System.currentTimeMillis() - batchStartedAt;
+        recordSlowBatchStages(records.size(), preparedMessages.size(), persistedMessages.size(),
+                metrics.normalizedRows.get(), prepareMs, ingressMs, normalizeMs,
+                persistencePlanMs, alarmWriteMs, influxMs, remainingDmMs, metrics);
         log.info("Syslog批次完成: lane={}, topicPartitions={}, records={}, prepared={}, persisted={}, "
                         + "success={}, filtered={}, retry={}, dlt={}, normalizedRows={}, dmSuccessfulRows={}, "
                         + "dmFailedMessages={}, ingressGroups={}, rawStorageBypassed={}, backlogDegraded={}, "
                         + "rawSampledOut={}, normalRowsSampledOut={}, nonNormalRowsSampledOut={}, "
-                        + "prepareMs={}, influxMs={}, normalizeMs={}, "
+                        + "prepareMs={}, ingressMs={}, influxMs={}, normalizeMs={}, "
                         + "ruleComputeMs={}, ruleSnapshotMs={}, messageParseMs={}, fieldMappingMs={}, "
                         + "filterCompleteMs={}, enrichmentMs={}, ruleAssembleMs={}, compiledRuleHit={}, "
                         + "compiledRuleMiss={}, messageParseHit={}, messageParseMiss={}, deviceUnique={}, "
@@ -315,7 +345,7 @@ public class SysLogProcessor {
                 metrics.rawStorageBypassed.get(),
                 metrics.backlogDegraded.get(), metrics.rawSampledOut.get(),
                 metrics.normalRowsSampledOut.get(), metrics.nonNormalRowsSampledOut.get(),
-                prepareMs, influxMs, normalizeMs, metrics.ruleComputeMs(), metrics.ruleSnapshotMs(),
+                prepareMs, ingressMs, influxMs, normalizeMs, metrics.ruleComputeMs(), metrics.ruleSnapshotMs(),
                 metrics.messageParseMs(), metrics.fieldMappingMs(), metrics.filterCompleteMs(),
                 metrics.enrichmentMs(), metrics.ruleAssembleMs(), metrics.compiledRuleHits(),
                 metrics.compiledRuleMisses(), metrics.messageParseHits(), metrics.messageParseMisses(),
@@ -324,7 +354,67 @@ public class SysLogProcessor {
                 metrics.nonNormalQueued(), metrics.nonNormalRejected(),
                 nonNormalAsyncCompletedTotal.get(), nonNormalAsyncFailedTotal.get(),
                 nonNormalAsyncRejectedTotal.get(),
-                System.currentTimeMillis() - batchStartedAt);
+                businessTotalMs);
+    }
+
+    private void beginSlowBatchTrace(String lane, List<ConsumerRecord<String, String>> records) {
+        slowBatchTrace.set(new SlowBatchTrace(lane, topicPartitions(records), records.size(),
+                System.nanoTime()));
+    }
+
+    private void recordSlowBatchStages(int records, int prepared, int persisted, int normalizedRows,
+                                       long prepareMs, long ingressMs, long normalizeMs,
+                                       long persistencePlanMs, long alarmDmMs, long influxMs,
+                                       long remainingDmMs, BatchMetrics metrics) {
+        SlowBatchTrace trace = slowBatchTrace.get();
+        if (trace == null) return;
+        trace.chunks++;
+        trace.recordsProcessed += records;
+        trace.prepared += prepared;
+        trace.persisted += persisted;
+        trace.normalizedRows += normalizedRows;
+        trace.prepareMs += prepareMs;
+        trace.ingressMs += ingressMs;
+        trace.normalizeMs += normalizeMs;
+        trace.persistencePlanMs += persistencePlanMs;
+        trace.alarmDmMs += alarmDmMs;
+        trace.influxMs += influxMs;
+        trace.remainingDmMs += remainingDmMs;
+        trace.ruleComputeMs += metrics.ruleComputeMs();
+        trace.messageParseMs += metrics.messageParseMs();
+        trace.fieldMappingMs += metrics.fieldMappingMs();
+    }
+
+    private void acknowledgeAndCompleteSlowBatchTrace(Acknowledgment ack, long businessProcessMs) {
+        long commitStartedAt = System.nanoTime();
+        try {
+            ack.acknowledge();
+            completeSlowBatchTrace(businessProcessMs, elapsedMillis(commitStartedAt), null);
+        } catch (RuntimeException commitFailure) {
+            completeSlowBatchTrace(businessProcessMs, elapsedMillis(commitStartedAt), commitFailure);
+            throw commitFailure;
+        }
+    }
+
+    private void completeSlowBatchTrace(long businessProcessMs, long kafkaCommitMs,
+                                        RuntimeException commitFailure) {
+        SlowBatchTrace trace = slowBatchTrace.get();
+        if (trace == null) return;
+        long endToEndMs = elapsedMillis(trace.startedAtNanos);
+        if (slowBatchThresholdMs <= 0L || endToEndMs < slowBatchThresholdMs) return;
+        log.warn("[SLOW_SYSLOG_BATCH] lane={}, topicPartitions={}, polledRecords={}, processedRecords={}, "
+                        + "chunks={}, prepared={}, persisted={}, normalizedRows={}, prepareMs={}, ingressMs={}, "
+                        + "normalizeMs={}, ruleComputeWorkerMs={}, messageParseWorkerMs={}, "
+                        + "fieldMappingWorkerMs={}, persistencePlanMs={}, alarmDmWriteMs={}, influxMs={}, "
+                        + "remainingDmWriteMs={}, businessProcessMs={}, kafkaCommitMs={}, "
+                        + "kafkaCommitStatus={}, kafkaCommitErrorType={}, endToEndMs={}",
+                trace.lane, trace.topicPartitions, trace.polledRecords, trace.recordsProcessed,
+                trace.chunks, trace.prepared, trace.persisted, trace.normalizedRows,
+                trace.prepareMs, trace.ingressMs, trace.normalizeMs, trace.ruleComputeMs,
+                trace.messageParseMs, trace.fieldMappingMs, trace.persistencePlanMs,
+                trace.alarmDmMs, trace.influxMs, trace.remainingDmMs,
+                businessProcessMs, kafkaCommitMs, commitFailure == null ? "SUCCESS" : "FAILED",
+                commitFailure == null ? "none" : commitFailure.getClass().getName(), endToEndMs);
     }
 
     private void routeContinuityBatch(List<ConsumerRecord<String, String>> records,
@@ -1190,6 +1280,36 @@ public class SysLogProcessor {
         }
     }
 
+    private static final class SlowBatchTrace {
+        private final String lane;
+        private final String topicPartitions;
+        private final int polledRecords;
+        private final long startedAtNanos;
+        private int chunks;
+        private int recordsProcessed;
+        private int prepared;
+        private int persisted;
+        private int normalizedRows;
+        private long prepareMs;
+        private long ingressMs;
+        private long normalizeMs;
+        private long ruleComputeMs;
+        private long messageParseMs;
+        private long fieldMappingMs;
+        private long persistencePlanMs;
+        private long alarmDmMs;
+        private long influxMs;
+        private long remainingDmMs;
+
+        private SlowBatchTrace(String lane, String topicPartitions, int polledRecords,
+                               long startedAtNanos) {
+            this.lane = lane;
+            this.topicPartitions = topicPartitions;
+            this.polledRecords = polledRecords;
+            this.startedAtNanos = startedAtNanos;
+        }
+    }
+
     /** 单批终态计数，仅用于聚合日志，不参与 ACK 决策。 */
     static final class BatchMetrics {
         private final AtomicInteger success = new AtomicInteger();

haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/KafkaReliabilityConfig.java

@@ -9,6 +9,7 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.listener.DefaultErrorHandler;
 import org.springframework.util.backoff.FixedBackOff;
 
@@ -33,7 +34,11 @@ public class KafkaReliabilityConfig {
             ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
             ConsumerFactory<Object, Object> consumerFactory,
             DefaultErrorHandler syslogDefaultErrorHandler) {
-        return configuredFactory(configurer, consumerFactory, syslogDefaultErrorHandler);
+        ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
+                configuredFactory(configurer, consumerFactory, syslogDefaultErrorHandler);
+        // 默认工厂承载普通 Retry/Replay listener，同样需要让 acknowledge() 覆盖真实提交耗时。
+        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
+        return factory;
     }
 
     /**
@@ -52,6 +57,9 @@ public class KafkaReliabilityConfig {
         ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
                 configuredFactory(configurer, consumerFactory, syslogDefaultErrorHandler);
         factory.setConcurrency(concurrency);
+        // 在 listener 线程调用 acknowledge() 时立即同步提交，使慢批次日志能够准确覆盖
+        // “业务处理完成 -> Kafka offset 提交完成”的全部时间；可靠性语义仍是处理成功后提交。
+        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
         return factory;
     }
 
@@ -78,6 +86,7 @@ public class KafkaReliabilityConfig {
         ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
                 configuredFactory(configurer, largeConsumerFactory, syslogDefaultErrorHandler);
         factory.setConcurrency(concurrency);
+        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
         return factory;
     }
 

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

@@ -82,6 +82,8 @@ 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}
+# 端到端批次耗时达到该阈值时，打印业务各阶段及 Kafka offset 同步提交耗时；0 表示关闭。
+syslog.slow-batch.threshold-ms=${SYSLOG_SLOW_BATCH_THRESHOLD_MS:500}
 # 连续性优先：特殊/系统性异常优先写 DLT；即使 DLT 失败也记录告警并推进原 offset，避免阻塞分区。
 syslog.failure-continuity.enabled=${SYSLOG_FAILURE_CONTINUITY_ENABLED:true}
 # DLT 故障时最多快速等待 1 秒；首次失败后整批直接记录并放行。

haobang-security-dm/syslog-consumer-rule/src/test/java/com/config/KafkaReliabilityConfigTest.java

@@ -0,0 +1,67 @@
+package com.config;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.listener.ContainerProperties;
+import org.springframework.kafka.listener.DefaultErrorHandler;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class KafkaReliabilityConfigTest {
+
+    @Test
+    void retryListenerCommitsInsideAcknowledgment() {
+        KafkaReliabilityConfig config = new KafkaReliabilityConfig();
+        ConcurrentKafkaListenerContainerFactoryConfigurer configurer =
+                mock(ConcurrentKafkaListenerContainerFactoryConfigurer.class);
+        @SuppressWarnings("unchecked")
+        ConsumerFactory<Object, Object> consumerFactory = mock(ConsumerFactory.class);
+
+        ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
+                config.kafkaListenerContainerFactory(configurer, consumerFactory,
+                        config.syslogDefaultErrorHandler());
+
+        assertEquals(ContainerProperties.AckMode.MANUAL_IMMEDIATE,
+                factory.getContainerProperties().getAckMode());
+    }
+
+    @Test
+    void rawListenerCommitsInsideAcknowledgmentForAccurateSlowBatchTiming() {
+        KafkaReliabilityConfig config = new KafkaReliabilityConfig();
+        ConcurrentKafkaListenerContainerFactoryConfigurer configurer =
+                mock(ConcurrentKafkaListenerContainerFactoryConfigurer.class);
+        @SuppressWarnings("unchecked")
+        ConsumerFactory<Object, Object> consumerFactory = mock(ConsumerFactory.class);
+        DefaultErrorHandler errorHandler = config.syslogDefaultErrorHandler();
+
+        ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
+                config.rawKafkaListenerContainerFactory(
+                        configurer, consumerFactory, errorHandler, 3);
+
+        assertEquals(ContainerProperties.AckMode.MANUAL_IMMEDIATE,
+                factory.getContainerProperties().getAckMode());
+    }
+
+    @Test
+    void largeListenerAlsoCommitsInsideAcknowledgment() {
+        KafkaReliabilityConfig config = new KafkaReliabilityConfig();
+        ConcurrentKafkaListenerContainerFactoryConfigurer configurer =
+                mock(ConcurrentKafkaListenerContainerFactoryConfigurer.class);
+        @SuppressWarnings("unchecked")
+        ConsumerFactory<Object, Object> consumerFactory = mock(ConsumerFactory.class);
+        when(consumerFactory.getConfigurationProperties()).thenReturn(Collections.emptyMap());
+
+        ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
+                config.largeRawKafkaListenerContainerFactory(configurer, consumerFactory,
+                        config.syslogDefaultErrorHandler(), 1, 5, 33554432, 67108864);
+
+        assertEquals(ContainerProperties.AckMode.MANUAL_IMMEDIATE,
+                factory.getContainerProperties().getAckMode());
+    }
+}