[JENKINS-76354] - resolve indefinite hang in s3Upload caused by untimeout'd future join (#364)

Signed-off-by: prathamalwayscomeslast <way4hell6@gmail.com>

Author: prathamalwayscomeslast

src/main/java/hudson/plugins/s3/S3Profile.java

@@ -29,6 +29,8 @@
 import java.util.Map;
 import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 
 public class S3Profile {
     private final String name;
@@ -275,15 +277,21 @@ public FingerprintRecord call() throws IOException, InterruptedException {
 
     private <T> T repeat(int maxRetries, int waitTime, Destination dest, Callable<T> func) throws IOException, InterruptedException {
         int retryCount = 0;
+        Logger LOGGER = Logger.getLogger(S3Profile.class.getName());
 
         while (true) {
             try {
                 return func.call();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw e;
             } catch (Exception e) {
                 retryCount++;
                 if(retryCount >= maxRetries){
                     throw new IOException("Call fails for " + dest + ": " + e + ":: Failed after " + retryCount + " tries.", e);
                 }
+                LOGGER.log(Level.WARNING, "Upload attempt {0}/{1} failed for {2}: {3}. Retrying in {4}s",
+                        new Object[]{retryCount, maxRetries, dest, e.getMessage(), waitTime});
                 Thread.sleep(TimeUnit.SECONDS.toMillis(waitTime));
             }
         }

src/main/java/hudson/plugins/s3/Uploads.java

@@ -4,7 +4,6 @@
 import software.amazon.awssdk.core.async.AsyncRequestBody;
 import software.amazon.awssdk.services.s3.model.PutObjectRequest;
 import software.amazon.awssdk.transfer.s3.S3TransferManager;
-import software.amazon.awssdk.transfer.s3.model.FileUpload;
 import software.amazon.awssdk.transfer.s3.model.Upload;
 import software.amazon.awssdk.transfer.s3.model.UploadRequest;
 import software.amazon.awssdk.transfer.s3.progress.TransferListener;
@@ -14,8 +13,14 @@
 import java.io.InputStream;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.function.Consumer;
 import java.util.logging.Logger;
 
@@ -25,9 +30,22 @@ private Uploads() {}
     public static final int MULTIPART_UPLOAD_THRESHOLD = 16*1024*1024; // 16 MB
 
     private static transient volatile Uploads instance;
-    private final transient HashMap<FilePath, Upload> startedUploads = new HashMap<>();
-    private final ExecutorService executors = Executors.newScheduledThreadPool(1, new NamedThreadFactory(Executors.defaultThreadFactory(), Uploads.class.getName()));
-    private final transient HashMap<FilePath, InputStream> openedStreams = new HashMap<>();
+
+    private final transient Map<FilePath, Upload> startedUploads = new ConcurrentHashMap<>();
+    private final ExecutorService executors;
+    // This creates a cached thread pool with an upper bound (5) on threads to be spawned on demand.
+    {
+        ThreadPoolExecutor pool = new ThreadPoolExecutor(
+            5, 5,
+            60L, TimeUnit.SECONDS,
+            new LinkedBlockingQueue<>(),
+            new NamedThreadFactory(Executors.defaultThreadFactory(), Uploads.class.getName())
+        );
+        pool.allowCoreThreadTimeOut(true);
+        executors = pool;
+    }
+
+    private final transient Map<FilePath, InputStream> openedStreams = new ConcurrentHashMap<>();
 
     public Upload startUploading(S3TransferManager manager, FilePath file, InputStream inputStream, String bucketName, String objectName, Metadata metadata, TransferListener listener) {
         UploadRequest.Builder request = UploadRequest.builder();
@@ -43,16 +61,24 @@ public Upload startUploading(S3TransferManager manager, FilePath file, InputStre
         return upload;
     }
 
-    public void finishUploading(FilePath filePath) throws InterruptedException {
+    public void finishUploading(FilePath filePath) throws InterruptedException, IOException {
         final Upload upload = startedUploads.remove(filePath);
         if (upload == null) {
             LOGGER.info("File: " + filePath.getName() + " already was uploaded");
             return;
         }
         try {
-            upload.completionFuture().join();
-        }
-        finally {
+            upload.completionFuture().get(1, TimeUnit.HOURS);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            upload.completionFuture().cancel(true); // cancel the upload
+            throw e;
+        } catch (ExecutionException e) {
+            throw new IOException("Upload failed for: " + filePath.getName(), e.getCause());
+        } catch (TimeoutException e) {
+            upload.completionFuture().cancel(true);
+            throw new IOException("Upload timed out for: " + filePath.getName(), e);
+        } finally {
             closeStream(filePath);
         }
     }

src/main/java/hudson/plugins/s3/callable/S3WaitUploadCallable.java

@@ -7,16 +7,17 @@
 import org.jenkinsci.remoting.RoleChecker;
 
 import java.io.File;
+import java.io.IOException;
 
 public final class S3WaitUploadCallable implements MasterSlaveCallable<Void> {
     @Override
-    public Void invoke(File f, VirtualChannel channel) throws InterruptedException {
+    public Void invoke(File f, VirtualChannel channel) throws InterruptedException, IOException {
         invoke(new FilePath(f));
         return null;
     }
 
     @Override
-    public Void invoke(FilePath file) throws InterruptedException {
+    public Void invoke(FilePath file) throws InterruptedException, IOException {
         Uploads.getInstance().finishUploading(file);
         return null;
     }