[FLINK-34957][autoscaler] Event handler records the exception stack trace when exception message is null

Author: 1996fanrui

flink-autoscaler-standalone/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();
         }

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) {

flink-autoscaler/src/main/java/org/apache/flink/autoscaler/event/AutoScalerEventHandler.java

@@ -22,6 +22,9 @@
 import org.apache.flink.autoscaler.ScalingSummary;
 import org.apache.flink.runtime.jobgraph.JobVertexID;
 
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+
 import javax.annotation.Nullable;
 
 import java.time.Duration;
@@ -63,6 +66,17 @@ 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) {
+            message = StringUtils.abbreviate(ExceptionUtils.getStackTrace(e), 2048);
+        }
+        handleEvent(context, Type.Warning, reason, message, null, null);
+    }
+
     /**
      * Handle scaling reports.
      *