[FLINK-31860] FlinkDeployments never finalize when namespace is deleted

Author: jiangzho

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

@@ -23,12 +23,17 @@
 import io.fabric8.kubernetes.api.model.ObjectMeta;
 import io.fabric8.kubernetes.api.model.ObjectReferenceBuilder;
 import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nullable;
 
+import java.net.HttpURLConnection;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.Map;
+import java.util.Optional;
 import java.util.function.Consumer;
 import java.util.function.Predicate;
 
@@ -37,6 +42,7 @@
  * https://github.com/EnMasseProject/enmasse/blob/master/k8s-api/src/main/java/io/enmasse/k8s/api/KubeEventLogger.java
  */
 public class EventUtils {
+    private static final Logger LOG = LoggerFactory.getLogger(EventUtils.class);
 
     public static String generateEventName(
             HasMetadata target,
@@ -108,7 +114,7 @@ public static boolean createIfNotExists(
             return false;
         } else {
             Event event = buildEvent(target, type, reason, message, component, eventName);
-            eventListener.accept(client.resource(event).createOrReplace());
+            createOrReplaceEvent(client, event).ifPresent(eventListener);
             return true;
         }
     }
@@ -139,7 +145,7 @@ public static boolean createOrUpdateEventWithLabels(
         } else {
             Event event = buildEvent(target, type, reason, message, component, eventName);
             setLabels(event, labels);
-            eventListener.accept(client.resource(event).createOrReplace());
+            createOrReplaceEvent(client, event).ifPresent(eventListener);
             return true;
         }
     }
@@ -154,7 +160,7 @@ private static void updatedEventWithLabels(
         existing.setCount(existing.getCount() + 1);
         existing.setMessage(message);
         setLabels(existing, labels);
-        eventListener.accept(client.resource(existing).createOrReplace());
+        createOrReplaceEvent(client, existing).ifPresent(eventListener);
     }
 
     private static void setLabels(Event existing, @Nullable Map<String, String> labels) {
@@ -213,4 +219,20 @@ private static boolean labelCheck(
                 || (existing.getMetadata() != null
                         && dedupePredicate.test(existing.getMetadata().getLabels()));
     }
+
+    private static Optional<Event> createOrReplaceEvent(KubernetesClient client, Event event) {
+        try {
+            Event createdEvent = client.resource(event).createOrReplace();
+            return Optional.of(createdEvent);
+        } catch (KubernetesClientException e) {
+            if (e.getCode() == HttpURLConnection.HTTP_FORBIDDEN) {
+                // fail the reconcile when events cannot be delivered, unless FORBIDDEN
+                // which can be a result of recoverable rbac issue, or namespace is terminating
+                LOG.warn("Cannot create or update events, proceeding.", e);
+            } else {
+                throw e;
+            }
+        }
+        return Optional.empty();
+    }
 }

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

@@ -20,6 +20,7 @@
 import org.apache.flink.kubernetes.operator.TestUtils;
 
 import io.fabric8.kubernetes.api.model.Event;
+import io.fabric8.kubernetes.api.model.EventBuilder;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
 import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
@@ -30,6 +31,7 @@
 
 import javax.annotation.Nullable;
 
+import java.net.HttpURLConnection;
 import java.time.Duration;
 import java.util.Map;
 import java.util.function.Consumer;
@@ -525,4 +527,49 @@ public void accept(Event event) {
         Assertions.assertEquals(1, event.getCount());
         Assertions.assertNull(eventConsumed);
     }
+
+    @Test
+    public void testCreateOrReplaceEventOnDeletedNamespace() {
+        var consumer =
+                new Consumer<Event>() {
+                    @Override
+                    public void accept(Event event) {
+                        eventConsumed = event;
+                    }
+                };
+        var flinkApp = TestUtils.buildApplicationCluster();
+        var reason = "Cleanup";
+        var message = "message";
+        var eventName =
+                EventUtils.generateEventName(
+                        flinkApp,
+                        EventRecorder.Type.Warning,
+                        reason,
+                        message,
+                        EventRecorder.Component.Operator);
+
+        var namespaceName = flinkApp.getMetadata().getNamespace();
+
+        String eventCreatePath = String.format("/api/v1/namespaces/%s/events", namespaceName);
+
+        mockServer
+                .expect()
+                .post()
+                .withPath(eventCreatePath)
+                .andReturn(HttpURLConnection.HTTP_FORBIDDEN, new EventBuilder().build())
+                .once();
+
+        Assertions.assertTrue(
+                EventUtils.createOrUpdateEventWithInterval(
+                        kubernetesClient,
+                        flinkApp,
+                        EventRecorder.Type.Warning,
+                        reason,
+                        message,
+                        EventRecorder.Component.Operator,
+                        consumer,
+                        null,
+                        null));
+        Assertions.assertNull(eventConsumed);
+    }
 }