fix: kafka send stat (#871)

Co-authored-by: saimu.msm <saimu.msm@antfin.com>
traas-stack · Jul 1, 2024 · 18199ce · 18199ce
1 parent 2d64ab8
commit 18199ce
Showing 1 changed file with 7 additions and 1 deletion.
diff --git a/...g/agg-dispatcher/src/main/java/io/holoinsight/server/agg/v1/dispatcher/AggDispatcher.java b/...g/agg-dispatcher/src/main/java/io/holoinsight/server/agg/v1/dispatcher/AggDispatcher.java
@@ -31,6 +31,7 @@
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.errors.RecordTooLargeException;
 import org.jetbrains.annotations.NotNull;
 import org.springframework.beans.factory.annotation.Autowired;
 
@@ -314,7 +315,12 @@ private boolean send(AggTaskKey key, AggProtos.AggTaskValue value) {
       if (value.hasDataTable()) {
         count += value.getDataTable().getRowCount();
       }
-      if (exception != null) {
+      if (exception instanceof RecordTooLargeException) {
+        log.warn("[agg] [{}] write kafka error for {}, metric=[{}] size=[{}]", //
+            key, exception.getMessage(), value.getMetric(), count);
+        StatUtils.KAFKA_SEND.add(StringsKey.of(topic, "DISCARD", "unknownPartition"),
+            new long[] {1, count, 0});
+      } else if (exception != null) {
         log.error("[agg] [{}] write kafka error, metric=[{}] size=[{}]", //
             key, value.getMetric(), count, exception);
         StatUtils.KAFKA_SEND.add(StringsKey.of(topic, "ERROR", "unknownPartition"),