feat: fail on error option for the default event recorder

docs/content/en/docs/documentation/operations/configuration.md

@@ -386,8 +386,24 @@ Recording events requires the operator's service account to be allowed `get`, `c
 created, and an event that does is a repeat of one already recorded, which is patched to count the
 new occurrence instead of being recorded a second time. The
 [generic Helm chart](helm-chart.md) grants all three. If you write your own RBAC and a permission is
-missing, recording fails silently as far as reconciliation is concerned: the failure is swallowed so
-that it cannot break a reconciliation, and only shows up as a warning in the operator log.
+missing, recording fails silently as far as reconciliation is concerned by default: the failure is
+swallowed so that it cannot break a reconciliation, and only shows up as a warning in the operator
+log. Kubernetes treats events as best effort, supplemental data, so this is usually what you want.
+Where an event that is not recorded must not go unnoticed, configure a recorder that propagates the
+failure to the reconciler instead, so that the reconciliation fails and is retried like any other
+failed one:
+
+```java
+KubernetesClient client = new KubernetesClientBuilder().build();
+Operator operator = new Operator(override -> override
+        .withKubernetesClient(client)
+        .withEventRecorder(DefaultEventRecorder.builder(new DefaultEventSink(client))
+            .failOnError(true)
+            .build()));
+```
+
+Every failure propagates then, transient ones included, such as the conflict two replicas run into
+when they count the same occurrence of an event at the same time.
 
 Note that events live in a namespace of their own choosing, which is the namespace of the object
 they are about, and for cluster scoped objects the namespace configured above. A namespaced `Role`
@@ -398,7 +414,7 @@ cluster scoped resources. A `ClusterRole` covers all of them at once.
 The API server only accepts events about cluster scoped objects in the `default` and `kube-system`
 namespaces, so those are the only values `josdk.events.cluster-scoped-namespace` can usefully be set
 to. In particular, the namespace the operator itself runs in is rejected, and the event dropped with
-a warning in the operator log.
+a warning in the operator log (or, with `failOnError(true)`, the reconciliation failed).
 
 #### Leader Election
 

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java

