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> records) { + long polledMessageBytes = 0L; + int maxMessageBytes = 0; + for (ConsumerRecord 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; } }