Implement skipIfExists for downloadBlobs in TransferManager

google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/DirectDownloadCallable.java

@@ -24,6 +24,7 @@
 import com.google.cloud.storage.StorageException;
 import com.google.common.io.ByteStreams;
 import java.nio.channels.FileChannel;
+import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.util.concurrent.Callable;
@@ -53,6 +54,11 @@ final class DirectDownloadCallable implements Callable<DownloadResult> {
 
   @Override
   public DownloadResult call() {
+    if (parallelDownloadConfig.isSkipIfExists() && Files.exists(destPath)) {
+      return DownloadResult.newBuilder(originalBlob, TransferStatus.SKIPPED)
+          .setOutputDestination(destPath)
+          .build();
+    }
     long bytesCopied = -1L;
     try (ReadChannel rc =
             storage.reader(

google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/ParallelDownloadConfig.java

@@ -34,22 +34,34 @@
  */
 public final class ParallelDownloadConfig {
 
+  private final boolean skipIfExists;
   @NonNull private final String stripPrefix;
   @NonNull private final Path downloadDirectory;
   @NonNull private final String bucketName;
   @NonNull private final List<BlobSourceOption> optionsPerRequest;
 
   private ParallelDownloadConfig(
+      boolean skipIfExists,
       @NonNull String stripPrefix,
       @NonNull Path downloadDirectory,
       @NonNull String bucketName,
       @NonNull List<BlobSourceOption> optionsPerRequest) {
+    this.skipIfExists = skipIfExists;
     this.stripPrefix = stripPrefix;
     this.downloadDirectory = downloadDirectory;
     this.bucketName = bucketName;
     this.optionsPerRequest = optionsPerRequest;
   }
 
+  /**
+   * If set Transfer Manager will skip downloading an object if it already exists.
+   *
+   * @see Builder#setSkipIfExists(boolean)
+   */
+  public boolean isSkipIfExists() {
+    return skipIfExists;
+  }
+
   /**
    * A common prefix removed from an object's name before being written to the filesystem.
    *
@@ -96,20 +108,23 @@ public boolean equals(Object o) {
       return false;
     }
     ParallelDownloadConfig that = (ParallelDownloadConfig) o;
-    return stripPrefix.equals(that.stripPrefix)
+    return skipIfExists == that.skipIfExists
+        && stripPrefix.equals(that.stripPrefix)
         && downloadDirectory.equals(that.downloadDirectory)
         && bucketName.equals(that.bucketName)
         && optionsPerRequest.equals(that.optionsPerRequest);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(stripPrefix, downloadDirectory, bucketName, optionsPerRequest);
+    return Objects.hash(
+        skipIfExists, stripPrefix, downloadDirectory, bucketName, optionsPerRequest);
   }
 
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
+        .add("skipIfExists", skipIfExists)
         .add("stripPrefix", stripPrefix)
         .add("downloadDirectory", downloadDirectory)
         .add("bucketName", bucketName)
@@ -128,18 +143,32 @@ public static Builder newBuilder() {
 
   public static final class Builder {
 
+    private boolean skipIfExists;
     @NonNull private String stripPrefix;
     @NonNull private Path downloadDirectory;
     @NonNull private String bucketName;
     @NonNull private List<BlobSourceOption> optionsPerRequest;
 
     private Builder() {
+      this.skipIfExists = false;
       this.stripPrefix = "";
       this.downloadDirectory = Paths.get("");
       this.bucketName = "";
       this.optionsPerRequest = ImmutableList.of();
     }
 
+    /**
+     * Sets the parameter for skipIfExists. When set to true Transfer Manager will skip downloading
+     * an object if it already exists.
+     *
+     * @return the builder instance with the value for skipIfExists modified.
+     * @see ParallelDownloadConfig#isSkipIfExists()
+     */
+    public Builder setSkipIfExists(boolean skipIfExists) {
+      this.skipIfExists = skipIfExists;
+      return this;
+    }
+
     /**
      * Sets the value for stripPrefix. This string will be removed from the beginning of all object
      * names before they are written to the filesystem.
@@ -197,7 +226,7 @@ public ParallelDownloadConfig build() {
       checkNotNull(downloadDirectory);
       checkNotNull(optionsPerRequest);
       return new ParallelDownloadConfig(
-          stripPrefix, downloadDirectory, bucketName, optionsPerRequest);
+          skipIfExists, stripPrefix, downloadDirectory, bucketName, optionsPerRequest);
     }
   }
 }

google-cloud-storage/src/main/java/com/google/cloud/storage/transfermanager/TransferManagerImpl.java

@@ -157,6 +157,14 @@ public void close() throws Exception {
         downloadTasks.add(ApiFutures.immediateFuture(skipped));
         continue;
       }
+      if (config.isSkipIfExists() && Files.exists(destPath)) {
+        DownloadResult skipped =
+            DownloadResult.newBuilder(blob, TransferStatus.SKIPPED)
+                .setOutputDestination(destPath)
+                .build();
+        downloadTasks.add(ApiFutures.immediateFuture(skipped));
+        continue;
+      }
       if (transferManagerConfig.isAllowDivideAndConquerDownload()) {
         BlobInfo validatedBlob = retrieveSizeAndGeneration(storage, blob, config.getBucketName());
         if (validatedBlob != null && qos.divideAndConquer(validatedBlob.getSize())) {

google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITTransferManagerTest.java

@@ -336,6 +336,34 @@ public void downloadBlobs() throws Exception {
     }
   }
 
+  @Test
+  public void downloadBlobsSkipIfExists() throws Exception {
+    TransferManagerConfig config =
+        TransferManagerConfigTestingInstances.defaults(storage.getOptions());
+    try (TransferManager transferManager = config.getService()) {
+      String bucketName = bucket.getName();
+      ParallelDownloadConfig parallelDownloadConfig =
+          ParallelDownloadConfig.newBuilder()
+              .setBucketName(bucketName)
+              .setDownloadDirectory(baseDir)
+              .setSkipIfExists(true)
+              .build();
+      DownloadJob job = transferManager.downloadBlobs(blobs, parallelDownloadConfig);
+      List<DownloadResult> downloadResults = job.getDownloadResults();
+      try {
+        assertThat(downloadResults).hasSize(3);
+        assertThat(downloadResults.get(0).getStatus()).isEqualTo(TransferStatus.SUCCESS);
+
+        DownloadJob job2 = transferManager.downloadBlobs(blobs, parallelDownloadConfig);
+        List<DownloadResult> downloadResults2 = job2.getDownloadResults();
+        assertThat(downloadResults2).hasSize(3);
+        assertThat(downloadResults2.get(0).getStatus()).isEqualTo(TransferStatus.SKIPPED);
+      } finally {
+        cleanUpFiles(downloadResults);
+      }
+    }
+  }
+
   @Test
   public void downloadBlobsAllowChunked() throws Exception {
     TransferManagerConfig config =

google-cloud-storage/src/test/java/com/google/cloud/storage/transfermanager/ParallelDownloadConfigTest.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.transfermanager;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.storage.Storage.BlobSourceOption;
+import com.google.common.collect.ImmutableList;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.List;
+import org.junit.Test;
+
+public final class ParallelDownloadConfigTest {
+
+  private static final String BUCKET_NAME = "test-bucket";
+  private static final String STRIP_PREFIX = "prefix/";
+  private static final Path DOWNLOAD_DIRECTORY = Paths.get("/tmp/downloads");
+  private static final List<BlobSourceOption> OPTIONS =
+      ImmutableList.of(BlobSourceOption.generationMatch(1L));
+
+  @Test
+  public void testBuilder() {
+    ParallelDownloadConfig config =
+        ParallelDownloadConfig.newBuilder()
+            .setBucketName(BUCKET_NAME)
+            .setStripPrefix(STRIP_PREFIX)
+            .setDownloadDirectory(DOWNLOAD_DIRECTORY)
+            .setOptionsPerRequest(OPTIONS)
+            .setSkipIfExists(true)
+            .build();
+
+    assertThat(config.getBucketName()).isEqualTo(BUCKET_NAME);
+    assertThat(config.getStripPrefix()).isEqualTo(STRIP_PREFIX);
+    assertThat(config.getDownloadDirectory())
+        .isEqualTo(DOWNLOAD_DIRECTORY.toAbsolutePath().normalize());
+    assertThat(config.getOptionsPerRequest()).isEqualTo(OPTIONS);
+    assertThat(config.isSkipIfExists()).isTrue();
+  }
+
+  @Test
+  public void testDefaultSkipIfExists() {
+    ParallelDownloadConfig config =
+        ParallelDownloadConfig.newBuilder().setBucketName(BUCKET_NAME).build();
+
+    assertThat(config.isSkipIfExists()).isFalse();
+  }
+
+  @Test
+  public void testEqualsAndHashCode() {
+    ParallelDownloadConfig config1 =
+        ParallelDownloadConfig.newBuilder()
+            .setBucketName(BUCKET_NAME)
+            .setSkipIfExists(true)
+            .build();
+    ParallelDownloadConfig config2 =
+        ParallelDownloadConfig.newBuilder()
+            .setBucketName(BUCKET_NAME)
+            .setSkipIfExists(true)
+            .build();
+    ParallelDownloadConfig config3 =
+        ParallelDownloadConfig.newBuilder()
+            .setBucketName(BUCKET_NAME)
+            .setSkipIfExists(false)
+            .build();
+
+    assertThat(config1).isEqualTo(config2);
+    assertThat(config1.hashCode()).isEqualTo(config2.hashCode());
+    assertThat(config1).isNotEqualTo(config3);
+    assertThat(config1.hashCode()).isNotEqualTo(config3.hashCode());
+  }
+
+  @Test
+  public void testToString() {
+    ParallelDownloadConfig config =
+        ParallelDownloadConfig.newBuilder()
+            .setBucketName(BUCKET_NAME)
+            .setSkipIfExists(true)
+            .build();
+
+    assertThat(config.toString()).contains("skipIfExists=true");
+    assertThat(config.toString()).contains("bucketName=" + BUCKET_NAME);
+  }
+}

google-cloud-storage/src/test/java/com/google/cloud/storage/transfermanager/TransferManagerTest.java

@@ -17,14 +17,46 @@
 package com.google.cloud.storage.transfermanager;
 
 import static com.google.common.truth.Truth.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
 import com.google.cloud.storage.transfermanager.ParallelUploadConfig.UploadBlobInfoFactory;
+import com.google.common.collect.ImmutableList;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
 import java.util.function.Function;
+import org.junit.Before;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
 
 public final class TransferManagerTest {
 
+  @Rule public final TemporaryFolder tmpDir = new TemporaryFolder();
+
+  private Storage storage;
+  private StorageOptions storageOptions;
+  private TransferManager transferManager;
+  private Path baseDir;
+
+  @Before
+  public void setUp() {
+    storage = mock(Storage.class);
+    storageOptions = mock(StorageOptions.class);
+    when(storageOptions.getService()).thenReturn(storage);
+    when(storageOptions.toBuilder()).thenReturn(StorageOptions.newBuilder());
+
+    TransferManagerConfig config =
+        TransferManagerConfig.newBuilder().setStorageOptions(storageOptions).build();
+    transferManager = config.getService();
+    baseDir = tmpDir.getRoot().toPath();
+  }
+
   @Test
   public void uploadBlobInfoFactory_prefixObjectNames_leadingSlash() {
     UploadBlobInfoFactory factory = UploadBlobInfoFactory.prefixObjectNames("asdf");
@@ -62,4 +94,27 @@ public void uploadBlobInfoFactory_default_doesNotModify() {
     assertThat(info.getBucket()).isEqualTo("bucket");
     assertThat(info.getName()).isEqualTo("/e.txt");
   }
+
+  @Test
+  public void downloadBlobs_skipIfExists() throws IOException {
+    String bucketName = "test-bucket";
+    String blobName = "test-blob.txt";
+    BlobInfo blobInfo = BlobInfo.newBuilder(bucketName, blobName).build();
+
+    Path destPath = baseDir.resolve(blobName);
+    Files.createFile(destPath); // Create the file locally
+
+    ParallelDownloadConfig config =
+        ParallelDownloadConfig.newBuilder()
+            .setBucketName(bucketName)
+            .setDownloadDirectory(baseDir)
+            .setSkipIfExists(true)
+            .build();
+
+    DownloadJob job = transferManager.downloadBlobs(ImmutableList.of(blobInfo), config);
+    List<DownloadResult> results = job.getDownloadResults();
+
+    assertThat(results).hasSize(1);
+    assertThat(results.get(0).getStatus()).isEqualTo(TransferStatus.SKIPPED);
+  }
 }