[NO ISSUE][STO] Limit local storage clean up to node partitions

- user model changes: no
- storage format changes: no
- interface changes: no

Details:

- When attempting to delete invalid indexes during global
  recovery, only delete invalid indexes of partitions where
  the node is the current master since the master of each
  partition will ask the replica to delete the invalid indexes.

Change-Id: I9b337e83ee138a1806bc0e38e2fc3e6870c65373
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/15503
Integration-Tests: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Reviewed-by: Murtadha Hubail <mhubail@apache.org>
Reviewed-by: Ali Alsuliman <ali.al.solaiman@gmail.com>

asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/StorageCleanupRequestMessage.java

@@ -21,6 +21,7 @@
 import static org.apache.hyracks.util.ExitUtil.EC_NC_FAILED_TO_NOTIFY_TASKS_COMPLETED;
 
 import java.util.Map;
+import java.util.Set;
 
 import org.apache.asterix.common.api.IDatasetLifecycleManager;
 import org.apache.asterix.common.api.INcApplicationContext;
@@ -57,10 +58,12 @@ public void handle(INcApplicationContext appContext) throws HyracksDataException
         PersistentLocalResourceRepository localResourceRepository =
                 (PersistentLocalResourceRepository) appContext.getLocalResourceRepository();
         Map<Long, LocalResource> localResources = localResourceRepository.loadAndGetAllResources();
+        Set<Integer> nodePartitions = appContext.getReplicaManager().getPartitions();
         for (LocalResource resource : localResources.values()) {
             DatasetLocalResource lr = (DatasetLocalResource) resource.getResource();
-            if (MetadataIndexImmutableProperties.isMetadataDataset(lr.getDatasetId())) {
-                // skip metadata indexes
+            if (!nodePartitions.contains(lr.getPartition())
+                    || MetadataIndexImmutableProperties.isMetadataDataset(lr.getDatasetId())) {
+                // skip replica partitions and metadata indexes
                 continue;
             }
             if (!validDatasetIds.contains(lr.getDatasetId())) {