Refactor GoogleMediaExporter to call thirdparty getMediaItem with retrying & error skipping. (#1295)

* Test for the getGoogleMediaItem method

* fixed GoogleMediaExporter constructor to take in photo interface

* tweak the tests

* Add a test for a set of photos with one failing

* Made the test have two failures, such that we know the retrying worked and both got logged

* Added todos for retrying & interface stuff

* Remove VisibleForTesting on exportPhotosContainer

* add nullable to the nullable fields in the constructor

* Comment null fields with the field name

* One more comment

* Constructor signatures fixed, removed duplicate constructor

* new line

* reorder the other constructor

Author: osamahan999

extensions/data-transfer/portability-data-transfer-google/src/main/java/org/datatransferproject/datatransfer/google/GoogleTransferExtension.java

@@ -151,7 +151,7 @@ public void initialize(ExtensionContext context) {
         PHOTOS, new GooglePhotosExporter(credentialFactory, jobStore, jsonFactory, monitor));
     exporterBuilder.put(VIDEOS, new GoogleVideosExporter(credentialFactory, jsonFactory));
     exporterBuilder.put(
-        MEDIA, new GoogleMediaExporter(credentialFactory, jobStore, jsonFactory, monitor, idempotentImportExecutor, enableRetrying));
+        MEDIA, new GoogleMediaExporter(credentialFactory, jobStore, jsonFactory, monitor, /* photosInterface= */ null, idempotentImportExecutor, enableRetrying));
     exporterBuilder.put(MUSIC, new GoogleMusicExporter(credentialFactory, jsonFactory, monitor));
 
     exporterMap = exporterBuilder.build();

extensions/data-transfer/portability-data-transfer-google/src/main/java/org/datatransferproject/datatransfer/google/media/GoogleMediaExporter.java

@@ -15,6 +15,8 @@
  */
 package org.datatransferproject.datatransfer.google.media;
 
+import static java.lang.String.format;
+
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.api.client.auth.oauth2.Credential;
@@ -31,6 +33,7 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
+import javax.annotation.Nullable;
 import org.datatransferproject.api.launcher.Monitor;
 import org.datatransferproject.datatransfer.google.common.GoogleCredentialFactory;
 import org.datatransferproject.datatransfer.google.mediaModels.AlbumListResponse;
@@ -82,26 +85,8 @@ public GoogleMediaExporter(
         credentialFactory,
         jobStore,
         jsonFactory,
-        null,
-        monitor
-    );
-  }
-
-  public GoogleMediaExporter(
-      GoogleCredentialFactory credentialFactory,
-      TemporaryPerJobDataStore jobStore,
-      JsonFactory jsonFactory,
-      Monitor monitor,
-      IdempotentImportExecutor retryingExecutor,
-      boolean enableRetrying) {
-    this(
-        credentialFactory,
-        jobStore,
-        jsonFactory,
-        null,
         monitor,
-        retryingExecutor,
-        enableRetrying
+        /* photosInterface= */ null
     );
   }
 
@@ -110,24 +95,27 @@ public GoogleMediaExporter(
       GoogleCredentialFactory credentialFactory,
       TemporaryPerJobDataStore jobStore,
       JsonFactory jsonFactory,
-      GooglePhotosInterface photosInterface,
-      Monitor monitor) {
+      Monitor monitor, @Nullable
+      GooglePhotosInterface photosInterface
+  ) {
     this(
         credentialFactory,
         jobStore,
         jsonFactory,
-        photosInterface,
         monitor,
-        null,
+        photosInterface,
+        /* retryingExecutor= */ null,
         false);
   }
 
