@@ -24,6 +24,7 @@
 import java.time.temporal.ChronoUnit;
 import java.util.HexFormat;
 import java.util.Objects;
+import java.util.regex.Pattern;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -48,10 +49,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 {
 
@@ -73,17 +75,26 @@ public class DefaultEventRecorder implements EventRecorder {
 
   private static final int IDENTITY_HASH_LENGTH = 32;
 
+  /** What the API server accepts as an object name, see RFC 1123 on DNS subdomains. */
+  private static final Pattern RFC_1123_SUBDOMAIN =
+      Pattern.compile("[a-z0-9]([-a-z0-9]*[a-z0-9])?(\\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*");
+
   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 +107,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 +146,7 @@ public Builder ownerReference(boolean ownerReference) {
     }
 
     public DefaultEventRecorder build() {
-      return new DefaultEventRecorder(sink, keyStrategy, ownerReference);
+      return new DefaultEventRecorder(sink, namingStrategy, keyStrategy, ownerReference);
     }
   }
 
@@ -162,14 +182,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 +272,40 @@ 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)
+        .filter(DefaultEventRecorder::isValidEventName)
+        .orElseGet(() -> identityHashName(regarding, record, reportingController));
+  }
+
+  private static boolean isValidEventName(String name) {
+    if (RFC_1123_SUBDOMAIN.matcher(name).matches()) {
+      return true;
+    }
+    log.warn(
+        "Falling back to the default event name: {} is not a valid RFC 1123 DNS subdomain", name);
+    return false;
+  }
+
+  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 +316,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,51 @@
+/*
+ * 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. 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 an invalid name falls back to the default {@code <object>.<identity hash>}
+ * name, rather than the event being lost to the API server rejecting the 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}; an
+   * invalid name falls back to the default name. 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;