updates per review

Author: david-leifker

metadata-io/src/main/java/com/linkedin/metadata/service/UpdateIndicesService.java

@@ -30,6 +30,7 @@
 import com.linkedin.mxe.MetadataChangeLog;
 import com.linkedin.mxe.SystemMetadata;
 import com.linkedin.structured.StructuredPropertyDefinition;
+import com.linkedin.util.Pair;
 import io.datahubproject.metadata.context.OperationContext;
 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
@@ -138,12 +139,15 @@ public void handleChangeEvent(
           // system metadata is last for tracing
           handleUpdateChangeEvent(opContext, mclItem, true);
         } else if (hookEvent.getChangeType() == ChangeType.DELETE) {
+          Pair<EntitySpec, AspectSpec> specPair = extractSpecPair(mclItem);
+          boolean isDeletingKey = isDeletingKey(specPair);
+
           // non-system metadata
-          handleDeleteChangeEvent(opContext, mclItem, false);
+          handleNonSystemMetadataDeleteChangeEvent(opContext, specPair, mclItem, isDeletingKey);
           // graph update
           updateGraphIndicesService.handleChangeEvent(opContext, event);
           // system metadata is last for tracing
-          handleDeleteChangeEvent(opContext, mclItem, true);
+          handleSystemMetadataDeleteChangeEvent(mclItem.getUrn(), specPair, isDeletingKey);
         }
       }
     } catch (IOException e) {
@@ -242,6 +246,25 @@ public void updateIndexMappings(
     }
   }
 
+  private static Pair<EntitySpec, AspectSpec> extractSpecPair(@Nonnull final MCLItem event) {
+    final EntitySpec entitySpec = event.getEntitySpec();
+    final Urn urn = event.getUrn();
+
+    AspectSpec aspectSpec = entitySpec.getAspectSpec(event.getAspectName());
+    if (aspectSpec == null) {
+      throw new RuntimeException(
+          String.format(
+              "Failed to retrieve Aspect Spec for entity with name %s, aspect with name %s. Cannot update indices for MCL.",
+              urn.getEntityType(), event.getAspectName()));
+    }
+
+    return Pair.of(entitySpec, aspectSpec);
+  }
+
+  private static boolean isDeletingKey(Pair<EntitySpec, AspectSpec> specPair) {
+    return specPair.getSecond().getName().equals(specPair.getFirst().getKeyAspectName());
+  }
+
   /**
    * This very important method processes {@link MetadataChangeLog} deletion events to cleanup the
    * Metadata Graph when an aspect or entity is removed.
@@ -252,33 +275,39 @@ public void updateIndexMappings(
    * <p>Note that if an entity's key aspect is deleted, the entire entity will be purged from
    * search, graph, timeseries, etc.
    *
+   * @param opContext operation's context
+   * @param specPair entity & aspect spec
    * @param event the change event to be processed.
+   * @param isDeletingKey whether the key aspect is being deleted
    */
-  private void handleDeleteChangeEvent(
+  private void handleNonSystemMetadataDeleteChangeEvent(
       @Nonnull OperationContext opContext,
+      Pair<EntitySpec, AspectSpec> specPair,
       @Nonnull final MCLItem event,
-      boolean forSystemMetadata) {
-
-    final EntitySpec entitySpec = event.getEntitySpec();
-    final Urn urn = event.getUrn();
-
-    AspectSpec aspectSpec = entitySpec.getAspectSpec(event.getAspectName());
-    if (aspectSpec == null) {
-      throw new RuntimeException(
-          String.format(
-              "Failed to retrieve Aspect Spec for entity with name %s, aspect with name %s. Cannot update indices for MCL.",
-              urn.getEntityType(), event.getAspectName()));
+      boolean isDeletingKey) {
+
+    if (!specPair.getSecond().isTimeseries()) {
+      deleteSearchData(
+          opContext,
+          event.getUrn(),
+          specPair.getFirst().getName(),
+          specPair.getSecond(),
+          event.getRecordTemplate(),
+          isDeletingKey);
     }
+  }
 
-    RecordTemplate aspect = event.getRecordTemplate();
-    Boolean isDeletingKey = event.getAspectName().equals(entitySpec.getKeyAspectName());
-
-    if (!aspectSpec.isTimeseries()) {
-      if (!forSystemMetadata) {
-        deleteSearchData(opContext, urn, entitySpec.getName(), aspectSpec, aspect, isDeletingKey);
-      } else {
-        deleteSystemMetadata(urn, aspectSpec, isDeletingKey);
-      }
+  /**
+   * Handle the system metadata separately for tracing
+   *
+   * @param urn delete urn
+   * @param specPair entity & aspect spec
+   * @param isDeletingKey whether the key aspect is being deleted
+   */
+  private void handleSystemMetadataDeleteChangeEvent(
+      @Nonnull Urn urn, Pair<EntitySpec, AspectSpec> specPair, boolean isDeletingKey) {
+    if (!specPair.getSecond().isTimeseries()) {
+      deleteSystemMetadata(urn, specPair.getSecond(), isDeletingKey);
     }
   }
 

metadata-io/src/main/java/com/linkedin/metadata/trace/KafkaTraceReader.java

@@ -136,7 +136,7 @@ public Map<Urn, Map<String, TraceStorageStatus>> tracePendingStatuses(
 
     try {
       List<Map.Entry<Urn, Map<String, TraceStorageStatus>>> results =
-          CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
+          CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))
               .thenApply(
                   v -> futures.stream().map(CompletableFuture::join).collect(Collectors.toList()))
               .get(timeoutSeconds, TimeUnit.SECONDS);