Not fail operator if an unknown job found in cluster

src/Exceptions/JobListenerException.cs

@@ -0,0 +1,17 @@
+using System;
+
+namespace Arcane.Operator.Exceptions;
+
+/// <summary>
+/// Thrown when an error occurs in the job listener
+/// </summary>
+public class JobListenerException: Exception
+{
+    /// <summary>
+    /// Thrown when an error occurs in the job listener
+    /// </summary>
+    /// <param name="message">Error message</param>
+    public JobListenerException(string message) : base(message)
+    {
+    }
+}

src/Extensions/V1JobExtensions.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using Arcane.Operator.Contracts;
+using Arcane.Operator.Exceptions;
 using Arcane.Operator.Models.Api;
 using Arcane.Operator.Models.Resources.StreamClass.Base;
 using Arcane.Operator.Models.StreamDefinitions.Base;
@@ -128,7 +129,7 @@ private static string GetApiGroup(this V1Job job)
             return value;
         }
 
-        throw new InvalidOperationException("Api group not found in job annotations.");
+        throw new JobListenerException("Api group not found in job annotations.");
     }
 
     private static string GetApiVersion(this V1Job job)
@@ -138,7 +139,7 @@ private static string GetApiVersion(this V1Job job)
             return value;
         }
 
-        throw new InvalidOperationException("Api version not found in job annotations.");
+        throw new JobListenerException("Api version not found in job annotations.");
     }
 
     private static string GetPluralName(this V1Job job)
@@ -148,6 +149,6 @@ private static string GetPluralName(this V1Job job)
             return value;
         }
 
-        throw new InvalidOperationException("Api plural name version not found in job annotations.");
+        throw new JobListenerException("Api plural name version not found in job annotations.");
     }
 }

src/Services/Operators/StreamingJobOperatorService.cs

@@ -6,6 +6,7 @@
 using Akka.Streams.Dsl;
 using Akka.Util;
 using Arcane.Operator.Configurations;
+using Arcane.Operator.Exceptions;
 using Arcane.Operator.Extensions;
 using Arcane.Operator.Models.Api;
 using Arcane.Operator.Models.Base;
@@ -23,6 +24,8 @@
 using Snd.Sdk.ActorProviders;
 using Snd.Sdk.Kubernetes;
 using Snd.Sdk.Tasks;
+using Akka.Streams.Supervision;
+
 
 namespace Arcane.Operator.Services.Operators;
 
@@ -70,6 +73,7 @@ public IRunnableGraph<Task> GetJobEventsGraph(CancellationToken cancellationToke
             .SelectAsync(parallelism, this.OnJobEvent)
             .SelectMany(e => e)
             .CollectOption()
+            .WithAttributes(ActorAttributes.CreateSupervisionStrategy(this.HandleError))
             .ToMaterialized(Sink.ForEachAsync<KubernetesCommand>(parallelism, this.HandleCommand), Keep.Right);
     }
 
@@ -159,4 +163,26 @@ private Task<List<Option<KubernetesCommand>>> OnJobDelete(V1Job job)
         RemoveAnnotationCommand<IStreamDefinition> command => this.removeAnnotationHandler.Handle(command),
         _ => throw new ArgumentOutOfRangeException(nameof(response), response, null)
     };
+
+    private Directive HandleError(Exception exception)
+    {
+        this.logger.LogError(exception, "Failed to handle stream definition event");
+        return exception switch
+        {
+            JobListenerException ex => this.LogAndContinue(ex),
+            _ => this.LogAndStop(exception),
+        };
+    }
+
+    private Directive LogAndContinue(Exception exception)
+    {
+        this.logger.LogWarning(exception, "Failed to handle stream definition event");
+        return Directive.Resume;
+    }
+
+    private Directive LogAndStop(Exception exception)
+    {
+        this.logger.LogError(exception, "Failed to handle stream definition event");
+        return Directive.Resume;
+    }
 }