@@ -48,10 +48,11 @@
  * can be overridden, see {@link
  * io.javaoperatorsdk.operator.api.config.ConfigurationService#clusterScopedEventNamespace()}.
  *
- * <p>Events are named deterministically, after the object they are about plus a hash of everything
- * that identifies the event, so that recording the same event again resolves to the event already
- * recorded for it. Repeat occurrences are then counted on that event rather than recorded as copies
- * of it, see {@link DefaultEventSink}.
+ * <p>By default, events are named deterministically, after the object they are about plus a hash of
+ * everything that identifies the event, so that recording the same event again resolves to the
+ * 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)}.
  */
 public class DefaultEventRecorder implements EventRecorder {
 
@@ -74,16 +75,21 @@ public class DefaultEventRecorder implements EventRecorder {
   private static final int IDENTITY_HASH_LENGTH = 32;
 
   private final EventSink sink;
+  private final EventNamingStrategy namingStrategy;
   private final EventKeyStrategy keyStrategy;
   private final boolean ownerReference;
 
   public DefaultEventRecorder(EventSink sink) {
-    this(sink, EventKeyStrategy.none(), false);
+    this(sink, EventNamingStrategy.none(), EventKeyStrategy.none(), false);
   }
 
   private DefaultEventRecorder(
-      EventSink sink, EventKeyStrategy keyStrategy, boolean ownerReference) {
+      EventSink sink,
+      EventNamingStrategy namingStrategy,
+      EventKeyStrategy keyStrategy,
+      boolean ownerReference) {
     this.sink = sink;
+    this.namingStrategy = namingStrategy;
     this.keyStrategy = keyStrategy;
     this.ownerReference = ownerReference;
   }
@@ -96,16 +102,25 @@ public static Builder builder(EventSink sink) {
   public static final class Builder {
 
     private final EventSink sink;
+    private EventNamingStrategy namingStrategy = EventNamingStrategy.none();
     private EventKeyStrategy keyStrategy = EventKeyStrategy.none();
     private boolean ownerReference = false;
 
     private Builder(EventSink sink) {
       this.sink = Objects.requireNonNull(sink, "sink must not be null");
     }
 
+    /** The strategy naming recorded events, see {@link EventNamingStrategy}. */
+    public Builder namingStrategy(EventNamingStrategy namingStrategy) {
+      this.namingStrategy =
+          Objects.requireNonNull(namingStrategy, "namingStrategy must not be null");
+      return this;
+    }
+
     /**
      * The strategy deriving the default aggregation key of records that do not set one, see {@link
-     * EventKeyStrategy}.
+     * EventKeyStrategy}. The key is ignored for events whose name is set by the record or resolved
+     * by the naming strategy, see {@link EventNamingStrategy}.
      */
     public Builder keyStrategy(EventKeyStrategy keyStrategy) {
       this.keyStrategy = Objects.requireNonNull(keyStrategy, "keyStrategy must not be null");
@@ -126,7 +141,7 @@ public Builder ownerReference(boolean ownerReference) {
     }
 
     public DefaultEventRecorder build() {
-      return new DefaultEventRecorder(sink, keyStrategy, ownerReference);
+      return new DefaultEventRecorder(sink, namingStrategy, keyStrategy, ownerReference);
     }
   }
 
@@ -162,14 +177,17 @@ private static String resolve() {
   public void record(EventRecord event, Context<?> context) {
     Objects.requireNonNull(context, "the context of the reconciliation must not be null");
     Objects.requireNonNull(event, "event must not be null");
+    Event assembled = null;
     try {
-      sink.emit(toEvent(context, event), 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
       log.warn(
-          "Could not record {} event with reason {} for resource {} in namespace {}",
+          "Could not record {} event named {} with reason {} for resource {} in namespace {}",
           event.type(),
+          assembled != null ? assembled.getMetadata().getName() : "unknown",
           event.reason(),
           context.getPrimaryResource().getMetadata().getName(),
           context.getPrimaryResource().getMetadata().getNamespace(),
@@ -249,6 +267,30 @@ private String eventNamespace(HasMetadata regarding, Context<?> context) {
         CLUSTER_SCOPED_EVENT_NAMESPACE);
   }
 
+  private String eventName(HasMetadata regarding, EventRecord record, String reportingController) {
+    return record
+        .name()
+        .filter(name -> !name.isBlank())
+        .or(() -> namingStrategy.nameFor(regarding, record).filter(name -> !name.isBlank()))
+        .map(DefaultEventRecorder::truncateToMaxNameLength)
+        .orElseGet(() -> identityHashName(regarding, record, reportingController));
+  }
+
+  private static String truncateToMaxNameLength(String name) {
+    if (name.length() <= MAX_NAME_LENGTH) {
+      return name;
+    }
+    var truncated = name.substring(0, MAX_NAME_LENGTH);
+    while (truncated.endsWith("-") || truncated.endsWith(".")) {
+      truncated = truncated.substring(0, truncated.length() - 1);
+    }
+    log.warn(
+        "Truncated the name of event {} to {} to stay within the Kubernetes name limit",
+        name,
+        truncated);
+    return truncated;
+  }
+
   /**
    * Names events {@code <object name>.<hash>}, following the convention of the Go client, hashing
    * everything that makes two events the same event: the object, the type, the reason, the
@@ -259,8 +301,12 @@ private String eventNamespace(HasMetadata regarding, Context<?> context) {
    *
    * <p>The object is identified by its uid, with the kind as a fallback for objects that do not
    * have one yet, such as a dependent resource that has only been built so far.
+   *
+   * <p>This is the fallback when the record does not set a name and the naming strategy resolves to
+   * nothing, see {@link EventNamingStrategy}.
    */
-  private String eventName(HasMetadata regarding, EventRecord record, String reportingController) {
+  private String identityHashName(
+      HasMetadata regarding, EventRecord record, String reportingController) {
     var metadata = regarding.getMetadata();
     var identity =
         String.join(
 
@@ -0,0 +1,52 @@
+/*
+ * Copyright Java Operator SDK Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.javaoperatorsdk.operator.api.event;
+
+import java.util.Optional;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.reconciler.Experimental;
+
+import static io.javaoperatorsdk.operator.api.reconciler.Experimental.API_MIGHT_CHANGE;
+
+/**
+ * Names the event recorded about an object. The name is what the sink looks a recorded event up by,
+ * so it is also the aggregation identity: two records resolving to the same name are counted as
+ * occurrences of one event. A name must therefore be unique among the events it should not
+ * aggregate with, and stable across operator restarts and replicas.
+ *
+ * <p>A name must be a valid RFC 1123 DNS subdomain: at most 253 lowercase alphanumeric characters,
+ * {@code -} or {@code .}, starting and ending with an alphanumeric character. The API server
+ * rejects other names, and since recording is best effort, an event with an invalid name is
+ * dropped. Names longer than the limit are truncated. A name derived from the object and a fixed
+ * lowercase suffix (such as {@code <object>-status-report}) satisfies all of this by construction.
+ *
+ * <p>An empty result or a blank name falls back to the default {@code <object>.<identity hash>}
+ * name.
+ *
+ * <p>Implementations are called from concurrent reconciliations and must be thread safe.
+ */
+@Experimental(API_MIGHT_CHANGE)
+@FunctionalInterface
+public interface EventNamingStrategy {
+
+  Optional<String> nameFor(HasMetadata regarding, EventRecord record);
+
+  /** No custom naming: every event gets the default {@code <object>.<identity hash>} name. */
+  static EventNamingStrategy none() {
+    return (regarding, record) -> Optional.empty();
+  }
+}
@@ -32,6 +32,7 @@ public final class EventRecord {
   private final EventType type;
   private final String reason;
   private final String message;
+  private final String name;
   private final String key;
   private final Boolean ownedByRegarding;
   private final String action;
@@ -43,6 +44,7 @@ private EventRecord(Builder builder) {
     this.type = builder.type;
     this.reason = builder.reason;
     this.message = builder.message;
+    this.name = builder.name;
     this.key = builder.key;
     this.action = builder.action;
     this.reportingComponent = builder.reportingComponent;
@@ -77,9 +79,21 @@ public String message() {
     return message;
   }
 
+  /**
+   * The name of the recorded event, overriding the recorder's naming. The name is the aggregation
+   * identity and must be a valid RFC 1123 DNS subdomain, see {@link EventNamingStrategy}. A blank
+   * name is treated as unset.
+   */
+  public Optional<String> name() {
+    return Optional.ofNullable(name);
+  }
+
   /**
    * Identifies this event among the events about the same object, so that repeated occurrences
    * resolve to the same event rather than to one event each.
+   *
+   * <p>The key is ignored when the record sets a {@link #name()} or the recorder's naming strategy
+   * resolves one: the name is then the aggregation identity on its own.
    */
   public Optional<String> key() {
     return Optional.ofNullable(key);
@@ -129,6 +143,7 @@ public static final class Builder {
     private EventType type = EventType.NORMAL;
     private String reason;
     private String message;
+    private String name;
     private String key;
     private Boolean ownedByRegarding;
     private String action;
@@ -153,6 +168,12 @@ public Builder message(String message) {
       return this;
     }
 
+    /** Sets the name of the recorded event, see {@link EventRecord#name()}. */
+    public Builder name(String name) {
+      this.name = name;
+      return this;
+    }
+
     /** Sets the key identifying this event, see {@link EventRecord#key()}. */
     public Builder key(String key) {
       this.key = key;
 
@@ -382,6 +382,105 @@ void setsNoOwnerReferenceWhenTheRegardingObjectHasNoUidYet() {
     assertThat(emitted.get(0).getMetadata().getOwnerReferences()).isEmpty();
   }
 
+  @Test
+  void namesTheEventAfterThePerRecordNameWhenOneIsSet() {
+    recorder.record(
+        EventRecord.builder().reason("Created").message("created").name("my-event-name").build(),
+        context(configMap()));
+
+    assertThat(emitted.get(0).getMetadata().getName()).isEqualTo("my-event-name");
+  }
+
+  @Test
+  void namesEventsThroughTheNamingStrategy() {
+    var recorder =
+        DefaultEventRecorder.builder((event, context) -> emitted.add(event))
+            .namingStrategy(
+                (regarding, record) ->
+                    Optional.of(regarding.getMetadata().getName() + "-" + record.reason()))
+            .build();
+
+    recorder.record(EventRecord.warning("failed", "first"), context(configMap()));
+    recorder.record(EventRecord.warning("failed", "second"), context(configMap()));
+
+    assertThat(emitted)
+        .allSatisfy(event -> assertThat(event.getMetadata().getName()).isEqualTo("test1-failed"));
+  }
+
+  @Test
+  void fallsBackToTheIdentityHashNameWhenTheNamingStrategyResolvesNothing() {
+    var recorder =
+        DefaultEventRecorder.builder((event, context) -> emitted.add(event))
+            .namingStrategy((regarding, record) -> Optional.empty())
+            .build();
+
+    recorder.record(EventRecord.warning("Failed", "message"), context(configMap()));
+
+    assertThat(emitted.get(0).getMetadata().getName())
+        .startsWith("test1.")
+        .hasSize("test1.".length() + 32);
+  }
+
+  @Test
+  void truncatesSuppliedNamesToTheKubernetesNameLengthLimit() {
+    recorder.record(
+        EventRecord.builder().reason("Created").message("created").name("a".repeat(300)).build(),
+        context(configMap()));
+
+    assertThat(emitted.get(0).getMetadata().getName()).hasSize(253);
+  }
+
+  @Test
+  void stripsTrailingSeparatorsFromTruncatedNames() {
+    recorder.record(
+        EventRecord.builder()
+            .reason("Created")
+            .message("created")
+            .name("a".repeat(252) + "." + "b".repeat(47))
+            .build(),
+        context(configMap()));
+    recorder.record(
+        EventRecord.builder()
+            .reason("Created")
+            .message("created")
+            .name("b".repeat(252) + "-" + "c".repeat(47))
+            .build(),
+        context(configMap()));
+
+    assertThat(emitted.get(0).getMetadata().getName()).isEqualTo("a".repeat(252));
+    assertThat(emitted.get(1).getMetadata().getName()).isEqualTo("b".repeat(252));
+  }
+
+  @Test
+  void treatsABlankNameAsUnset() {
+    var recorder =
+        DefaultEventRecorder.builder((event, context) -> emitted.add(event))
+            .namingStrategy((regarding, record) -> Optional.of(" "))
+            .build();
+
+    recorder.record(
+        EventRecord.builder().reason("Created").message("created").name("").build(),
+        context(configMap()));
+
+    assertThat(emitted.get(0).getMetadata().getName()).matches("test1\\.[0-9a-f]{32}");
+  }
+
+  @Test
+  void aFailingNamingStrategyNeverFailsTheCaller() {
+    var recorder =
+        DefaultEventRecorder.builder((event, context) -> emitted.add(event))
+            .namingStrategy(
+                (regarding, record) -> {
+                  throw new RuntimeException("cannot derive a name");
+                })
+            .build();
+
+    assertThatCode(
+            () -> recorder.record(EventRecord.normal("Created", "created"), context(configMap())))
+        .doesNotThrowAnyException();
+    assertThat(emitted).isEmpty();
+  }
+
   Context<?> context(HasMetadata primaryResource) {
     return context(primaryResource, DefaultEventRecorder.CLUSTER_SCOPED_EVENT_NAMESPACE);
   }