mbwaheed commented on a change in pull request #1188: SOLR-14044: Support
collection and shard deletion in shared storage
URL: https://github.com/apache/lucene-solr/pull/1188#discussion_r374929762
##########
File path:
solr/core/src/java/org/apache/solr/store/blob/process/BlobDeleterTask.java
##########
@@ -18,94 +18,244 @@
package org.apache.solr.store.blob.process;
import java.lang.invoke.MethodHandles;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
import java.util.Locale;
import java.util.Set;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.solr.store.blob.client.CoreStorageClient;
+import
org.apache.solr.store.blob.process.BlobDeleterTask.BlobDeleterTaskResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * Task in charge of deleting Blobs (files) from blob store.
+ * Generic deletion task for files located on shared storage
*/
-class BlobDeleterTask implements Runnable {
+public abstract class BlobDeleterTask implements
Callable<BlobDeleterTaskResult> {
private static final Logger log =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
- /**
- * Note we sleep() after each failed attempt, so multiply this value by
{@link #SLEEP_MS_FAILED_ATTEMPT} to find
- * out how long we'll retry (at least) if Blob access fails for some reason
("at least" because we
- * re-enqueue at the tail of the queue ({@link BlobDeleteManager} creates a
list), so there might be additional
- * processing delay if the queue is not empty and is processed before the
enqueued retry is processed).
- */
- private static int MAX_DELETE_ATTEMPTS = 50;
- private static long SLEEP_MS_FAILED_ATTEMPT = TimeUnit.SECONDS.toMillis(10);
-
+
private final CoreStorageClient client;
- private final String sharedBlobName;
- private final Set<String> blobNames;
+ private final String collectionName;
private final AtomicInteger attempt;
- private final ThreadPoolExecutor executor;
+
private final long queuedTimeMs;
+ private final int maxAttempts;
+ private final boolean allowRetry;
+ private Throwable err;
- BlobDeleterTask(CoreStorageClient client, String sharedBlobName, Set<String>
blobNames, ThreadPoolExecutor executor) {
- this.client = client;
- this.sharedBlobName = sharedBlobName;
- this.blobNames = blobNames;
+ public BlobDeleterTask(CoreStorageClient client, String collectionName,
boolean allowRetry,
+ int maxAttempts) {
+ this.client = client;
+ this.collectionName = collectionName;
this.attempt = new AtomicInteger(0);
- this.executor = executor;
- this.queuedTimeMs = System.nanoTime();
+ this.queuedTimeMs = System.nanoTime() / 1000000;
+ this.allowRetry = allowRetry;
+ this.maxAttempts = maxAttempts;
}
-
+
+ /**
+ * Performs a deletion action and request against the shared storage for the
given collection
+ * and returns the list of file paths deleted
+ */
+ public abstract Collection<String> doDelete() throws Exception;
+
+ /**
+ * Return a String representing the action performed by the BlobDeleterTask
for logging purposes
+ */
+ public abstract String getActionName();
+
@Override
- public void run() {
- final long startTimeMs = System.nanoTime();
+ public BlobDeleterTaskResult call() {
+ List<String> filesDeleted = new LinkedList<>();
+ final long startTimeMs = System.nanoTime() / 1000000;
boolean isSuccess = true;
-
+ boolean shouldRetry = false;
try {
+ filesDeleted.addAll(doDelete());
+ attempt.incrementAndGet();
+ return new BlobDeleterTaskResult(this, filesDeleted, isSuccess,
shouldRetry, err);
+ } catch (Exception ex) {
+ if (err == null) {
+ err = ex;
+ } else {
+ err.addSuppressed(ex);
+ }
+ int attempts = attempt.incrementAndGet();
+ isSuccess = false;
+ log.warn("BlobDeleterTask failed on attempt=" + attempts + "
collection=" + collectionName
+ + " task=" + toString(), ex);
+ if (allowRetry) {
+ if (attempts < maxAttempts) {
+ shouldRetry = true;
+ } else {
+ log.warn("Reached " + maxAttempts + " attempt limit for deletion
task " + toString() +
+ ". This task won't be retried.");
+ }
+ }
+ } finally {
+ long now = System.nanoTime() / 1000000;
+ long runTime = now - startTimeMs;
+ long startLatency = now - this.queuedTimeMs;
+ log(getActionName(), collectionName, runTime, startLatency, isSuccess,
getAdditionalLogMessage());
+ }
+ return new BlobDeleterTaskResult(this, filesDeleted, isSuccess,
shouldRetry, err);
+ }
+
+ /**
+ * Override-able by deletion tasks to provide additional action specific
logging
+ */
+ public String getAdditionalLogMessage() {
+ return "";
+ }
+
+ @Override
+ public String toString() {
+ return "collectionName=" + collectionName + " allowRetry=" + allowRetry +
+ " queuedTimeMs=" + queuedTimeMs + " attemptsTried=" + attempt.get();
+ }
+
+ public int getAttempts() {
+ return attempt.get();
+ }
+
+ public void log(String action, String collectionName, long runTime, long
startLatency, boolean isSuccess,
+ String additionalMessage) {
+ String message = String.format(Locale.ROOT,
+ "action=%s storageProvider=%s bucketRegion=%s bucketName=%s,
runTime=%s "
+ + "startLatency=%s attempt=%s isSuccess=%s %s",
+ action, client.getStorageProvider().name(), client.getBucketRegion(),
client.getBucketName(),
+ runTime, startLatency, attempt.get(), isSuccess, additionalMessage);
+ log.info(message);
+ }
+
+ /**
+ * Represents the result of a deletion task
+ */
+ public static class BlobDeleterTaskResult {
+ private final BlobDeleterTask task;
+ private final Collection<String> filesDeleted;
+ private final boolean isSuccess;
+ private final boolean shouldRetry;
+ private final Throwable err;
+
+ public BlobDeleterTaskResult(BlobDeleterTask task, Collection<String>
filesDeleted,
+ boolean isSuccess, boolean shouldRetry, Throwable errs) {
+ this.task = task;
+ this.filesDeleted = filesDeleted;
+ this.isSuccess = isSuccess;
+ this.shouldRetry = shouldRetry;
+ this.err = errs;
+ }
+
+ public boolean isSuccess() {
+ return isSuccess;
+ }
+
+ public boolean shouldRetry() {
+ return shouldRetry;
+ }
+
+ public BlobDeleterTask getTask() {
+ return task;
+ }
+
+ /**
+ * @return the files that are being deleted. Note if the task wasn't
successful there is no gaurantee
Review comment:
typo: gaurantee
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@lucene.apache.org
For additional commands, e-mail: issues-h...@lucene.apache.org
