fix(alerts): dedup successful change events to prevent Postgres ON CONFLICT abort (#28827)

* fix(alerts): dedup successful change events to prevent Postgres ON CONFLICT abort

* refactor(alerts): publishEvent returns delivery result; add consumer-level recipient-dedup tests

Author: manerow

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

@@ -0,0 +1,86 @@
+/*
+ *  Copyright 2025 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.it.tests;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.List;
+import java.util.UUID;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.openmetadata.service.Entity;
+import org.openmetadata.service.jdbi3.CollectionDAO.EventSubscriptionDAO;
+
+/**
+ * Regression test for {@code EventSubscriptionDAO.batchUpsertSuccessfulChangeEvents}.
+ *
+ * <p>When an alert fans an event out to multiple destination types, the same
+ * {@code (change_event_id, event_subscription_id)} pair was added to the batch more than once. With
+ * the Postgres JDBC driver's {@code reWriteBatchedInserts=true}, the batch collapses into a single
+ * multi-row {@code INSERT ... ON CONFLICT (...) DO UPDATE}, and Postgres aborts the whole statement
+ * with "ON CONFLICT DO UPDATE command cannot affect row a second time" — so no success records were
+ * persisted. This guard must run under a Postgres profile to exercise that path; on MySQL it still
+ * verifies the dedup invariant.
+ */
+@Execution(ExecutionMode.CONCURRENT)
+public class EventSubscriptionBatchUpsertIT {
+
+  private static EventSubscriptionDAO dao() {
+    return Entity.getCollectionDAO().eventSubscriptionDAO();
+  }
+
+  private static String json(String changeEventId) {
+    return "{\"id\":\"" + changeEventId + "\"}";
+  }
+
+  @Test
+  void duplicateKeysInBatchDoNotFailAndCollapseToOneRow() {
+    String subscriptionId = UUID.randomUUID().toString();
+    String changeEventId = UUID.randomUUID().toString();
+    long timestamp = System.currentTimeMillis();
+    try {
+      assertDoesNotThrow(
+          () ->
+              dao()
+                  .batchUpsertSuccessfulChangeEvents(
+                      List.of(changeEventId, changeEventId),
+                      List.of(subscriptionId, subscriptionId),
+                      List.of(json(changeEventId), json(changeEventId)),
+                      List.of(timestamp, timestamp)));
+      assertEquals(1L, dao().getSuccessfulRecordCount(subscriptionId));
+    } finally {
+      dao().deleteSuccessfulChangeEventBySubscriptionId(subscriptionId);
+    }
+  }
+
+  @Test
+  void mixedBatchKeepsOneRowPerDistinctKey() {
+    String subscriptionId = UUID.randomUUID().toString();
+    String changeEventA = UUID.randomUUID().toString();
+    String changeEventB = UUID.randomUUID().toString();
+    long timestamp = System.currentTimeMillis();
+    try {
+      dao()
+          .batchUpsertSuccessfulChangeEvents(
+              List.of(changeEventA, changeEventB, changeEventA),
+              List.of(subscriptionId, subscriptionId, subscriptionId),
+              List.of(json(changeEventA), json(changeEventB), json(changeEventA)),
+              List.of(timestamp, timestamp, timestamp));
+      assertEquals(2L, dao().getSuccessfulRecordCount(subscriptionId));
+    } finally {
+      dao().deleteSuccessfulChangeEventBySubscriptionId(subscriptionId);
+    }
+  }
+}

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

@@ -170,16 +170,6 @@ public void handleFailedEvent(EventPublisherException ex, boolean errorOnSub) {
             source.toString());
   }
 
