EPRMPP-83651 || Clean storage job out of memory

src/main/java/com/epam/reportportal/jobs/clean/CleanStorageJob.java

@@ -4,11 +4,13 @@
 import com.epam.reportportal.model.BlobNotFoundException;
 import com.epam.reportportal.storage.DataStorageService;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.Base64;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
 import net.javacrumbs.shedlock.spring.annotation.SchedulerLock;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.logging.log4j.util.Strings;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.scheduling.annotation.Scheduled;
@@ -27,9 +29,13 @@ public class CleanStorageJob extends BaseJob {
   private static final String SELECT_AND_DELETE_DATA_CHUNK_QUERY =
       "DELETE FROM attachment_deletion WHERE id IN "
           + "(SELECT id FROM attachment_deletion ORDER BY id LIMIT ?) RETURNING *";
+
+  private static final int MAX_BATCH_SIZE = 200000;
   private final DataStorageService storageService;
   private final int chunkSize;
 
+  private final int batchSize;
+
   /**
    * Initializes {@link CleanStorageJob}.
    *
@@ -42,6 +48,7 @@ public CleanStorageJob(JdbcTemplate jdbcTemplate, DataStorageService storageServ
     super(jdbcTemplate);
     this.chunkSize = chunkSize;
     this.storageService = storageService;
+    this.batchSize = chunkSize <= MAX_BATCH_SIZE ? chunkSize : MAX_BATCH_SIZE;
   }
 
   /**
@@ -54,31 +61,44 @@ public void execute() {
     logStart();
     AtomicInteger counter = new AtomicInteger(0);
 
-    jdbcTemplate.query(SELECT_AND_DELETE_DATA_CHUNK_QUERY, rs -> {
+    int batchNumber = 1;
+    while (batchNumber * batchSize <= chunkSize) {
+      List<String> attachments = new ArrayList<>();
+      List<String> thumbnails = new ArrayList<>();
+      jdbcTemplate.query(SELECT_AND_DELETE_DATA_CHUNK_QUERY, rs -> {
+        do {
+          String attachment = rs.getString("file_id");
+          String thumbnail = rs.getString("thumbnail_id");
+          if (attachment != null) {
+            attachments.add(attachment);
+          }
+          if (thumbnail != null) {
+            thumbnails.add(thumbnail);
+          }
+        } while (rs.next());
+      }, batchSize);
+
+      int attachmentsSize = thumbnails.size() + attachments.size();
+      if (attachmentsSize == 0){
+        break;
+      }
       try {
-        delete(rs.getString("file_id"), rs.getString("thumbnail_id"));
-        counter.incrementAndGet();
-        while (rs.next()) {
-          delete(rs.getString("file_id"), rs.getString("thumbnail_id"));
-          counter.incrementAndGet();
-        }
+        storageService.deleteAll(
+            thumbnails.stream().map(this::decode).collect(Collectors.toList()));
+        storageService.deleteAll(
+            attachments.stream().map(this::decode).collect(Collectors.toList()));
       } catch (BlobNotFoundException e) {
-        LOGGER.info("File {} is not found when executing clean storage job", e.getFileName());
+        LOGGER.info("File is not found when executing clean storage job");
       } catch (Exception e) {
         throw new RuntimeException(ROLLBACK_ERROR_MESSAGE, e);
       }
-    }, chunkSize);
-
-    logFinish(counter.get());
-  }
 
-  private void delete(String fileId, String thumbnailId) throws Exception {
-    if (Strings.isNotBlank(fileId)) {
-      storageService.delete(decode(fileId));
-    }
-    if (Strings.isNotBlank(thumbnailId)) {
-      storageService.delete(decode(thumbnailId));
+      counter.addAndGet(attachmentsSize);
+      LOGGER.info("Iteration {}, deleted {} attachments", batchNumber, attachmentsSize);
+      batchNumber++;
     }
+
+    logFinish(counter.get());
   }
 
   private String decode(String data) {

src/main/java/com/epam/reportportal/storage/DataStorageService.java

@@ -16,9 +16,11 @@
 
 package com.epam.reportportal.storage;
 
+import java.util.List;
+
 /**
  * Storage service interface
  */
 public interface DataStorageService {
-    void delete(String filePath) throws Exception;
+    void deleteAll(List<String> paths) throws Exception;
 }

src/main/java/com/epam/reportportal/storage/LocalDataStorageService.java

@@ -16,6 +16,7 @@
 
 package com.epam.reportportal.storage;
 
+import java.util.List;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -28,21 +29,23 @@
  */
 public class LocalDataStorageService implements DataStorageService {
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(LocalDataStorageService.class);
+  private static final Logger LOGGER = LoggerFactory.getLogger(LocalDataStorageService.class);
 
-    private final String storageRootPath;
+  private final String storageRootPath;
 
-    public LocalDataStorageService(String storageRootPath) {
-        this.storageRootPath = storageRootPath;
-    }
+  public LocalDataStorageService(String storageRootPath) {
+    this.storageRootPath = storageRootPath;
+  }
 
-    @Override
-    public void delete(String filePath) throws IOException {
-        try {
-            Files.deleteIfExists(Paths.get(storageRootPath, filePath));
-        } catch (IOException e) {
-            LOGGER.error("Unable to delete file '{}'", filePath, e);
-            throw e;
-        }
+  @Override
+  public void deleteAll(List<String> paths) throws IOException {
+    for (String path : paths) {
+      try {
+        Files.deleteIfExists(Paths.get(storageRootPath, path));
+      } catch (IOException e) {
+        LOGGER.error("Unable to delete file '{}'", path, e);
+        throw e;
+      }
     }
+  }
 }

src/main/java/com/epam/reportportal/storage/S3DataStorageService.java

@@ -16,14 +16,18 @@
 
 package com.epam.reportportal.storage;
 
-import com.epam.reportportal.model.BlobNotFoundException;
 import com.epam.reportportal.utils.FeatureFlag;
 import com.epam.reportportal.utils.FeatureFlagHandler;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import org.jclouds.blobstore.BlobStore;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.util.CollectionUtils;
 
 /**
  * S3 storage service.
@@ -55,35 +59,42 @@ public S3DataStorageService(BlobStore blobStore, String bucketPrefix, String def
   }
 
   @Override
-  public void delete(String filePath) throws Exception {
-    Path targetPath = Paths.get(filePath);
-    int nameCount = targetPath.getNameCount();
-
-    String bucket;
-    String objectName;
-
+  public void deleteAll(List<String> paths) throws Exception {
+    if (CollectionUtils.isEmpty(paths)) {
+      return;
+    }
     if (featureFlagHandler.isEnabled(FeatureFlag.SINGLE_BUCKET)) {
-      bucket = defaultBucketName;
-      objectName = filePath;
+      removeFiles(defaultBucketName, paths);
     } else {
-      if (nameCount > 1) {
-        bucket = bucketPrefix + retrievePath(targetPath, 0, 1);
-        objectName = retrievePath(targetPath, 1, nameCount);
-      } else {
-        bucket = defaultBucketName;
-        objectName = retrievePath(targetPath, 0, 1);
+      Map<String, List<String>> bucketPathMap = new HashMap<>();
+      for (String path : paths) {
+        Path targetPath = Paths.get(path);
+        int nameCount = targetPath.getNameCount();
+        String bucket = retrievePath(targetPath, 0, 1);
+        String cutPath = retrievePath(targetPath, 1, nameCount);
+        if (bucketPathMap.containsKey(bucket)) {
+          bucketPathMap.get(bucket).add(cutPath);
+        } else {
+          List<String> bucketPaths = new ArrayList<>();
+          bucketPaths.add(cutPath);
+          bucketPathMap.put(bucket, bucketPaths);
+        }
+      }
+      for (Map.Entry<String, List<String>> bucketPaths : bucketPathMap.entrySet()) {
+        removeFiles(bucketPrefix + bucketPaths.getKey(), bucketPaths.getValue());
       }
-    }
-
-    try {
-      blobStore.removeBlob(bucket, objectName);
-    } catch (Exception e) {
-      LOGGER.error("Unable to delete file '{}'", filePath, e);
-      throw new BlobNotFoundException(e);
     }
   }
 
   private String retrievePath(Path path, int beginIndex, int endIndex) {
     return String.valueOf(path.subpath(beginIndex, endIndex));
   }
+
+  private void removeFiles(String bucketName, List<String> paths) {
+    try {
+      blobStore.removeBlobs(bucketName, paths);
+    } catch (Exception e) {
+      LOGGER.warn("Exception {} is occurred during deleting file", e.getMessage());
+    }
+  }
 }

src/main/resources/application.yml

@@ -1,4 +1,6 @@
 rp:
+  feature:
+    flags:
   environment:
     variable:
       elements-counter:
@@ -8,6 +10,7 @@ rp:
           ## 30 seconds
           cron: '*/30 * * * * *'
           chunkSize: 1000
+          batchSize: 100
         attachment:
           ## 2 minutes
           cron: '0 */2 * * * *'