Fixes #31331: stop a single unevaluable event from discarding its whole change-event batch (#32953)

Author: manerow

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

@@ -72,6 +72,9 @@
  * after the entity is gone — the re-read then threw {@code EntityNotFoundException} and aborted the
  * whole subscription instead of simply not matching. The re-read also used {@code NON_DELETED},
  * which made every soft-delete event throw as well.
+ *
+ * <p>Also covers open-metadata/OpenMetadata#31331, the same batch loss reached through a different
+ * exception: an entity type whose schema does not declare the filtered field at all.
  */
 @Execution(ExecutionMode.CONCURRENT)
 @ExtendWith(TestNamespaceExtension.class)
@@ -161,6 +164,74 @@ void getFilteredEvents_deletedEntityInBatch_stillDeliversTheLiveEvent(TestNamesp
         "the hard-deleted entity no longer resolves a domain, so its event must not match");
   }
 
+  /**
+   * #31331: a batch is filtered as a unit and its offset is committed either way, so an event whose
+   * entity type cannot supply the filtered field must be dropped on its own. {@code domain.json}
+   * declares no {@code domains} property, so re-reading it raised {@code IllegalArgumentException}
+   * and took every other event in the batch down with it.
+   */
+  @Test
+  void getFilteredEvents_entityTypeWithoutDomains_stillDeliversTheMatchingEvent(TestNamespace ns) {
+    Domain domain = createDomain(ns);
+    Table table = createTable(ns, createDomainAssignment(domain), null);
+
+    ChangeEvent tableEvent =
+        updateEvent(Entity.TABLE, payloadWithoutRelationships(table)).withId(UUID.randomUUID());
+    ChangeEvent domainEvent = updateEvent(Entity.DOMAIN, domain).withId(UUID.randomUUID());
+
+    Map<ChangeEvent, Set<UUID>> batch = new LinkedHashMap<>();
+    batch.put(domainEvent, Set.of(UUID.randomUUID()));
+    batch.put(tableEvent, Set.of(UUID.randomUUID()));
+
+    Map<ChangeEvent, Set<UUID>> delivered =
+        AlertUtil.getFilteredEvents(
+            subscriptionOnAllResourcesFilteringOnDomain(domain.getFullyQualifiedName()),
+            batch,
+            null);
+
+    assertTrue(
+        delivered.containsKey(tableEvent),
+        "the matching event must survive a batch holding an entity type without domains");
+    assertFalse(
+        delivered.containsKey(domainEvent),
+        "a domain declares no domains of its own, so its event must not match");
+  }
+
+  /**
+   * The same evaluation backs {@code /diagnosticInfo}, which walks unprocessed events in a parallel
+   * stream with no isolation and answered 500 while such an event sat unprocessed.
+   */
+  @Test
+  void isChangeEventAllowed_entityTypeWithoutDomains_returnsFalseInsteadOfThrowing(
+      TestNamespace ns) {
+    Domain domain = createDomain(ns);
+    ChangeEvent domainEvent = updateEvent(Entity.DOMAIN, domain).withId(UUID.randomUUID());
+
+    assertFalse(
+        AlertUtil.isChangeEventAllowed(
+            domainEvent,
+            subscriptionOnAllResourcesFilteringOnDomain(domain.getFullyQualifiedName())
+                .getFilteringRules(),
+            null,
+            AlertUtil.LOG_EVALUATION_ERROR),
+        "a domain event must evaluate to false rather than throw out of the matcher");
+  }
+
+  private EventSubscription subscriptionOnAllResourcesFilteringOnDomain(String domainFqn) {
+    EventFilterRule rule =
+        new EventFilterRule()
+            .withName("matchAnyDomain")
+            .withEffect(ArgumentsInput.Effect.INCLUDE)
+            .withCondition("matchAnyDomain({'" + domainFqn + "'})");
+    return new EventSubscription()
+        .withName("alertAllResourcesDomainSubscription")
+        .withFilteringRules(
+            new FilteringRules()
+                .withResources(List.of("all"))
+                .withRules(List.of(rule))
+                .withActions(List.of()));
+  }
+
   private EventSubscription subscriptionFilteringOnDomain(String domainFqn) {
     EventFilterRule rule =
         new EventFilterRule()

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/changeEvent/AbstractEventConsumer.java

@@ -29,6 +29,7 @@
 import lombok.Getter;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.tuple.Pair;
 import org.openmetadata.schema.api.events.CreateEventSubscription;
 import org.openmetadata.schema.entity.events.AlertMetrics;
 import org.openmetadata.schema.entity.events.EventSubscription;
@@ -254,7 +255,7 @@ public void publishEvents(Map<ChangeEvent, Set<UUID>> events) {
       return;
     }
     Map<ChangeEvent, Set<UUID>> filteredEvents =
-        getFilteredEvents(eventSubscription, events, startingTimestamp);
+        getFilteredEvents(eventSubscription, events, startingTimestamp, this::deadLetterEvent);
     RecipientResolver resolver = new RecipientResolver();
     int successDeliveries = 0;
     int failedDeliveries = 0;
@@ -273,6 +274,20 @@ public void publishEvents(Map<ChangeEvent, Set<UUID>> events) {
     alertMetrics.withFailedEvents(alertMetrics.getFailedEvents() + failedDeliveries);
   }
 
+  /** An event we could not even filter is a publisher-side failure, so record it as one. */
+  private void deadLetterEvent(ChangeEvent event, Exception error) {
+    LOG.error(
+        "Event Subscription: {} could not evaluate filters for change event {}",
+        eventSubscription.getName(),
+        event.getId(),
+        error);
+    handleFailedEvent(
+        new EventPublisherException(
+            String.format("Failed to evaluate alert filters: %s", error.getMessage()),
+            Pair.of(eventSubscription.getId(), event)),
+        false);
+  }
+
   private EventDeliveryResult publishEvent(
       ChangeEvent event, Set<UUID> destinationIds, RecipientResolver resolver) {
     // Group destinations by type to enable cross-destination recipient deduplication

openmetadata-service/src/main/java/org/openmetadata/service/events/scheduled/EventSubscriptionScheduler.java

@@ -385,7 +385,8 @@ public long getRelevantUnprocessedEvents(UUID subscriptionId) {
         .map(
             eventJson -> {
               ChangeEvent event = JsonUtils.readValue(eventJson, ChangeEvent.class);
-              return AlertUtil.checkIfChangeEventIsAllowed(event, filteringRules, startingTimestamp)
+              return AlertUtil.isChangeEventAllowed(
+                      event, filteringRules, startingTimestamp, AlertUtil.LOG_EVALUATION_ERROR)
                   ? event
                   : null;
             })
@@ -489,7 +490,8 @@ public List<ChangeEvent> getRelevantUnprocessedEvents(
         .map(
             eventJson -> {
               ChangeEvent event = JsonUtils.readValue(eventJson, ChangeEvent.class);
-              return AlertUtil.checkIfChangeEventIsAllowed(event, filteringRules, startingTimestamp)
+              return AlertUtil.isChangeEventAllowed(
+                      event, filteringRules, startingTimestamp, AlertUtil.LOG_EVALUATION_ERROR)
                   ? event
                   : null;
             })

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

@@ -26,13 +26,15 @@
 import java.net.URI;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.OptionalLong;
 import java.util.Set;
 import java.util.UUID;
+import java.util.function.BiConsumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
@@ -68,6 +70,15 @@ public final class AlertUtil {
 
   private static final String FIELD_PIPELINE_STATUS = "pipelineStatus";
 
+  /** Default handler for {@link #isChangeEventAllowed}: log the event and leave it out. */
+  public static final BiConsumer<ChangeEvent, Exception> LOG_EVALUATION_ERROR =
+      (event, error) ->
+          LOG.error(
+              "Excluding change event {} on {}: alert filter evaluation failed",
+              event.getId(),
+              event.getEntityType(),
+              error);
+
   private AlertUtil() {}
 
   public static <T> void validateExpression(String condition, Class<T> clz) {
@@ -253,13 +264,60 @@ public static Map<ChangeEvent, Set<UUID>> getFilteredEvents(
       EventSubscription eventSubscription,
       Map<ChangeEvent, Set<UUID>> events,
       Long startingTimestamp) {
+    return getFilteredEvents(eventSubscription, events, startingTimestamp, LOG_EVALUATION_ERROR);
+  }
+
+  public static Map<ChangeEvent, Set<UUID>> getFilteredEvents(
+      EventSubscription eventSubscription,
+      Map<ChangeEvent, Set<UUID>> events,
+      Long startingTimestamp,
+      BiConsumer<ChangeEvent, Exception> onEvaluationError) {
     Long watermark = alertingWatermark(eventSubscription, startingTimestamp);
-    return events.entrySet().stream()
-        .filter(
-            entry ->
-                checkIfChangeEventIsAllowed(
-                    entry.getKey(), eventSubscription.getFilteringRules(), watermark))
-        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    FilteringRules filteringRules = eventSubscription.getFilteringRules();
+    Map<ChangeEvent, Set<UUID>> filteredEvents = new HashMap<>();
+    for (Map.Entry<ChangeEvent, Set<UUID>> entry : events.entrySet()) {
+      if (isChangeEventAllowed(entry.getKey(), filteringRules, watermark, onEvaluationError)) {
+        filteredEvents.put(entry.getKey(), entry.getValue());
+      }
+    }
+    return filteredEvents;
+  }
+
+  /**
+   * Evaluates one event in isolation, excluding it rather than letting the failure escape. Callers
+   * evaluate whole batches, and the consumer commits its offset either way, so an exception thrown
+   * out of a single matcher would silently drop every other event with it (issue #31331). The catch
+   * is deliberately cause-agnostic: matchers reach the store, the SpEL runtime and the event
+   * payload, and none of those failures may cost an unrelated event its notification.
+   */
+  public static boolean isChangeEventAllowed(
+      ChangeEvent event,
+      FilteringRules filteringRules,
+      Long startingTimestamp,
+      BiConsumer<ChangeEvent, Exception> onEvaluationError) {
+    boolean allowed;
+    try {
+      allowed = checkIfChangeEventIsAllowed(event, filteringRules, startingTimestamp);
+    } catch (Exception e) {
+      reportEvaluationError(onEvaluationError, event, e);
+      allowed = false;
+    }
+    return allowed;
+  }
+
+  /**
+   * Runs the failure handler without letting it become a second failure. The consumer's handler
+   * writes a dead-letter row, so a transient database error there would otherwise escape this
+   * method, abort the surrounding batch loop and lose the very events this isolation exists to
+   * protect.
+   */
+  private static void reportEvaluationError(
+      BiConsumer<ChangeEvent, Exception> onEvaluationError, ChangeEvent event, Exception error) {
+    try {
+      onEvaluationError.accept(event, error);
+    } catch (Exception handlerError) {
+      LOG.error("Failed to record unevaluable change event {}", event.getId(), handlerError);
+    }
   }
 
   /**

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

@@ -22,6 +22,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
 import org.openmetadata.schema.EntityInterface;
@@ -51,11 +52,17 @@
 import org.openmetadata.service.Entity;
 import org.openmetadata.service.exception.EntityNotFoundException;
 import org.openmetadata.service.formatter.util.FormatterUtil;
+import org.openmetadata.service.jdbi3.EntityRepository;
 import org.openmetadata.service.jdbi3.TaskRepository;
 import org.openmetadata.service.resources.feeds.MessageParser;
 import org.openmetadata.service.util.EntityUtil.RelationIncludes;
 import org.openmetadata.service.util.FullyQualifiedName;
 
+/**
+ * SpEL matchers for alert filtering rules. A matcher returns {@code false} when it cannot evaluate
+ * and must never throw for a well-formed event: it runs inside a change-event batch whose offset is
+ * committed either way, so an escaping exception silently discards every other event in that batch.
+ */
 @Slf4j
 public class AlertsRuleEvaluator {
   private static final String FIELD_TEST_SUITES_AND_OWNERS =
@@ -150,7 +157,8 @@ private boolean matchesEntityOrTestSuiteOwner(
 
   private List<EntityReference> resolveOwners(EntityInterface entity) {
     List<EntityReference> ownerReferences = entity.getOwners();
-    if (nullOrEmpty(ownerReferences)) {
+    if (nullOrEmpty(ownerReferences)
+        && supports(changeEvent.getEntityType(), EntityRepository::isSupportsOwners)) {
       EntityInterface storedEntity = readStoredEntity(entity.getId(), Entity.FIELD_OWNERS);
       ownerReferences = storedEntity == null ? ownerReferences : storedEntity.getOwners();
     }
@@ -496,9 +504,11 @@ public boolean matchAnyDomain(List<String> fieldChangeUpdate) {
   }
 
   private boolean matchesEntityOrTestSuiteDomain(EntityInterface entity, List<String> domainFqns) {
-    EntityInterface storedEntity = readStoredEntity(entity.getId(), Entity.FIELD_DOMAINS);
-    List<EntityReference> domains =
-        storedEntity == null ? entity.getDomains() : storedEntity.getDomains();
+    List<EntityReference> domains = entity.getDomains();
+    if (supports(changeEvent.getEntityType(), EntityRepository::isSupportsDomains)) {
+      EntityInterface storedEntity = readStoredEntity(entity.getId(), Entity.FIELD_DOMAINS);
+      domains = storedEntity == null ? domains : storedEntity.getDomains();
+    }
     boolean matched = matchesAnyDomainFqn(domains, domainFqns);
     if (!matched && TEST_CASE.equals(changeEvent.getEntityType())) {
       // If we did not match on the domain and are dealing with a test case,
@@ -531,6 +541,24 @@ private <T extends EntityInterface> T readStoredEntity(UUID entityId, String fie
         changeEvent.getEntityType(), entityId, fields, DELETED_TOLERANT_SUBJECT);
   }
 
+  /**
+   * Whether {@code entityType}'s repository declares the capability, so a matcher can skip the store
+   * re-read for a field the entity's schema does not declare. Reading it anyway raises
+   * {@code IllegalArgumentException} out of the matcher and discards the whole change-event batch
+   * (issue #31331). An unregistered type reads as unsupported: a feed subject can name one.
+   */
+  private static boolean supports(String entityType, Predicate<EntityRepository<?>> capability) {
+    boolean supported = false;
+    if (entityType != null) {
+      try {
+        supported = capability.test(Entity.getEntityRepository(entityType));
+      } catch (EntityNotFoundException e) {
+        LOG.debug("No repository for {}, treating the field as unsupported", entityType);
+      }
+    }
+    return supported;
+  }
+
   private List<TestSuite> resolveTestSuites(TestCase testCase, String fields) {
     TestCase storedTestCase = readStoredEntity(testCase.getId(), fields);
     List<TestSuite> testSuites =
@@ -770,18 +798,28 @@ private static UUID parseUuidOrNull(String id) {
 
   private boolean threadSubjectMatchesOwner(List<String> ownerNameList) {
     EntityInterface subject =
-        Entity.getEntityOrNull(threadSubject(), Entity.FIELD_OWNERS, Include.NON_DELETED);
+        readFeedSubject(Entity.FIELD_OWNERS, EntityRepository::isSupportsOwners);
     return subject != null
         && !nullOrEmpty(subject.getOwners())
         && matchOwners(subject.getOwners(), ownerNameList);
   }
 
   private boolean threadSubjectMatchesDomain(List<String> domainFqns) {
     EntityInterface subject =
-        Entity.getEntityOrNull(threadSubject(), Entity.FIELD_DOMAINS, Include.NON_DELETED);
+        readFeedSubject(Entity.FIELD_DOMAINS, EntityRepository::isSupportsDomains);
     return subject != null && matchesAnyDomainFqn(subject.getDomains(), domainFqns);
   }
 
+  /** The feed's subject read with {@code field}, or null when its type cannot supply that field. */
+  private EntityInterface readFeedSubject(String field, Predicate<EntityRepository<?>> capability) {
+    EntityReference subject = threadSubject();
+    EntityInterface entity = null;
+    if (subject != null && supports(subject.getType(), capability)) {
+      entity = Entity.getEntityOrNull(subject, field, Include.NON_DELETED);
+    }
+    return entity;
+  }
+
   private boolean matchOwners(List<EntityReference> ownerReferences, List<String> ownerNameList) {
     Set<String> ownerNames =
         ownerNameList.stream().map(EntityInterfaceUtil::unquoteName).collect(Collectors.toSet());