-  GoogleMediaExporter(GoogleCredentialFactory credentialFactory,
+  @VisibleForTesting
+  public GoogleMediaExporter(
+      GoogleCredentialFactory credentialFactory,
       TemporaryPerJobDataStore jobStore,
       JsonFactory jsonFactory,
-      GooglePhotosInterface photosInterface,
       Monitor monitor,
-      IdempotentImportExecutor retryingExecutor,
+      @Nullable GooglePhotosInterface photosInterface,
+      @Nullable IdempotentImportExecutor retryingExecutor,
       boolean enableRetrying) {
     this.credentialFactory = credentialFactory;
     this.jobStore = jobStore;
@@ -224,7 +212,11 @@ private ExportResult<MediaContainerResource> exportPhotosContainer(
 
     for (PhotoModel photo : container.getPhotos()) {
       GoogleMediaItem googleMediaItem =
-          getOrCreatePhotosInterface(authData).getMediaItem(photo.getDataId());
+          getGoogleMediaItem(photo.getIdempotentId(), photo.getDataId(), photo.getName(), authData);
+      if (googleMediaItem == null) {
+        continue;
+      }
+
       photosBuilder.add(GoogleMediaItem.convertToPhotoModel(Optional.empty(), googleMediaItem));
     }
 
@@ -253,14 +245,23 @@ private ExportResult<MediaContainerResource> exportMediaContainer(
     }
 
     for (PhotoModel photo : container.getPhotos()) {
-      GoogleMediaItem googleMediaItem =
-          getOrCreatePhotosInterface(authData).getMediaItem(photo.getDataId());
-      photosBuilder.add(GoogleMediaItem.convertToPhotoModel(Optional.empty(), googleMediaItem));
+      GoogleMediaItem photoMediaItem =
+          getGoogleMediaItem(photo.getIdempotentId(), photo.getDataId(), photo.getName(), authData);
+      if (photoMediaItem == null) {
+        continue;
+      }
+
+      photosBuilder.add(GoogleMediaItem.convertToPhotoModel(Optional.empty(), photoMediaItem));
     }
 
     for (VideoModel video : container.getVideos()) {
-      GoogleMediaItem googleMediaItem = getOrCreatePhotosInterface(authData).getMediaItem(video.getDataId());
-      videosBuilder.add(GoogleMediaItem.convertToVideoModel(Optional.empty(), googleMediaItem));
+      GoogleMediaItem videoMediaItem =
+          getGoogleMediaItem(video.getIdempotentId(), video.getDataId(), video.getName(), authData);
+      if (videoMediaItem == null) {
+        continue;
+      }
+
+      videosBuilder.add(GoogleMediaItem.convertToVideoModel(Optional.empty(), videoMediaItem));
     }
 
     MediaContainerResource mediaContainerResource =
@@ -310,7 +311,7 @@ ExportResult<MediaContainerResource> exportAlbums(
         albums.add(album);
 
         monitor.debug(
-            () -> String.format("%s: Google Photos exporting album: %s", jobId, album.getId()));
+            () -> format("%s: Google Photos exporting album: %s", jobId, album.getId()));
 
         // Add album id to continuation data
         continuationData.addContainerResource(new IdOnlyContainerResource(googleAlbum.getId()));
@@ -438,20 +439,41 @@ private MediaContainerResource convertMediaListToResource(
           photos.add(photoModel);
 
           monitor.debug(
-              () -> String.format("%s: Google exporting photo: %s", jobId, photoModel.getDataId()));
+              () -> format("%s: Google exporting photo: %s", jobId, photoModel.getDataId()));
         }
       } else if (mediaItem.isVideo()) {
         if (shouldUpload) {
           VideoModel videoModel = GoogleMediaItem.convertToVideoModel(albumId, mediaItem);
           videos.add(videoModel);
           monitor.debug(
-              () -> String.format("%s: Google exporting video: %s", jobId, videoModel.getDataId()));
+              () -> format("%s: Google exporting video: %s", jobId, videoModel.getDataId()));
         }
       }
     }
     return new MediaContainerResource(null  /*albums*/, photos, videos);
   }
 
