Use retryingImporter in GooglePhotosImporter

extensions/data-transfer/portability-data-transfer-flickr/src/main/java/org/datatransferproject/datatransfer/flickr/FlickrTransferExtension.java

@@ -28,6 +28,8 @@
 import org.datatransferproject.datatransfer.flickr.photos.FlickrPhotosImporter;
 import org.datatransferproject.spi.cloud.storage.AppCredentialStore;
 import org.datatransferproject.spi.cloud.storage.TemporaryPerJobDataStore;
+import org.datatransferproject.spi.transfer.idempotentexecutor.IdempotentImportExecutor;
+import org.datatransferproject.spi.transfer.idempotentexecutor.IdempotentImportExecutorExtension;
 import org.datatransferproject.types.common.models.DataVertical;
 import org.datatransferproject.spi.transfer.extension.TransferExtension;
 import org.datatransferproject.spi.transfer.provider.Exporter;
@@ -36,6 +38,7 @@
 import org.datatransferproject.types.transfer.serviceconfig.TransferServiceConfig;
 
 public class FlickrTransferExtension implements TransferExtension {
+
   private static final String SERVICE_ID = "flickr";
   private static final String FLICKR_KEY = "FLICKR_KEY";
   private static final String FLICKR_SECRET = "FLICKR_SECRET";
@@ -69,7 +72,9 @@ public String getServiceId() {
 
   @Override
   public void initialize(ExtensionContext context) {
-    if (initialized) return;
+    if (initialized) {
+      return;
+    }
     jobStore = context.getService(TemporaryPerJobDataStore.class);
     Monitor monitor = context.getMonitor();
 
@@ -89,7 +94,11 @@ public void initialize(ExtensionContext context) {
 
     TransferServiceConfig serviceConfig = context.getService(TransferServiceConfig.class);
 
-    importer = new FlickrPhotosImporter(appCredentials, jobStore, monitor, serviceConfig);
+    IdempotentImportExecutor idempotentImportExecutor = context.getService(
+        IdempotentImportExecutorExtension.class).getRetryingIdempotentImportExecutor(context);
+    boolean enableRetrying = context.getSetting("enableRetrying", false);
+
+    importer = new FlickrPhotosImporter(appCredentials, jobStore, monitor, serviceConfig, idempotentImportExecutor, enableRetrying);
     exporter = new FlickrPhotosExporter(appCredentials, serviceConfig);
     initialized = true;
   }

extensions/data-transfer/portability-data-transfer-flickr/src/main/java/org/datatransferproject/datatransfer/flickr/photos/FlickrPhotosImporter.java

@@ -59,6 +59,10 @@ public class FlickrPhotosImporter implements Importer<AuthData, PhotosContainerR
   private final PhotosetsInterface photosetsInterface;
   private final Monitor monitor;
   private final RateLimiter perUserRateLimiter;
+  private IdempotentImportExecutor retryingIdempotentExecutor;
+  private Boolean enableRetrying;
+
+
 
   public FlickrPhotosImporter(
       AppCredentials appCredentials,
@@ -74,6 +78,18 @@ public FlickrPhotosImporter(
     this.perUserRateLimiter = serviceConfig.getPerUserRateLimiter();
   }
 
+  public FlickrPhotosImporter(
+      AppCredentials appCredentials,
+      TemporaryPerJobDataStore jobStore,
+      Monitor monitor,
+      TransferServiceConfig serviceConfig,
+      IdempotentImportExecutor retryingIdempotentExecutor,
+      Boolean enableRetrying) {
+    this(appCredentials, jobStore, monitor, serviceConfig);
+    this.retryingIdempotentExecutor = retryingIdempotentExecutor;
+    this.enableRetrying = enableRetrying;
+  }
+
   @VisibleForTesting
   FlickrPhotosImporter(
       Flickr flickr,
@@ -112,10 +128,13 @@ public ImportResult importItem(
       storeAlbums(jobId, data.getAlbums());
     }
 
+      IdempotentImportExecutor executor =
+          (retryingIdempotentExecutor != null && enableRetrying) ? retryingIdempotentExecutor : idempotentExecutor;
+
     if (data.getPhotos() != null) {
       for (PhotoModel photo : data.getPhotos()) {
         try {
-          importSinglePhoto(idempotentExecutor, jobId, photo);
+          importSinglePhoto(executor, jobId, photo);
         } catch (FlickrException e) {
           if (e.getMessage().contains("Upload limit reached")) {
             throw new DestinationMemoryFullException("Flickr destination memory reached", e);

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

@@ -37,6 +37,8 @@
 import org.datatransferproject.datatransfer.google.videos.GoogleVideosImporter;
 import org.datatransferproject.spi.cloud.storage.AppCredentialStore;
 import org.datatransferproject.spi.cloud.storage.JobStore;
+import org.datatransferproject.spi.transfer.idempotentexecutor.IdempotentImportExecutor;
+import org.datatransferproject.spi.transfer.idempotentexecutor.IdempotentImportExecutorExtension;
 import org.datatransferproject.types.common.models.DataVertical;
 import org.datatransferproject.spi.transfer.extension.TransferExtension;
 import org.datatransferproject.spi.transfer.provider.Exporter;
@@ -107,6 +109,9 @@ public void initialize(ExtensionContext context) {
     GoogleCredentialFactory credentialFactory =
         new GoogleCredentialFactory(httpTransport, jsonFactory, appCredentials, monitor);
 
+    IdempotentImportExecutor idempotentImportExecutor = context.getService(
+        IdempotentImportExecutorExtension.class).getRetryingIdempotentImportExecutor(context);
+
     ImmutableMap.Builder<DataVertical, Importer> importerBuilder = ImmutableMap.builder();
     importerBuilder.put(BLOBS, new DriveImporter(credentialFactory, jobStore, monitor));
     importerBuilder.put(CONTACTS, new GoogleContactsImporter(credentialFactory));
@@ -120,7 +125,8 @@ public void initialize(ExtensionContext context) {
             jobStore,
             jsonFactory,
             monitor,
-            context.getSetting("googleWritesPerSecond", 1.0)));
+            context.getSetting("googleWritesPerSecond", 1.0),
+            idempotentImportExecutor));
     importerBuilder.put(VIDEOS, new GoogleVideosImporter(appCredentials, jobStore, monitor));
     importerMap = importerBuilder.build();
 

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

@@ -79,6 +79,8 @@ public class GooglePhotosImporter
   private final GooglePhotosInterface photosInterface;
   private final HashMap<UUID, BaseMultilingualDictionary> multilingualStrings = new HashMap<>();
 
+  private IdempotentImportExecutor retryingIdempotentExecutor;
+
   public GooglePhotosImporter(
       GoogleCredentialFactory credentialFactory,
       JobStore jobStore,
@@ -95,6 +97,21 @@ public GooglePhotosImporter(
         monitor,
         writesPerSecond);
   }
+  public GooglePhotosImporter(
+      GoogleCredentialFactory credentialFactory,
+      JobStore jobStore,
+      JsonFactory jsonFactory,
+      Monitor monitor,
+      double writesPerSecond,
+      IdempotentImportExecutor retryingIdempotentExecutor) {
+    this(
+        credentialFactory,
+        jobStore,
+        jsonFactory,
+        monitor,
+        writesPerSecond);
+    this.retryingIdempotentExecutor = retryingIdempotentExecutor;
+  }
 
   @VisibleForTesting
   GooglePhotosImporter(
@@ -131,10 +148,11 @@ public ImportResult importItem(
       // Nothing to do
       return ImportResult.OK;
     }
-    GPhotosUpload gPhotosUpload = new GPhotosUpload(jobId, idempotentImportExecutor, authData);
+    IdempotentImportExecutor executor = retryingIdempotentExecutor == null ? idempotentImportExecutor : retryingIdempotentExecutor;
+    GPhotosUpload gPhotosUpload = new GPhotosUpload(jobId, executor, authData);
 
     for (PhotoAlbum album : data.getAlbums()) {
-      idempotentImportExecutor.executeAndSwallowIOExceptions(
+      executor.executeAndSwallowIOExceptions(
           album.getId(), album.getName(), () -> importSingleAlbum(jobId, authData, album));
     }
     long bytes = importPhotos(data.getPhotos(), gPhotosUpload);