Refactored tile range deletion to be fully streamed.

Author: axl8713

geowebcache/azureblob/src/main/java/org/geowebcache/azure/AzureBlobStore.java

@@ -15,18 +15,19 @@
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static java.util.Objects.isNull;
+import static org.geowebcache.azure.DeleteManager.PAGE_SIZE;
 
 import com.azure.core.util.BinaryData;
 import com.azure.storage.blob.models.BlobDownloadContentResponse;
 import com.azure.storage.blob.models.BlobItem;
 import com.azure.storage.blob.models.BlobProperties;
 import com.azure.storage.blob.models.BlobStorageException;
 import com.azure.storage.blob.specialized.BlockBlobClient;
-import com.google.common.collect.Iterators;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.UncheckedIOException;
 import java.time.OffsetDateTime;
+import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -198,26 +199,18 @@ public boolean delete(TileRange tileRange) throws StorageException {
         if (listeners.isEmpty()) {
             // if there are no listeners, don't bother requesting every tile
             // metadata to notify the listeners
-            List<String> keysToDelete = blobsToDelete.map(BlobItem::getName).collect(Collectors.toList());
-
-            // split the iteration in parts to avoid memory accumulation
-            Iterator<List<String>> partition = Iterators.partition(keysToDelete.iterator(), DeleteManager.PAGE_SIZE);
-
-            while (partition.hasNext() && !shutDown) {
-                deleteManager.deleteParallel(partition.next());
+            if (!shutDown) {
+                deleteManager.deleteStreamed(blobsToDelete);
             }
-
         } else {
             // if we need to gather info, we'll end up just calling "delete" on each tile
             // this is run here instead of inside the delete manager as we need high level info
             // about tiles, e.g., TileObject, to inform the listeners
-            List<Callable<?>> tilesDeletions = blobsToDelete
-                    .map(blobItem -> {
-                        TileObject tile = createTileObject(blobItem, tileRange);
-                        tile.setParametersId(tileRange.getParametersId());
-                        return (Callable<Object>) () -> delete(tile);
-                    })
-                    .collect(Collectors.toList());
+            Stream<Callable<?>> tilesDeletions = blobsToDelete.map(blobItem -> {
+                TileObject tile = createTileObject(blobItem, tileRange);
+                tile.setParametersId(tileRange.getParametersId());
+                return () -> delete(tile);
+            });
 
             executeParallelDeletions(tilesDeletions);
         }
@@ -278,13 +271,18 @@ private long[] extractTileIndex(BlobItem blobItem) {
         };
     }
 
-    private void executeParallelDeletions(List<Callable<?>> tilesDeletions) throws StorageException {
-        Iterator<List<Callable<?>>> tilesDeletionsPartitions =
-                Iterators.partition(tilesDeletions.iterator(), DeleteManager.PAGE_SIZE);
+    private void executeParallelDeletions(Stream<Callable<?>> tilesDeletions) throws StorageException {
+        Iterator<Callable<?>> tilesDeletionsIterator = tilesDeletions.iterator();
+
+        while (tilesDeletionsIterator.hasNext() && !shutDown) {
+
+            // once a page of callables is ready, run them in parallel on the delete manager
+            List<Callable<?>> callables = new ArrayList<>(PAGE_SIZE);
+            for (int i = 0; i < PAGE_SIZE && tilesDeletionsIterator.hasNext(); i++) {
+                callables.add(tilesDeletionsIterator.next());
+            }
 
-        // once a page of callables is ready, run them in parallel on the delete manager
-        while (tilesDeletionsPartitions.hasNext() && !shutDown) {
-            deleteManager.executeParallel(tilesDeletionsPartitions.next());
+            deleteManager.executeParallel(callables);
         }
     }
 

geowebcache/azureblob/src/main/java/org/geowebcache/azure/DeleteManager.java

@@ -26,6 +26,7 @@
 import java.time.OffsetDateTime;
 import java.util.ArrayList;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
@@ -39,6 +40,7 @@
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import org.geotools.util.logging.Logging;
 import org.geowebcache.GeoWebCacheException;
 import org.geowebcache.locks.LockProvider;
@@ -111,12 +113,12 @@ public void executeParallel(List<Callable<?>> callables) throws StorageException
         }
     }
 
-    /** Executes the removal of the specified keys in a parallel fashion, returning the number of removed keys */
-    public Long deleteParallel(List<String> keys) throws StorageException {
+    /** Executes the removal of the specified blobs in a streamed fashion, returning the number of removed keys */
+    public Long deleteStreamed(Stream<BlobItem> blobs) throws StorageException {
         try {
-            return new KeysBulkDelete(keys).call();
+            return new KeysBulkDelete(blobs.map(BlobItem::getName)).call();
         } catch (Exception e) {
-            throw new StorageException("Failed to submit parallel keys execution", e);
+            throw new StorageException("Failed to submit keys deletions", e);
         }
     }
 
@@ -305,29 +307,40 @@ private void clearPendingBulkDelete(final String prefix, final long timestamp) t
 
     public class KeysBulkDelete implements Callable<Long> {
 
-        private final List<String> keys;
+        private final Stream<String> keyStream;
 
-        public KeysBulkDelete(List<String> keys) {
-            this.keys = keys;
+        public KeysBulkDelete(Stream<String> keyStream) {
+            this.keyStream = keyStream;
         }
 
         @Override
         public Long call() throws Exception {
             long count = 0L;
             try {
                 checkInterrupted();
-                if (LOG.isLoggable(Level.FINER)) {
-                    LOG.finer("Running delete delete on list of items on '%s':%s ... (only the first 100 items listed)"
-                            .formatted(client.getContainerName(), keys.subList(0, Math.min(keys.size(), 100))));
-                }
 
                 BlobContainerClient container = client.getContainer();
                 BlobBatchClient batch = client.getBatch();
 
-                for (int i = 0; i < keys.size(); i += PAGE_SIZE) {
-                    count += deleteItems(container, batch, keys.subList(i, Math.min(i + PAGE_SIZE, keys.size())));
-                }
+                Iterator<String> keysIterator = keyStream.iterator();
+
+                while (keysIterator.hasNext()) {
+
+                    List<String> keys = new ArrayList<>(PAGE_SIZE);
+                    for (int i = 0; i < PAGE_SIZE && keysIterator.hasNext(); i++) {
+                        keys.add(keysIterator.next());
+                    }
+
+                    if (LOG.isLoggable(Level.FINER)) {
+                        LOG.finer(
+                                "Running delete delete on list of items on '%s':%s ... (only the first 100 items listed)"
+                                        .formatted(
+                                                client.getContainerName(),
+                                                keys.subList(0, Math.min(keys.size(), 100))));
+                    }
 
+                    count += deleteItems(container, batch, keys);
+                }
             } catch (InterruptedException | IllegalStateException e) {
                 LOG.log(Level.INFO, "Azure bulk delete aborted", e);
                 throw e;