-  private void recordSuccessfulChangeEvent(UUID eventSubscriptionId, ChangeEvent event) {
-    Entity.getCollectionDAO()
-        .eventSubscriptionDAO()
-        .upsertSuccessfulChangeEvent(
-            event.getId().toString(),
-            eventSubscriptionId.toString(),
-            JsonUtils.pojoToJson(event),
-            System.currentTimeMillis());
-  }
-
   private EventSubscriptionOffset loadInitialOffset(JobExecutionContext context) {
     Object offsetValue = jobDetail.getJobDataMap().get(ALERT_OFFSET_KEY);
     if (offsetValue != null) {
@@ -232,54 +222,68 @@ public void publishEvents(Map<ChangeEvent, Set<UUID>> events) {
     if (events.isEmpty()) {
       return;
     }
-
-    // Filter events based on subscription configuration (entity type, conditions, etc.)
     Map<ChangeEvent, Set<UUID>> filteredEvents = getFilteredEvents(eventSubscription, events);
     RecipientResolver resolver = new RecipientResolver();
+    int successDeliveries = 0;
+    int failedDeliveries = 0;
+    for (Map.Entry<ChangeEvent, Set<UUID>> eventWithReceivers : filteredEvents.entrySet()) {
+      EventDeliveryResult result =
+          publishEvent(eventWithReceivers.getKey(), eventWithReceivers.getValue(), resolver);
+      // Record once per (event, subscription): the table has no destination dimension, so
+      // recording per type would duplicate rows and break Postgres ON CONFLICT.
+      if (result.delivered()) {
+        successfulEvents.add(eventWithReceivers.getKey());
+      }
+      successDeliveries += result.successCount();
+      failedDeliveries += result.failedCount();
+    }
+    alertMetrics.withSuccessEvents(alertMetrics.getSuccessEvents() + successDeliveries);
+    alertMetrics.withFailedEvents(alertMetrics.getFailedEvents() + failedDeliveries);
+  }
 
-    for (var eventWithReceivers : filteredEvents.entrySet()) {
-      ChangeEvent event = eventWithReceivers.getKey();
-      Set<UUID> destinationIds = eventWithReceivers.getValue();
-
-      // Group destinations by type to enable cross-destination recipient deduplication
-      Map<SubscriptionType, List<Destination<ChangeEvent>>> destinationsByType =
-          groupDestinationsByType(destinationIds);
-
-      for (var entry : destinationsByType.entrySet()) {
-        List<Destination<ChangeEvent>> destinations = entry.getValue();
-        Destination<ChangeEvent> publisher = destinations.getFirst();
-
-        // Resolve recipients from all destinations of this type for deduplication
-        Set<Recipient> recipients = Set.of();
-        if (publisher.requiresRecipients()) {
-          List<SubscriptionDestination> subDestinations =
-              destinations.stream().map(Destination::getSubscriptionDestination).toList();
-          recipients = resolver.resolveRecipients(event, subDestinations);
-        }
-
-        // Send via primary destination only, with deduplicated recipients (one send per type)
-        boolean status = true;
-        if (!publisher.requiresRecipients() || !recipients.isEmpty()) {
-          try {
-            publisher.sendMessage(event, recipients);
-          } catch (EventPublisherException e) {
-            LOG.error("Failed to send alert: {}", e.getMessage());
-            handleFailedEvent(e, true);
-            status = false;
-          }
-        }
+  private EventDeliveryResult publishEvent(
+      ChangeEvent event, Set<UUID> destinationIds, RecipientResolver resolver) {
+    // Group destinations by type to enable cross-destination recipient deduplication
+    Map<SubscriptionType, List<Destination<ChangeEvent>>> destinationsByType =
+        groupDestinationsByType(destinationIds);
+    int successCount = 0;
+    int failedCount = 0;
+    for (Map.Entry<SubscriptionType, List<Destination<ChangeEvent>>> entry :
+        destinationsByType.entrySet()) {
+      if (sendToDestinationType(event, entry.getValue(), resolver)) {
+        successCount++;
+      } else {
+        failedCount++;
+      }
+    }
+    return new EventDeliveryResult(successCount > 0, successCount, failedCount);
+  }
 
-        if (status) {
-          // Collect successful events instead of writing immediately
-          // Batch write happens in commit() to reduce connection pool contention
-          // Note: Empty recipients is treated as successful (no-op send)
-          successfulEvents.add(eventWithReceivers.getKey());
-          alertMetrics.withSuccessEvents(alertMetrics.getSuccessEvents() + 1);
-        } else {
-          alertMetrics.withFailedEvents(alertMetrics.getFailedEvents() + 1);
-        }
+  private record EventDeliveryResult(boolean delivered, int successCount, int failedCount) {}
+
+  private boolean sendToDestinationType(
+      ChangeEvent event, List<Destination<ChangeEvent>> destinations, RecipientResolver resolver) {
+    Destination<ChangeEvent> publisher = destinations.getFirst();
+    // Resolve recipients from all destinations of this type for deduplication
+    Set<Recipient> recipients = Set.of();
+    if (publisher.requiresRecipients()) {
+      List<SubscriptionDestination> subDestinations =
+          destinations.stream().map(Destination::getSubscriptionDestination).toList();
+      recipients = resolver.resolveRecipients(event, subDestinations);
+    }
+    // Send via primary destination only, with deduplicated recipients (one send per type).
+    // Empty recipients is treated as successful (no-op send).
+    boolean status = true;
+    if (!publisher.requiresRecipients() || !recipients.isEmpty()) {
+      try {
+        publisher.sendMessage(event, recipients);
+      } catch (EventPublisherException e) {
+        LOG.error("Failed to send alert: {}", e.getMessage());
+        handleFailedEvent(e, true);
+        status = false;
       }
     }
+    return status;
   }
 
   private Map<SubscriptionType, List<Destination<ChangeEvent>>> groupDestinationsByType(

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/CollectionDAO.java

@@ -5333,27 +5333,47 @@ void upsertFailedEvent(
         @Bind("json") String json,
         @Bind("source") String source);
 
-    @ConnectionAwareSqlUpdate(
-        value =
-            "INSERT INTO successful_sent_change_events (change_event_id, event_subscription_id, json, timestamp) "
-                + "VALUES (:change_event_id, :event_subscription_id, :json, :timestamp) "
-                + "ON DUPLICATE KEY UPDATE json = :json, timestamp = :timestamp",
-        connectionType = MYSQL)
-    @ConnectionAwareSqlUpdate(
-        value =
-            "INSERT INTO successful_sent_change_events (change_event_id, event_subscription_id, json, timestamp) "
-                + "VALUES (:change_event_id, :event_subscription_id, CAST(:json AS jsonb), :timestamp) "
-                + "ON CONFLICT (change_event_id, event_subscription_id) "
-                + "DO UPDATE SET json = EXCLUDED.json, timestamp = EXCLUDED.timestamp",
-        connectionType = POSTGRES)
-    void upsertSuccessfulChangeEvent(
-        @Bind("change_event_id") String changeEventId,
-        @Bind("event_subscription_id") String eventSubscriptionId,
-        @Bind("json") String json,
-        @Bind("timestamp") long timestamp);
-
     // Batch insert for successful events - reduces connection pool contention
-    // from N connections to 1 when processing multiple events
+    // from N connections to 1 when processing multiple events.
+    // Deduplicates by (change_event_id, event_subscription_id) first: the primary key has no
+    // destination dimension, and Postgres rejects a rewritten multi-row INSERT whose own rows
+    // conflict on the arbiter key ("ON CONFLICT DO UPDATE command cannot affect row a second
+    // time").
+    default void batchUpsertSuccessfulChangeEvents(
+        List<String> changeEventIds,
+        List<String> eventSubscriptionIds,
+        List<String> jsonList,
+        List<Long> timestamps) {
+      List<Integer> keep = distinctLastIndexes(changeEventIds, eventSubscriptionIds);
+      if (keep.size() == changeEventIds.size()) {
+        batchUpsertSuccessfulChangeEventsInternal(
+            changeEventIds, eventSubscriptionIds, jsonList, timestamps);
+      } else {
+        batchUpsertSuccessfulChangeEventsInternal(
+            pickByIndex(changeEventIds, keep),
+            pickByIndex(eventSubscriptionIds, keep),
+            pickByIndex(jsonList, keep),
+            pickByIndex(timestamps, keep));
+      }
+    }
+
+    static List<Integer> distinctLastIndexes(
+        List<String> changeEventIds, List<String> eventSubscriptionIds) {
+      LinkedHashMap<String, Integer> lastIndexByKey = new LinkedHashMap<>();
+      for (int i = 0; i < changeEventIds.size(); i++) {
+        lastIndexByKey.put(changeEventIds.get(i) + "|" + eventSubscriptionIds.get(i), i);
+      }
+      return new ArrayList<>(lastIndexByKey.values());
+    }
+
+    static <T> List<T> pickByIndex(List<T> source, List<Integer> indexes) {
+      List<T> result = new ArrayList<>(indexes.size());
+      for (int index : indexes) {
+        result.add(source.get(index));
+      }
+      return result;
+    }
+
     @Transaction
     @ConnectionAwareSqlBatch(
         value =
@@ -5368,7 +5388,7 @@ void upsertSuccessfulChangeEvent(
                 + "ON CONFLICT (change_event_id, event_subscription_id) "
                 + "DO UPDATE SET json = EXCLUDED.json, timestamp = EXCLUDED.timestamp",
         connectionType = POSTGRES)
-    void batchUpsertSuccessfulChangeEvents(
+    void batchUpsertSuccessfulChangeEventsInternal(
         @Bind("change_event_id") List<String> changeEventIds,
         @Bind("event_subscription_id") List<String> eventSubscriptionIds,
         @Bind("json") List<String> jsonList,