fix comparison, reduce duplicate logging

Author: david-leifker

entity-registry/src/main/java/com/linkedin/metadata/aspect/batch/AspectsBatch.java

@@ -159,12 +159,17 @@ static Stream<MCLItem> applyMCLSideEffects(
         .flatMap(mclSideEffect -> mclSideEffect.apply(items, retrieverContext));
   }
 
-  default boolean containsDuplicateAspects() {
+  default Map<String, Set<? extends BatchItem>> duplicateAspects() {
     return getItems().stream()
-            .map(i -> String.format("%s_%s", i.getClass().getName(), i.hashCode()))
-            .distinct()
-            .count()
-        != getItems().size();
+        .collect(
+            Collectors.groupingBy(
+                i -> String.format("%s_%s", i.getClass().getName(), i.hashCode()),
+                Collectors.collectingAndThen(
+                    Collectors.toSet(), set -> set.size() > 1 ? set : null)))
+        .entrySet()
+        .stream()
+        .filter(entry -> entry.getValue() != null)
+        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
   }
 
   default Map<String, Set<String>> getUrnAspectsMap() {

metadata-io/src/main/java/com/linkedin/metadata/entity/EntityServiceImpl.java

@@ -853,8 +853,9 @@ private List<UpdateAspectResult> ingestAspectsToLocalDB(
       @Nonnull final AspectsBatch inputBatch,
       boolean overwrite) {
 
-    if (inputBatch.containsDuplicateAspects()) {
-      log.warn(String.format("Batch contains duplicates: %s", inputBatch));
+    Map<String, Set<? extends BatchItem>> duplicates = inputBatch.duplicateAspects();
+    if (!duplicates.isEmpty()) {
+      log.warn(String.format("Batch contains duplicates: %s", duplicates.values()));
       MetricUtils.counter(EntityServiceImpl.class, "batch_with_duplicate").inc();
     }
 
@@ -2550,7 +2551,8 @@ private UpdateAspectResult ingestAspectToLocalDB(
       } catch (CloneNotSupportedException e) {
         throw new RuntimeException(e);
       }
-      latestSystemMetadata.setLastObserved(writeItem.getSystemMetadata().getLastObserved());
+      latestSystemMetadata.setLastObserved(
+          writeItem.getSystemMetadata().getLastObserved(), SetMode.IGNORE_NULL);
       latestSystemMetadata.setLastRunId(
           writeItem.getSystemMetadata().getLastRunId(GetMode.NULL), SetMode.IGNORE_NULL);