[FLINK-37730]Reverse the exception to get newer exceptions

Author: vsantwana

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java

@@ -38,7 +38,10 @@
 
 import java.time.Instant;
 import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeoutException;
 
@@ -129,8 +132,9 @@ protected void observeJobManagerExceptions(FlinkResourceContext<R> ctx) {
             }
 
             var exceptionHistory = history.getExceptionHistory();
-            var exceptions = exceptionHistory.getEntries();
-            if (exceptions.isEmpty()) {
+            List<JobExceptionsInfoWithHistory.RootExceptionInfo> exceptions =
+                    exceptionHistory.getEntries();
+            if (exceptions == null || exceptions.isEmpty()) {
                 return;
             }
 
@@ -157,13 +161,20 @@ protected void observeJobManagerExceptions(FlinkResourceContext<R> ctx) {
             int maxEvents = operatorConfig.getReportedExceptionEventsMaxCount();
             int maxStackTraceLines = operatorConfig.getReportedExceptionEventsMaxStackTraceLength();
 
+            // Sort and reverse to prioritize the newest exceptions
+            var sortedExceptions = new ArrayList<>(exceptions);
+            sortedExceptions.sort(
+                    Comparator.comparingLong(
+                                    JobExceptionsInfoWithHistory.RootExceptionInfo::getTimestamp)
+                            .reversed());
+
             int count = 0;
-            for (var exception : exceptions) {
+            for (var exception : sortedExceptions) {
                 Instant exceptionTime = Instant.ofEpochMilli(exception.getTimestamp());
+                // Skip already recorded exceptions
                 if (lastRecorded != null && exceptionTime.isBefore(lastRecorded)) {
                     continue;
                 }
-
                 emitJobManagerExceptionEvent(ctx, exception, exceptionTime, maxStackTraceLines);
                 if (++count >= maxEvents) {
                     break;
@@ -229,7 +240,6 @@ private void emitJobManagerExceptionEvent(
                 exceptionName.length() > 128 ? exceptionName.substring(0, 128) : exceptionName;
 
         eventRecorder.triggerEventOnceWithAnnotations(
-                exceptionTime.toEpochMilli(),
                 ctx.getResource(),
                 EventRecorder.Type.Warning,
                 EventRecorder.Reason.JobException,

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java

@@ -172,7 +172,6 @@ public boolean triggerEventOnce(
     }
 
     public boolean triggerEventOnceWithAnnotations(
-            long eventTimestampMillis,
             AbstractFlinkResource<?, ?> resource,
             Type type,
             Reason reason,
@@ -184,7 +183,6 @@ public boolean triggerEventOnceWithAnnotations(
         return EventUtils.createWithAnnotationsIfNotExists(
                 client,
                 resource,
-                eventTimestampMillis,
                 type,
                 reason.toString(),
                 message,

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventUtils.java

@@ -23,7 +23,6 @@
 import io.fabric8.kubernetes.api.model.Event;
 import io.fabric8.kubernetes.api.model.EventBuilder;
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.fabric8.kubernetes.api.model.MicroTime;
 import io.fabric8.kubernetes.api.model.ObjectMeta;
 import io.fabric8.kubernetes.api.model.ObjectReference;
 import io.fabric8.kubernetes.api.model.Pod;
@@ -38,8 +37,6 @@
 import java.net.HttpURLConnection;
 import java.time.Duration;
 import java.time.Instant;
-import java.time.ZoneId;
-import java.time.ZonedDateTime;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -110,7 +107,6 @@ public static Event findExistingEvent(
     public static boolean createWithAnnotationsIfNotExists(
             KubernetesClient client,
             HasMetadata target,
-            long eventTime,
             EventRecorder.Type type,
             String reason,
             String message,
@@ -128,7 +124,6 @@ public static boolean createWithAnnotationsIfNotExists(
         } else {
             Event event = buildEvent(target, type, reason, message, component, eventName);
             setAnnotations(event, annotations);
-            setEventTime(event, Instant.ofEpochMilli(eventTime));
             createOrReplaceEvent(client, event).ifPresent(eventListener);
             return true;
         }
@@ -222,12 +217,6 @@ private static void setAnnotations(Event existing, @Nullable Map<String, String>
         existing.getMetadata().setAnnotations(annotations);
     }
 
-    private static void setEventTime(Event existing, Instant eventTime) {
-        ZonedDateTime zonedDateTime = ZonedDateTime.ofInstant(eventTime, ZoneId.of("UTC"));
-        MicroTime microTime = new MicroTime(zonedDateTime.toString());
-        existing.setEventTime(microTime);
-    }
-
     private static Event buildEvent(
             HasMetadata target,
             EventRecorder.Type type,

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/EventUtilsTest.java

@@ -600,7 +600,6 @@ public void accept(Event event) {
                 EventUtils.createWithAnnotationsIfNotExists(
                         kubernetesClient,
                         flinkApp,
-                        System.currentTimeMillis(),
                         EventRecorder.Type.Normal,
                         reason,
                         message,
@@ -631,7 +630,6 @@ public void accept(Event event) {
                 EventUtils.createWithAnnotationsIfNotExists(
                         kubernetesClient,
                         flinkApp,
-                        System.currentTimeMillis(),
                         EventRecorder.Type.Normal,
                         reason,
                         message,