From 44d327139b170fabf82633330369eb49f2a0210c Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E7=8E=8B=E7=BB=8D=E5=AE=8F?= <994233303@qq.com>
Date: Sun, 13 Sep 2026 04:04:47 +0800
Subject: [PATCH] =?UTF-8?q?feat(syslog-consumer-rule):=20=E8=A1=A5?=
 =?UTF-8?q?=E5=85=85=E6=85=A2=E6=89=B9=E6=AC=A1=E6=B6=88=E6=81=AF=E9=95=BF?=
 =?UTF-8?q?=E5=BA=A6=E7=BB=9F=E8=AE=A1?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 记录Kafka批次消费消息条数
- 记录批次日志正文总字节数
- 记录批次内单条日志最大字节数
- 优先使用Kafka序列化长度，避免重复编码带来的额外开销
---
 .../Modules/NormalData/SysLogProcessor.java   | 26 ++++++++++++++++---
 1 file changed, 22 insertions(+), 4 deletions(-)

diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java
index 617c937..b76d896 100644
--- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java
@@ -358,8 +358,19 @@ public class SysLogProcessor {
     }
 
     private void beginSlowBatchTrace(String lane, List<ConsumerRecord<String, String>> records) {
+        long polledMessageBytes = 0L;
+        int maxMessageBytes = 0;
+        for (ConsumerRecord<String, String> record : records) {
+            int messageBytes = record.serializedValueSize();
+            if (messageBytes < 0) {
+                String value = record.value();
+                messageBytes = value == null ? 0 : value.getBytes(StandardCharsets.UTF_8).length;
+            }
+            polledMessageBytes += messageBytes;
+            maxMessageBytes = Math.max(maxMessageBytes, messageBytes);
+        }
         slowBatchTrace.set(new SlowBatchTrace(lane, topicPartitions(records), records.size(),
-                System.nanoTime()));
+                polledMessageBytes, maxMessageBytes, System.nanoTime()));
     }
 
     private void recordSlowBatchStages(int records, int prepared, int persisted, int normalizedRows,
@@ -402,13 +413,15 @@ public class SysLogProcessor {
         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={}, "
+        log.warn("[SLOW_SYSLOG_BATCH] lane={}, topicPartitions={}, polledRecords={}, "
+                        + "polledMessageBytes={}, maxMessageBytes={}, 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.lane, trace.topicPartitions, trace.polledRecords,
+                trace.polledMessageBytes, trace.maxMessageBytes, trace.recordsProcessed,
                 trace.chunks, trace.prepared, trace.persisted, trace.normalizedRows,
                 trace.prepareMs, trace.ingressMs, trace.normalizeMs, trace.ruleComputeMs,
                 trace.messageParseMs, trace.fieldMappingMs, trace.persistencePlanMs,
@@ -1284,6 +1297,8 @@ public class SysLogProcessor {
         private final String lane;
         private final String topicPartitions;
         private final int polledRecords;
+        private final long polledMessageBytes;
+        private final int maxMessageBytes;
         private final long startedAtNanos;
         private int chunks;
         private int recordsProcessed;
@@ -1302,10 +1317,13 @@ public class SysLogProcessor {
         private long remainingDmMs;
 
         private SlowBatchTrace(String lane, String topicPartitions, int polledRecords,
+                               long polledMessageBytes, int maxMessageBytes,
                                long startedAtNanos) {
             this.lane = lane;
             this.topicPartitions = topicPartitions;
             this.polledRecords = polledRecords;
+            this.polledMessageBytes = polledMessageBytes;
+            this.maxMessageBytes = maxMessageBytes;
             this.startedAtNanos = startedAtNanos;
         }
     }