fix: replace parallelStream with stream in backfill mae && add logging (#333)

* use stream && add logging

* update log

Author: Jinxin Ma

restli-resources/src/main/java/com/linkedin/metadata/restli/BaseEntityAgnosticResource.java

@@ -83,12 +83,13 @@ public Task<BackfillItem[]> backfillMAE(@Nonnull BackfillItem[] backfillRequests
           log.warn("LocalDAO not found for entity type: " + entityType);
           continue;
         }
-        final List<BackfillItem> items = entityTypeToRequestsMap.get(entityType);
-        backfillResults.addAll(items.parallelStream()
-            // immutable dao, should be thread-safe
+        final List<BackfillItem> itemsToBackfill = entityTypeToRequestsMap.get(entityType);
+        final List<BackfillItem> backfilledItems = itemsToBackfill.stream()
             .map(item -> backfillMAEForUrn(item.getUrn(), item.getAspects(), backfillMode, dao.get()).orElse(null))
             .filter(Objects::nonNull)
-            .collect(Collectors.toList()));
+            .collect(Collectors.toList());
+        log.info(String.format("Given requests: %s, backfill results: %s", itemsToBackfill, backfilledItems));
+        backfillResults.addAll(backfilledItems);
       }
       return backfillResults.toArray(new BackfillItem[0]); // insert order is not guaranteed the same as input
     });
@@ -97,9 +98,11 @@ public Task<BackfillItem[]> backfillMAE(@Nonnull BackfillItem[] backfillRequests
   protected Optional<BackfillItem> backfillMAEForUrn(@Nonnull String urn, @Nonnull List<String> aspectSet,
       @Nonnull BackfillMode backfillMode, @Nonnull BaseLocalDAO<? extends UnionTemplate, ? extends Urn> dao) {
     try {
+      log.info(String.format("Attempt to backfill MAE for urn: %s, aspectSet: %s, backfillMode: %s", urn, aspectSet, backfillMode));
       // set aspectSetToUse to null if empty to backfill all aspects
       Set<String> aspectSetToUse = aspectSet.isEmpty() ? null : new HashSet<>(aspectSet);
       Set<String> backfilledAspects = dao.backfillMAE(backfillMode, aspectSetToUse, Collections.singleton(urn)).get(urn);
+      log.info(String.format("Backfilled aspects: %s, for urn: %s, aspectSet: %s, backfillMode: %s", backfilledAspects, urn, aspectSet, backfillMode));
       if (backfilledAspects == null || backfilledAspects.isEmpty()) {
         return Optional.empty();
       }

restli-resources/src/test/java/com/linkedin/metadata/restli/BaseEntityAgnosticResourceTest.java

@@ -1,5 +1,6 @@
 package com.linkedin.metadata.restli;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.linkedin.data.template.StringArray;
@@ -15,6 +16,7 @@
 import com.linkedin.testing.urn.BarUrn;
 import com.linkedin.testing.urn.FooUrn;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
@@ -144,6 +146,26 @@ public void testBackfillMAEEmptyBackfillResult() {
     verify(_fooLocalDAO, times(3)).backfillMAE(any(), any(), any());
   }
 
+  @Test
+  public void testBackfillMAEFilterEmptyAspectUrn() {
+    TestResource testResource = new TestResource();
+    Set<String> urnSet = ImmutableSet.of(makeFooUrn(1).toString(), makeFooUrn(2).toString());
+    when(_fooLocalDAO.backfillMAE(BackfillMode.BACKFILL_INCLUDING_LIVE_INDEX, null, Collections.singleton(makeFooUrn(1).toString())))
+        .thenReturn(ImmutableMap.of(makeFooUrn(1).toString(), multiAspectsSet));
+    BackfillItem[] result = runAndWait(testResource.backfillMAE(provideBackfillItems(urnSet, null), IngestionMode.BACKFILL));
+    assertEqualBackfillItemArrays(result, provideBackfillItems(ImmutableSet.of(makeFooUrn(1).toString()), multiAspectsSet));
+  }
+
+  @Test
+  public void testBackfillMAEDuplicateUrn() {
+    TestResource testResource = new TestResource();
+    List<String> urnList = ImmutableList.of(makeFooUrn(1).toString(), makeFooUrn(1).toString());
+    when(_fooLocalDAO.backfillMAE(BackfillMode.BACKFILL_INCLUDING_LIVE_INDEX, null, Collections.singleton(makeFooUrn(1).toString())))
+        .thenReturn(ImmutableMap.of(makeFooUrn(1).toString(), multiAspectsSet));
+    BackfillItem[] result = runAndWait(testResource.backfillMAE(provideBackfillItems(urnList, null), IngestionMode.BACKFILL));
+    assertEqualBackfillItemArrays(result, provideBackfillItems(ImmutableList.of(makeFooUrn(1).toString(), makeFooUrn(1).toString()), multiAspectsSet));
+  }
+
   @Test
   public void testBackfillMAENoSuchEntity() {
     TestResource testResource = new TestResource();
@@ -186,7 +208,7 @@ public void testBackfillMAEException() {
     assertEqualBackfillItemArrays(result, expectedItems);
   }
 
-  private BackfillItem[] provideBackfillItems(Set<String> urnSet, Set<String> aspects) {
+  private BackfillItem[] provideBackfillItems(Collection<String> urnSet, Set<String> aspects) {
     return urnSet.stream().map(urn -> {
       BackfillItem item = new BackfillItem();
       item.setUrn(urn);