fix(aws-android-sdk-s3): implement retry mechanism for upload part (#2504)

* implement retry mechanism for upload part

* reduce backoff time and max attempts

* lgtm warning

Author: raphkim

aws-android-sdk-s3/src/main/java/com/amazonaws/mobileconnectors/s3/transferutility/UploadPartTask.java

@@ -15,6 +15,7 @@
 
 package com.amazonaws.mobileconnectors.s3.transferutility;
 
+import com.amazonaws.AbortedException;
 import com.amazonaws.event.ProgressEvent;
 import com.amazonaws.event.ProgressListener;
 import com.amazonaws.services.s3.AmazonS3;
@@ -25,13 +26,14 @@
 import com.amazonaws.logging.LogFactory;
 
 import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
 
 class UploadPartTask implements Callable<Boolean> {
     private static final Log LOGGER = LogFactory.getLog(UploadPartTask.class);
-
+    private static final int RETRY_COUNT = 3;
 
     private final UploadTask.UploadPartTaskMetadata uploadPartTaskMetadata;
-    private final UploadTask.UploadTaskProgressListener uploadTaskProgressListener;
+    private final UploadPartTaskProgressListener uploadPartTaskProgressListener;
     private final UploadPartRequest uploadPartRequest;
     private final AmazonS3 s3;
     private final TransferDBUtil dbUtil;
@@ -42,7 +44,7 @@ public UploadPartTask(UploadTask.UploadPartTaskMetadata uploadPartTaskMetadata,
                           AmazonS3 s3,
                           TransferDBUtil dbUtil) {
         this.uploadPartTaskMetadata = uploadPartTaskMetadata;
-        this.uploadTaskProgressListener = uploadTaskProgressListener;
+        this.uploadPartTaskProgressListener = new UploadPartTaskProgressListener(uploadTaskProgressListener);
         this.uploadPartRequest = uploadPartRequest;
         this.s3 = s3;
         this.dbUtil = dbUtil;
@@ -53,53 +55,75 @@ public UploadPartTask(UploadTask.UploadPartTaskMetadata uploadPartTaskMetadata,
      */
     @Override
     public Boolean call() throws Exception {
-        try {
-            uploadPartTaskMetadata.state = TransferState.IN_PROGRESS;
-            uploadPartRequest.setGeneralProgressListener(new UploadPartTaskProgressListener(uploadTaskProgressListener));
-            final UploadPartResult putPartResult = s3.uploadPart(uploadPartRequest);
-            uploadPartTaskMetadata.state = TransferState.PART_COMPLETED;
-            dbUtil.updateState(uploadPartRequest.getId(), TransferState.PART_COMPLETED);
-            dbUtil.updateETag(uploadPartRequest.getId(), putPartResult.getETag());
-            return true;
-        } catch (final Exception e) {
-            LOGGER.error("Upload part interrupted: " + e);
-            ProgressEvent resetEvent = new ProgressEvent(0);
-            resetEvent.setEventCode(ProgressEvent.RESET_EVENT_CODE);
-            uploadTaskProgressListener.progressChanged(new ProgressEvent(0));
-
-            // Check if network is not connected, set the state to WAITING_FOR_NETWORK.
+        uploadPartTaskMetadata.state = TransferState.IN_PROGRESS;
+        uploadPartRequest.setGeneralProgressListener(uploadPartTaskProgressListener);
+        int retried = 1;
+        while (true) {
             try {
-                if (TransferNetworkLossHandler.getInstance() != null &&
-                    !TransferNetworkLossHandler.getInstance().isNetworkConnected()) {
-                    LOGGER.info("Thread: [" + Thread.currentThread().getId() + "]: Network wasn't available.");
-                    /*
-                     * Network connection is being interrupted. Moving the TransferState
-                     * to WAITING_FOR_NETWORK till the network availability resumes.
-                     */
-                    uploadPartTaskMetadata.state = TransferState.WAITING_FOR_NETWORK;
-                    dbUtil.updateState(uploadPartRequest.getId(), TransferState.WAITING_FOR_NETWORK);
-                    LOGGER.info("Network Connection Interrupted: " +
-                            "Moving the TransferState to WAITING_FOR_NETWORK");
-                    return false;
+                final UploadPartResult putPartResult = s3.uploadPart(uploadPartRequest);
+                setTaskState(TransferState.PART_COMPLETED);
+                dbUtil.updateETag(uploadPartRequest.getId(), putPartResult.getETag());
+                return true;
+            } catch (AbortedException e) {
+                // If request got aborted, operation was paused or canceled. do not retry.
+                LOGGER.debug("Upload part aborted.");
+                resetProgress();
+                return false;
+            } catch (final Exception e) {
+                LOGGER.error("Unexpected error occurred: " + e);
+                resetProgress();
+
+                // Check if network is not connected, set the state to WAITING_FOR_NETWORK.
+                try {
+                    if (TransferNetworkLossHandler.getInstance() != null &&
+                            !TransferNetworkLossHandler.getInstance().isNetworkConnected()) {
+                        LOGGER.info("Thread: [" + Thread.currentThread().getId() + "]: Network wasn't available.");
+                        /*
+                         * Network connection is being interrupted. Moving the TransferState
+                         * to WAITING_FOR_NETWORK till the network availability resumes.
+                         */
+                        uploadPartTaskMetadata.state = TransferState.WAITING_FOR_NETWORK;
+                        dbUtil.updateState(uploadPartRequest.getId(), TransferState.WAITING_FOR_NETWORK);
+                        LOGGER.info("Network Connection Interrupted: " +
+                                "Moving the TransferState to WAITING_FOR_NETWORK");
+                        return false;
+                    }
+                } catch (TransferUtilityException transferUtilityException) {
+                    LOGGER.error("TransferUtilityException: [" + transferUtilityException + "]");
+                }
+
+                if (retried >= RETRY_COUNT) {
+                    setTaskState(TransferState.FAILED);
+                    LOGGER.error("Encountered error uploading part ", e);
+                    throw e;
                 }
-            } catch (TransferUtilityException transferUtilityException) {
-                LOGGER.error("TransferUtilityException: [" + transferUtilityException + "]");
-            }
 
-            // In other cases, set the transfer state to FAILED.
-            uploadPartTaskMetadata.state = TransferState.FAILED;
-            dbUtil.updateState(uploadPartRequest.getId(), TransferState.FAILED);
-            LOGGER.error("Encountered error uploading part ", e);
-            throw e;
+                // Sleep before retrying
+                long delayMs = exponentialBackoffWithJitter(retried);
+                LOGGER.info("Retrying in " + delayMs + " ms.");
+                TimeUnit.MILLISECONDS.sleep(delayMs);
+                LOGGER.debug("Retry attempt: " + retried++, e);
+            }
         }
     }
 
+    private void setTaskState(TransferState newState) {
+        uploadPartTaskMetadata.state = newState;
+        dbUtil.updateState(uploadPartRequest.getId(),newState);
+    }
+
+    private void resetProgress() {
+        ProgressEvent resetEvent = new ProgressEvent(0);
+        resetEvent.setEventCode(ProgressEvent.RESET_EVENT_CODE);
+        uploadPartTaskProgressListener.progressChanged(resetEvent);
+    }
+
     /**
      * Progress Listener for a part
      */
     private class UploadPartTaskProgressListener implements ProgressListener {
 
-        private UploadTask.UploadTaskProgressListener uploadTaskProgressListener;
+        private final UploadTask.UploadTaskProgressListener uploadTaskProgressListener;
 
         private long bytesTransferredSoFar;
 
@@ -111,15 +135,24 @@ public UploadPartTaskProgressListener(UploadTask.UploadTaskProgressListener prog
         public void progressChanged(ProgressEvent progressEvent) {
             if (ProgressEvent.RESET_EVENT_CODE == progressEvent.getEventCode()) {
                 // Reset will discard what's been transferred
-                LOGGER.info("Reset Event triggered. Resetting the bytesCurrent to 0.");
+                LOGGER.debug("Reset Event triggered. Resetting the bytesCurrent to 0.");
                 // Reset the local counter to 0.
                 bytesTransferredSoFar = 0;
             } else {
                 bytesTransferredSoFar += progressEvent.getBytesTransferred();          
             }
-            this.uploadTaskProgressListener
-                    .onProgressChanged(UploadPartTask.this.uploadPartRequest.getPartNumber(), 
-                        bytesTransferredSoFar);
+            this.uploadTaskProgressListener.onProgressChanged(
+                    uploadPartRequest.getPartNumber(),
+                    bytesTransferredSoFar
+            );
         }
     }
+
+    private long exponentialBackoffWithJitter(int retryAttempt) {
+        final long baseTimeMs = 1000L;
+        final long jitterFactor = 1000L;
+        long delay = baseTimeMs * (1 << retryAttempt);
+        long jitter = (long) (jitterFactor * Math.random());
+        return delay + jitter;
+    }
 }