feat(syslog-consumer-rule): 补充慢批次消息长度统计

- 记录Kafka批次消费消息条数
- 记录批次日志正文总字节数
- 记录批次内单条日志最大字节数
- 优先使用Kafka序列化长度，避免重复编码带来的额外开销

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;
         }
     }