@@ -391,9 +391,9 @@ default Optional<LeaderElectionConfiguration> getLeaderElectionConfiguration() {
    * recorded by the built-in controllers. Operators whose RBAC does not allow creating events in
    * the {@code default} namespace can override this with {@code kube-system}. The API server
    * rejects events about cluster scoped resources in any other namespace, such as the namespace the
-   * operator itself runs in. Note that recording an event is best effort, so a missing permission
-   * or a rejected namespace results in the event being dropped with a warning in the log rather
-   * than in an error.
+   * operator itself runs in. Note that recording an event is best effort by default, so a missing
+   * permission or a rejected namespace results in the event being dropped with a warning in the log
+   * rather than in an error.
    *
    * @return the namespace to record events about cluster scoped resources in
    */

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/DefaultEventRecorder.java

@@ -58,6 +58,11 @@
  * event already recorded for it. Repeat occurrences are then counted on that event rather than
  * recorded as copies of it, see {@link DefaultEventSink}. How events aggregate, how they are named
  * and whether they carry an owner reference can be configured, see {@link #builder(EventSink)}.
+ *
+ * <p>Recording is best effort by default: an event that cannot be written is logged and dropped,
+ * and the caller carries on. A recorder built with {@link Builder#failOnError(boolean)} propagates
+ * the failure to the caller instead, so that a reconciliation that cannot record its events fails
+ * and is retried.
  */
 public class DefaultEventRecorder implements EventRecorder {
 
@@ -95,20 +100,23 @@ public class DefaultEventRecorder implements EventRecorder {
   private final EventNamingStrategy namingStrategy;
   private final EventKeyStrategy keyStrategy;
   private final boolean ownerReference;
+  private final boolean failOnError;
 
   public DefaultEventRecorder(EventSink sink) {
-    this(sink, EventNamingStrategy.none(), EventKeyStrategy.none(), false);
+    this(sink, EventNamingStrategy.none(), EventKeyStrategy.none(), false, false);
   }
 
   private DefaultEventRecorder(
       EventSink sink,
       EventNamingStrategy namingStrategy,
       EventKeyStrategy keyStrategy,
-      boolean ownerReference) {
+      boolean ownerReference,
+      boolean failOnError) {
     this.sink = sink;
     this.namingStrategy = namingStrategy;
     this.keyStrategy = keyStrategy;
     this.ownerReference = ownerReference;
+    this.failOnError = failOnError;
   }
 
   public static Builder builder(EventSink sink) {
@@ -122,6 +130,7 @@ public static final class Builder {
     private EventNamingStrategy namingStrategy = EventNamingStrategy.none();
     private EventKeyStrategy keyStrategy = EventKeyStrategy.none();
     private boolean ownerReference = false;
+    private boolean failOnError = false;
 
     private Builder(EventSink sink) {
       this.sink = Objects.requireNonNull(sink, "sink must not be null");
@@ -157,8 +166,26 @@ public Builder ownerReference(boolean ownerReference) {
       return this;
     }
 
+    /**
+     * When set, a failure to record an event is propagated to the caller rather than logged and
+     * swallowed, so that the reconciliation recording the event fails and goes through the usual
+     * error handling and retries. Off by default: Kubernetes treats events as best effort,
+     * supplemental data, and a controller that cannot reconcile because it cannot write an event is
+     * usually worse off than one that records nothing. Enable it only where an event that is not
+     * recorded must not go unnoticed.
+     *
+     * <p>Every failure propagates, transient ones included, such as a conflicting write when two
+     * replicas aggregate the same event at the same time; the reconciliation is then retried like
+     * any other failed one.
+     */
+    public Builder failOnError(boolean failOnError) {
+      this.failOnError = failOnError;
+      return this;
+    }
+
     public DefaultEventRecorder build() {
-      return new DefaultEventRecorder(sink, namingStrategy, keyStrategy, ownerReference);
+      return new DefaultEventRecorder(
+          sink, namingStrategy, keyStrategy, ownerReference, failOnError);
     }
   }
 
@@ -199,8 +226,12 @@ public void record(EventRecord event, Context<?> context) {
       assembled = toEvent(context, event);
       sink.emit(assembled, context);
     } catch (Exception e) {
-      // recording an event must never break the caller: a controller that fails to reconcile
-      // because it could not write an event is strictly worse than one that records nothing
+      if (failOnError) {
+        throw e;
+      }
+      // by default recording an event must not break the caller: a controller that fails to
+      // reconcile because it could not write an event is usually worse off than one that records
+      // nothing
       log.warn(
           "Could not record {} event named {} with reason {} for resource {} in namespace {}",
           event.type(),

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/DefaultEventSink.java

@@ -41,8 +41,7 @@
  * <p>Recording therefore needs {@code get}, {@code create} and {@code patch} on {@code events} in
  * the core API group. The patch is made on the resource version the event was read at, so two
  * writers counting the same occurrence at the same time conflict, and the one that loses leaves the
- * occurrence uncounted and the conflict logged by {@link DefaultEventRecorder}, recording being
- * best effort.
+ * occurrence uncounted and the conflict to {@link DefaultEventRecorder} to handle.
  */
 public class DefaultEventSink implements EventSink {
 

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/EventRecorder.java

@@ -34,8 +34,8 @@
  * io.javaoperatorsdk.operator.api.reconciler.Context#eventRecorder()}, which is already bound to
  * the context.
  *
- * <p>Recording an event is best effort: failures to write the event to the cluster are logged and
- * swallowed, and never fail the caller.
+ * <p>Whether a failure to record an event fails the caller is up to the implementation, see {@link
+ * DefaultEventRecorder} for the default.
  */
 @Experimental(API_MIGHT_CHANGE)
 public interface EventRecorder {

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/ResourceEventRecorder.java

@@ -22,9 +22,6 @@
 /**
  * An {@link EventRecorder} bound to a single object, typically the primary resource of the current
  * reconciliation.
- *
- * <p>Recording an event is best effort: failures to write the event to the cluster are logged and
- * swallowed, and never fail the caller.
  */
 @Experimental(API_MIGHT_CHANGE)
 public interface ResourceEventRecorder {

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/event/DefaultEventRecorderTest.java

@@ -35,6 +35,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -175,6 +176,36 @@ void aFailingSinkNeverFailsTheCaller() {
         .doesNotThrowAnyException();
   }
 
+  @Test
+  void aFailingSinkFailsTheCallerWhenTheRecorderIsBuiltToFailOnError() {
+    var failure = new RuntimeException("API server said no");
+    var failing =
+        DefaultEventRecorder.builder(
+                (event, context) -> {
+                  throw failure;
+                })
+            .failOnError(true)
+            .build();
+
+    assertThatThrownBy(
+            () -> failing.record(EventRecord.normal("Created", "created"), context(configMap())))
+        .isSameAs(failure);
+  }
+
+  @Test
+  void aFailingSinkNeverFailsTheCallerWhenTheRecorderIsBuiltWithDefaults() {
+    var failing =
+        DefaultEventRecorder.builder(
+                (event, context) -> {
+                  throw new RuntimeException("API server said no");
+                })
+            .build();
+
+    assertThatCode(
+            () -> failing.record(EventRecord.normal("Created", "created"), context(configMap())))
+        .doesNotThrowAnyException();
+  }
+
   @Test
   void passesTheContextOnToTheSink() {
     var contexts = new ArrayList<Context<?>>();