[FLINK-34957][autoscaler] Event handler records the exception stack t…

…race when exception message is null
1996fanrui · Mar 28, 2024 · 1949f30 · 1949f30
1 parent 96f8b20
commit 1949f30
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 14 deletions.
diff --git a/...ne/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutor.java b/...ne/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutor.java
@@ -141,13 +141,7 @@ protected void scalingSingleJob(Context jobContext) {
             autoScaler.scale(jobContext);
         } catch (Throwable e) {
             LOG.error("Error while scaling job", e);
-            eventHandler.handleEvent(
-                    jobContext,
-                    AutoScalerEventHandler.Type.Warning,
-                    AUTOSCALER_ERROR,
-                    e.getMessage(),
-                    null,
-                    null);
+            eventHandler.handleException(jobContext, AUTOSCALER_ERROR, e);
         } finally {
             MDC.clear();
         }

diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java
@@ -228,13 +228,7 @@ private void runScalingLogic(Context ctx, AutoscalerFlinkMetrics autoscalerMetri
     private void onError(Context ctx, AutoscalerFlinkMetrics autoscalerMetrics, Throwable e) {
         LOG.error("Error while scaling job", e);
         autoscalerMetrics.incrementError();
-        eventHandler.handleEvent(
-                ctx,
-                AutoScalerEventHandler.Type.Warning,
-                AUTOSCALER_ERROR,
-                e.getMessage(),
-                null,
-                null);
+        eventHandler.handleException(ctx, AUTOSCALER_ERROR, e);
     }
 
     private AutoscalerFlinkMetrics getOrInitAutoscalerFlinkMetrics(Context ctx) {

diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/event/AutoScalerEventHandler.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/event/AutoScalerEventHandler.java
@@ -24,6 +24,8 @@
 
 import javax.annotation.Nullable;
 
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
 import java.time.Duration;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -63,6 +65,19 @@ void handleEvent(
             @Nullable String messageKey,
             @Nullable Duration interval);
 
+    /**
+     * Handle exception, and the exception event is warning type and don't deduplicate by default.
+     */
+    default void handleException(Context context, String reason, Throwable e) {
+        var message = e.getMessage();
+        if (message == null) {
+            var stream = new ByteArrayOutputStream();
+            e.printStackTrace(new PrintStream(stream));
+            message = stream.toString();
+        }
+        handleEvent(context, Type.Warning, reason, message, null, null);
+    }
+
     /**
      * Handle scaling reports.
      *