+  //TODO(#1308): Make the retrying methods API & adaptor agnostic
+  @VisibleForTesting
+  @Nullable
+  GoogleMediaItem getGoogleMediaItem(String photoIdempotentId, String photoDataId,
+      String photoName, TokensAndUrlAuthData authData) throws IOException, InvalidTokenException, PermissionDeniedException {
+    if (retryingExecutor == null || !enableRetrying) {
+      return getOrCreatePhotosInterface(authData).getMediaItem(photoDataId);
+    }
+
+    try {
+      GoogleMediaItem googleMediaItem = retryingExecutor.executeAndSwallowIOExceptions(
+          photoIdempotentId, photoName,
+          () -> getOrCreatePhotosInterface(authData).getMediaItem(photoDataId)
+      );
+      return googleMediaItem;
+    } catch (Exception e) {
+      monitor.info(() -> format("Retry exception encountered while fetching a photo: %s", e));
+    }
+    return null;
+  }
+
   private synchronized GooglePhotosInterface getOrCreatePhotosInterface(
       TokensAndUrlAuthData authData) {
     return photosInterface == null ? makePhotosInterface(authData) : photosInterface;

extensions/data-transfer/portability-data-transfer-google/src/main/java/org/datatransferproject/datatransfer/google/photos/GooglePhotosInterface.java

@@ -65,6 +65,7 @@
 import org.datatransferproject.spi.transfer.types.PermissionDeniedException;
 import org.datatransferproject.spi.transfer.types.UploadErrorException;
 
+// TODO (#1307): Find a way to consolidate all 3P API interfaces
 public class GooglePhotosInterface {
 
   public static final String ERROR_HASH_MISMATCH = "Hash mismatch";

extensions/data-transfer/portability-data-transfer-google/src/test/java/org/datatransferproject/datatransfer/google/media/GoogleMediaExporterTest.java

@@ -20,6 +20,7 @@
 import static org.datatransferproject.datatransfer.google.media.GoogleMediaExporter.MEDIA_TOKEN_PREFIX;
 
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
@@ -30,6 +31,7 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.api.client.json.gson.GsonFactory;
+import com.google.common.collect.ImmutableList;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Collection;
@@ -49,24 +51,27 @@
 import org.datatransferproject.datatransfer.google.photos.GooglePhotosInterface;
 import org.datatransferproject.spi.cloud.storage.TemporaryPerJobDataStore;
 import org.datatransferproject.spi.cloud.storage.TemporaryPerJobDataStore.InputStreamWrapper;
+import org.datatransferproject.spi.transfer.idempotentexecutor.RetryingInMemoryIdempotentImportExecutor;
 import org.datatransferproject.spi.transfer.provider.ExportResult;
 import org.datatransferproject.spi.transfer.types.ContinuationData;
 import org.datatransferproject.spi.transfer.types.InvalidTokenException;
 import org.datatransferproject.spi.transfer.types.PermissionDeniedException;
 import org.datatransferproject.spi.transfer.types.TempMediaData;
 import org.datatransferproject.spi.transfer.types.UploadErrorException;
+import org.datatransferproject.types.common.ExportInformation;
 import org.datatransferproject.types.common.PaginationData;
 import org.datatransferproject.types.common.StringPaginationToken;
 import org.datatransferproject.types.common.models.ContainerResource;
 import org.datatransferproject.types.common.models.IdOnlyContainerResource;
 import org.datatransferproject.types.common.models.photos.PhotoAlbum;
 import org.datatransferproject.types.common.models.photos.PhotoModel;
 import org.datatransferproject.types.common.models.photos.PhotosContainerResource;
-import org.datatransferproject.types.common.models.videos.VideoAlbum;
 import org.datatransferproject.types.common.models.videos.VideoModel;
-import org.datatransferproject.types.common.models.videos.VideosContainerResource;
 import org.datatransferproject.types.common.models.media.MediaAlbum;
 import org.datatransferproject.types.common.models.media.MediaContainerResource;
+import org.datatransferproject.types.transfer.auth.TokensAndUrlAuthData;
+import org.datatransferproject.types.transfer.retry.RetryStrategyLibrary;
+import org.datatransferproject.types.transfer.retry.UniformRetryStrategy;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
@@ -78,9 +83,14 @@ public class GoogleMediaExporterTest {
   private static String ALBUM_TOKEN = "some-upstream-generated-album-token";
   private static String MEDIA_TOKEN = "some-upstream-generated-media-token";
 
-  private static UUID uuid = UUID.randomUUID();
+  private static long RETRY_INTERVAL_MILLIS = 100L;
+  private static int RETRY_MAX_ATTEMPTS = 1;
 
+  private static UUID uuid = UUID.randomUUID();
+  private TokensAndUrlAuthData authData;
+  private RetryingInMemoryIdempotentImportExecutor retryingExecutor;
   private GoogleMediaExporter googleMediaExporter;
+  private GoogleMediaExporter retryingGoogleMediaExporter;
   private TemporaryPerJobDataStore jobStore;
   private GooglePhotosInterface photosInterface;
 
@@ -99,10 +109,20 @@ public void setup()
     mediaItemSearchResponse = mock(MediaItemSearchResponse.class);
 
     Monitor monitor = mock(Monitor.class);
+    authData = mock(TokensAndUrlAuthData.class);
+
+    retryingExecutor = new RetryingInMemoryIdempotentImportExecutor(monitor,
+        new RetryStrategyLibrary(ImmutableList.of(), new UniformRetryStrategy(RETRY_MAX_ATTEMPTS, RETRY_INTERVAL_MILLIS, "identifier"))
+    );
 
     googleMediaExporter =
         new GoogleMediaExporter(
-            credentialFactory, jobStore, GsonFactory.getDefaultInstance(), photosInterface, monitor);
+            credentialFactory, jobStore, GsonFactory.getDefaultInstance(), monitor, photosInterface);
+
+    retryingGoogleMediaExporter = new GoogleMediaExporter(
+        credentialFactory, jobStore, GsonFactory.getDefaultInstance(), monitor, photosInterface,
+        retryingExecutor,
+        true);
 
     when(photosInterface.listAlbums(any(Optional.class))).thenReturn(albumListResponse);
     when(photosInterface.listMediaItems(any(Optional.class), any(Optional.class)))
@@ -354,6 +374,92 @@ public void onlyExportAlbumlessPhoto()
         .containsExactly(albumlessPhotoUri + "=d"); // download
   }
 
+  @Test
+  public void testGetGoogleMediaItemSucceeds() throws IOException, InvalidTokenException, PermissionDeniedException {
+    String mediaItemID = "media_id";
+    when(photosInterface.getMediaItem(any())).thenReturn(setUpSingleMediaItem(mediaItemID, mediaItemID, null));
+
+    assertThat(retryingGoogleMediaExporter.getGoogleMediaItem(mediaItemID, mediaItemID, mediaItemID, authData)).isInstanceOf(GoogleMediaItem.class);
+
+  }
+
+  @Test
+  public void testExportPhotosContainerRetrying() throws IOException, InvalidTokenException, PermissionDeniedException, UploadErrorException {
+    String PHOTO_ID_TO_FAIL_1 = "photo3";
+    String PHOTO_ID_TO_FAIL_2 = "photo5";
+
+    ImmutableList<PhotoAlbum> albums = ImmutableList.of();
+    ImmutableList<PhotoModel> photos = ImmutableList.of(
+        setUpSinglePhotoModel("", "photo1"),
+        setUpSinglePhotoModel("", "photo2"),
+        setUpSinglePhotoModel("", PHOTO_ID_TO_FAIL_1),
+        setUpSinglePhotoModel("", "photo4"),
+        setUpSinglePhotoModel("", PHOTO_ID_TO_FAIL_2),
+        setUpSinglePhotoModel("", "photo6")
+    );
+
+    PhotosContainerResource container = new PhotosContainerResource(albums, photos);
+    ExportInformation exportInfo = new ExportInformation(null, container);
+
+    MediaMetadata photoMediaMetadata = new MediaMetadata();
+    photoMediaMetadata.setPhoto(new Photo());
+
+
+    // For the photo_id_to_fail photos, throw an exception.
+    when(photosInterface.getMediaItem(PHOTO_ID_TO_FAIL_1)).thenThrow(IOException.class);
+    when(photosInterface.getMediaItem(PHOTO_ID_TO_FAIL_2)).thenThrow(IOException.class);
+    // For all other photos, return a media item.
+    for (PhotoModel photoModel: photos) {
+      if (photoModel.getDataId().equals(PHOTO_ID_TO_FAIL_1) || photoModel.getDataId().equals(PHOTO_ID_TO_FAIL_2)) {
+        continue;
+      }
+      when(photosInterface.getMediaItem(photoModel.getDataId())).thenReturn(
+          setUpSingleMediaItem(photoModel.getDataId(), photoModel.getDataId(), photoMediaMetadata)
+      );
+    }
+
+    ExportResult<MediaContainerResource> result = retryingGoogleMediaExporter.export(
+        uuid, authData, Optional.of(exportInfo)
+    );
+    assertThat(
+        result.getExportedData().getPhotos().stream().map(x -> x.getDataId()).collect(Collectors.toList())
+    ).isEqualTo(
+        photos.stream().map(
+            x -> x.getDataId()
+        ).filter(
+            dataId -> !(dataId.equals(PHOTO_ID_TO_FAIL_1) || dataId.equals(PHOTO_ID_TO_FAIL_2))
+        ).collect(
+            Collectors.toList()
+        )
+    );
+    assertThat(result.getExportedData().getPhotos().size()).isEqualTo(photos.size() - 2);
+    assertThat(retryingExecutor.getErrors().size()).isEqualTo(2);
+    assertThat(retryingExecutor.getErrors().stream().findFirst().toString().contains("IOException")).isTrue();
+  }
+
+  @Test
+  public void testGetGoogleMediaItemFailed() throws IOException, InvalidTokenException, PermissionDeniedException {
+    String mediaItemID = "media_id";
+    when(photosInterface.getMediaItem(mediaItemID)).thenThrow(IOException.class);
+
+    long start = System.currentTimeMillis();
+    assertThat(retryingGoogleMediaExporter.getGoogleMediaItem(mediaItemID, mediaItemID, mediaItemID, authData)).isNull();
+    long end = System.currentTimeMillis();
+
+    // If retrying occurred, then the retry_interval must have been waited at least max_attempts
+    // amount of times.
+    assertThat(end - start).isAtLeast(RETRY_INTERVAL_MILLIS * RETRY_MAX_ATTEMPTS);
+    assertThat(retryingExecutor.getErrors().size()).isEqualTo(1);
+    assertThat(retryingExecutor.getErrors().stream().findFirst().toString()).contains("IOException");
+
+
+    start = System.currentTimeMillis();
+    assertThrows(IOException.class, () -> googleMediaExporter.getGoogleMediaItem(mediaItemID, mediaItemID, mediaItemID, authData));
+    end = System.currentTimeMillis();
+
+    assertThat(end - start).isLessThan(RETRY_INTERVAL_MILLIS * RETRY_MAX_ATTEMPTS);
+  }
+
   /** Sets up a response with a single album, containing a single photo */
   private void setUpSingleAlbum() {
     GoogleAlbum albumEntry = new GoogleAlbum();
@@ -363,6 +469,10 @@ private void setUpSingleAlbum() {
     when(albumListResponse.getAlbums()).thenReturn(new GoogleAlbum[] {albumEntry});
   }
 
+  private static PhotoModel setUpSinglePhotoModel(String albumId, String dataId) {
+    return new PhotoModel("Title", "fetchableUrl", "description", "photo", dataId, albumId, false);
+  }
+
   /** Sets up a response for a single photo */
   // TODO(zacsh) delete this helper in favor of explicitly setting the fields that an assertion will
   // _actually_ use (and doing so _inlined, visibly_ in the arrange phase).