fix(alerts): declare the event types each resource can deliver, reject and stop offering the rest

openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/EventSubscriptionResourceIT.java

@@ -7,11 +7,17 @@
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.stream.StreamSupport;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.parallel.Execution;
 import org.junit.jupiter.api.parallel.ExecutionMode;
@@ -286,6 +292,77 @@ void test_eventSubscriptionWithResources_200_OK(TestNamespace ns) {
     assertNotNull(subscription.getFilteringRules());
   }
 
+  @Test
+  void test_unreachableEventType_400(TestNamespace ns) {
+    CreateEventSubscription request =
+        eventTypeSubscription(ns, "sub_bad_event", "glossaryTerm", "suggestionCreated");
+
+    Exception exception = assertThrows(Exception.class, () -> createEntity(request));
+    assertTrue(
+        exception.getMessage().contains("suggestionCreated"),
+        "the error must name the offending event type, was: " + exception.getMessage());
+  }
+
+  @Test
+  void test_reachableEventType_200_OK(TestNamespace ns) {
+    EventSubscription subscription =
+        createEntity(
+            eventTypeSubscription(ns, "sub_thread_event", "glossaryTerm", "threadCreated"));
+    assertNotNull(subscription.getFilteringRules());
+  }
+
+  @Test
+  void test_notificationResourcesServeSupportedEventTypes() throws Exception {
+    HttpRequest request =
+        HttpRequest.newBuilder()
+            .uri(
+                URI.create(
+                    SdkClients.getServerUrl() + "/v1/events/subscriptions/notification/resources"))
+            .header("Authorization", "Bearer " + SdkClients.getAdminToken())
+            .GET()
+            .build();
+    HttpResponse<String> response =
+        HttpClient.newHttpClient().send(request, HttpResponse.BodyHandlers.ofString());
+    assertEquals(200, response.statusCode());
+
+    JsonNode glossaryTerm =
+        StreamSupport.stream(
+                new ObjectMapper().readTree(response.body()).get("data").spliterator(), false)
+            .filter(descriptor -> "glossaryTerm".equals(descriptor.get("name").asText()))
+            .findFirst()
+            .orElseThrow();
+    List<String> eventTypes =
+        StreamSupport.stream(glossaryTerm.get("supportedEventTypes").spliterator(), false)
+            .map(JsonNode::asText)
+            .toList();
+    assertTrue(eventTypes.contains("threadCreated"));
+    assertTrue(eventTypes.contains("entityCreated"));
+    assertFalse(eventTypes.contains("suggestionCreated"));
+  }
+
+  private CreateEventSubscription eventTypeSubscription(
+      TestNamespace ns, String name, String resource, String... eventTypes) {
+    return new CreateEventSubscription()
+        .withName(ns.prefix(name))
+        .withDescription("Subscription filtering on event types")
+        .withAlertType(CreateEventSubscription.AlertType.NOTIFICATION)
+        .withResources(List.of(resource))
+        .withEnabled(false)
+        .withInput(
+            new AlertFilteringInput()
+                .withFilters(
+                    List.of(
+                        new ArgumentsInput()
+                            .withName("filterByEventType")
+                            .withEffect(ArgumentsInput.Effect.INCLUDE)
+                            .withArguments(
+                                List.of(
+                                    new Argument()
+                                        .withName("eventTypeList")
+                                        .withInput(List.of(eventTypes)))))))
+        .withDestinations(getWebhookDestination(ns));
+  }
+
   @Test
   void test_slackDestination_200_OK(TestNamespace ns) {
     OpenMetadataClient client = SdkClients.adminClient();

openmetadata-service/src/main/java/org/openmetadata/service/events/subscription/AlertUtil.java

@@ -47,6 +47,8 @@
 import org.openmetadata.schema.entity.events.TestDestinationStatus;
 import org.openmetadata.schema.entity.feed.Thread;
 import org.openmetadata.schema.type.ChangeEvent;
+import org.openmetadata.schema.type.EventType;
+import org.openmetadata.schema.type.FilterResourceDescriptor;
 import org.openmetadata.schema.utils.JsonUtils;
 import org.openmetadata.service.Entity;
 import org.openmetadata.service.exception.CatalogExceptionMessage;
@@ -60,6 +62,9 @@ public final class AlertUtil {
   private static final Cache<String, Expression> COMPILED_CONDITIONS =
       Caffeine.newBuilder().maximumSize(1000).build();
 
+  private static final String FILTER_BY_EVENT_TYPE = "filterByEventType";
+  private static final String EVENT_TYPE_LIST_ARGUMENT = "eventTypeList";
+
   private AlertUtil() {}
 
   public static <T> void validateExpression(String condition, Class<T> clz) {
@@ -290,10 +295,11 @@ public static FilteringRules validateAndBuildFilteringConditions(
     }
 
     if (alertType.equals(CreateEventSubscription.AlertType.NOTIFICATION)) {
+      FilterResourceDescriptor descriptor =
+          EventsSubscriptionRegistry.getEntityNotificationDescriptor(resource.get(0));
       Map<String, EventFilterRule> supportedFilters =
-          buildFilteringRulesMap(
-              EventsSubscriptionRegistry.getEntityNotificationDescriptor(resource.get(0))
-                  .getSupportedFilters());
+          buildFilteringRulesMap(descriptor.getSupportedFilters());
+      validateEventTypes(descriptor, input);
       // Input validation
       if (input != null) {
         return new FilteringRules()
@@ -328,6 +334,47 @@ public static FilteringRules validateAndBuildFilteringConditions(
         .withActions(Collections.emptyList());
   }
 
+  private static void validateEventTypes(
+      FilterResourceDescriptor descriptor, AlertFilteringInput input) {
+    if (input == null) {
+      return;
+    }
+    for (ArgumentsInput filter : listOrEmpty(input.getFilters())) {
+      // Excluding an event type the resource cannot produce is a no-op, not a silent no-op alert.
+      if (FILTER_BY_EVENT_TYPE.equals(filter.getName())
+          && filter.getEffect() != ArgumentsInput.Effect.EXCLUDE) {
+        eventTypeValues(filter).forEach(value -> validateEventType(descriptor, value));
+      }
+    }
+  }
+
+  private static List<String> eventTypeValues(ArgumentsInput filter) {
+    return listOrEmpty(filter.getArguments()).stream()
+        .filter(argument -> EVENT_TYPE_LIST_ARGUMENT.equals(argument.getName()))
+        .map(Argument::getInput)
+        .flatMap(input -> listOrEmpty(input).stream())
+        .toList();
+  }
+
+  private static void validateEventType(FilterResourceDescriptor descriptor, String value) {
+    EventType eventType = parse(value);
+    if (eventType == null
+        || !listOrEmpty(descriptor.getSupportedEventTypes()).contains(eventType)) {
+      throw new BadRequestException(
+          CatalogExceptionMessage.eventTypeNotSupportedByResource(descriptor.getName(), value));
+    }
+  }
+
+  private static EventType parse(String value) {
+    EventType eventType = null;
+    try {
+      eventType = EventType.fromValue(value);
+    } catch (IllegalArgumentException e) {
+      LOG.debug("Unknown event type '{}' in an alert filter", value);
+    }
+    return eventType;
+  }
+
   private static Map<String, EventFilterRule> buildFilteringRulesMap(
       List<EventFilterRule> filteringRules) {
     return filteringRules.stream()

openmetadata-service/src/main/java/org/openmetadata/service/events/subscription/ResourceEventTypes.java

@@ -0,0 +1,112 @@
+/*
+ *  Copyright 2021 Collate
+ *  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 org.openmetadata.service.events.subscription;
+
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import org.openmetadata.schema.type.EventType;
+import org.openmetadata.service.Entity;
+
+/**
+ * Populates {@code FilterResourceDescriptor.supportedEventTypes}: the event types an alert on a
+ * given notification resource can actually receive, derived from what the emitters produce rather
+ * than from the {@link EventType} enum. Consumers read the descriptor, never this class.
+ *
+ * <p>An alert matches an event when {@code AlertUtil.shouldTriggerAlert} lets it through: the "all"
+ * resource takes everything; a THREAD event matches {@code thread.type} for the thread-type
+ * resources and {@code thread.entityRef.type} for every other resource; anything else matches on
+ * {@code entityType}. Values no emitter produces, or that never reach {@code change_event}, are
+ * deliberately absent — see {@code UNREACHABLE}.
+ */
+public final class ResourceEventTypes {
+  public static final String ALL_RESOURCE = "all";
+  private static final String CONVERSATION = "conversation";
+
+  /** Emitted for every entity by the generic CRUD paths in EntityRepository. */
+  private static final List<EventType> ENTITY_EVENTS =
+      List.of(
+          EventType.ENTITY_CREATED,
+          EventType.ENTITY_UPDATED,
+          EventType.ENTITY_SOFT_DELETED,
+          EventType.ENTITY_DELETED,
+          EventType.ENTITY_RESTORED);
+
+  /** Emitted with entityType=THREAD; reaches an entity resource via the thread's parent entity. */
+  private static final List<EventType> THREAD_EVENTS =
+      List.of(
+          EventType.THREAD_CREATED,
+          EventType.THREAD_UPDATED,
+          EventType.POST_CREATED,
+          EventType.POST_UPDATED);
+
+  /** Only usage reporting emits entityFieldsChanged, and only for these entities. */
+  private static final Set<String> USAGE_RESOURCES =
+      Set.of(Entity.TABLE, Entity.DASHBOARD, Entity.PIPELINE, Entity.CHART, Entity.MLMODEL);
+
+  /** Reachable only through "all": their entity types are not notification resources. */
+  private static final List<EventType> ALL_ONLY_EVENTS =
+      List.of(
+          EventType.LOGICAL_TEST_CASE_ADDED,
+          EventType.ENTITY_LINEAGE_ADDED,
+          EventType.ENTITY_LINEAGE_UPDATED,
+          EventType.ENTITY_LINEAGE_DELETED);
+
+  /** Legacy thread-tasks still emit these; retired with the Recognizer migration (#30559). */
+  private static final List<EventType> LEGACY_TASK_EVENTS =
+      List.of(EventType.TASK_RESOLVED, EventType.TASK_CLOSED);
+
+  /**
+   * Values no resource can advertise: {@code ENTITY_NO_CHANGE} is a sentinel that ChangeEventHandler
+   * never inserts, {@code USER_LOGIN}/{@code USER_LOGOUT} are written to the audit log only, and the
+   * rest lost their emitter in the task redesign (#29039).
+   */
+  public static final Set<EventType> UNREACHABLE =
+      Set.of(
+          EventType.ENTITY_NO_CHANGE,
+          EventType.TASK_CREATED,
+          EventType.TASK_UPDATED,
+          EventType.SUGGESTION_CREATED,
+          EventType.SUGGESTION_UPDATED,
+          EventType.SUGGESTION_ACCEPTED,
+          EventType.SUGGESTION_REJECTED,
+          EventType.SUGGESTION_DELETED,
+          EventType.USER_LOGIN,
+          EventType.USER_LOGOUT);
+
+  private ResourceEventTypes() {}
+
+  public static List<EventType> forResource(String resource) {
+    Set<EventType> eventTypes = new LinkedHashSet<>();
+    if (ALL_RESOURCE.equalsIgnoreCase(resource)) {
+      eventTypes.addAll(ENTITY_EVENTS);
+      eventTypes.add(EventType.ENTITY_FIELDS_CHANGED);
+      eventTypes.addAll(THREAD_EVENTS);
+      eventTypes.addAll(LEGACY_TASK_EVENTS);
+      eventTypes.addAll(ALL_ONLY_EVENTS);
+    } else if (CONVERSATION.equalsIgnoreCase(resource)) {
+      eventTypes.addAll(THREAD_EVENTS);
+    } else {
+      eventTypes.addAll(ENTITY_EVENTS);
+      eventTypes.addAll(THREAD_EVENTS);
+      if (USAGE_RESOURCES.contains(resource)) {
+        eventTypes.add(EventType.ENTITY_FIELDS_CHANGED);
+      }
+      if (Entity.TASK.equalsIgnoreCase(resource)) {
+        eventTypes.addAll(LEGACY_TASK_EVENTS);
+      }
+    }
+    return List.copyOf(eventTypes);
+  }
+}

openmetadata-service/src/main/java/org/openmetadata/service/exception/CatalogExceptionMessage.java

@@ -153,6 +153,12 @@ public static String resourceTypeNotFound(String resourceType) {
     return String.format("Resource type %s not found", resourceType);
   }
 
+  public static String eventTypeNotSupportedByResource(String resourceType, String eventType) {
+    return String.format(
+        "Resource %s does not produce event type %s, so an alert filtering on it can never fire",
+        resourceType, eventType);
+  }
+
   public static String entityTypeNotSupported(String entityType) {
     return String.format("Entity type %s not supported", entityType);
   }

openmetadata-service/src/main/java/org/openmetadata/service/resources/events/subscription/EventSubscriptionResource.java

@@ -84,6 +84,7 @@
 import org.openmetadata.service.events.errors.EventPublisherException;
 import org.openmetadata.service.events.scheduled.EventSubscriptionScheduler;
 import org.openmetadata.service.events.subscription.EventsSubscriptionRegistry;
+import org.openmetadata.service.events.subscription.ResourceEventTypes;
 import org.openmetadata.service.exception.EntityNotFoundException;
 import org.openmetadata.service.jdbi3.CollectionDAO;
 import org.openmetadata.service.jdbi3.EventSubscriptionRepository;
@@ -1586,7 +1587,9 @@ public static List<FilterResourceDescriptor> getNotificationsFilterDescriptors()
                   return new FilterResourceDescriptor()
                       .withName(descriptor.getName())
                       .withSupportedFilters(rules)
-                      .withContainerEntities(descriptor.getContainerEntities());
+                      .withContainerEntities(descriptor.getContainerEntities())
+                      .withSupportedEventTypes(
+                          ResourceEventTypes.forResource(descriptor.getName()));
                 })
             .toList();
     setAllResourceContainerEntities(descriptors);