Add flag guarded retrying idempotent executor to flickr importer (#1244)

* Add flag guarded retrying idempotent executor to flickr importer * fixes
dtinit · May 25, 2023 · e00d334 · e00d334
1 parent 4e3154e
commit e00d334
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 4 deletions.
diff --git a/...kr/src/main/java/org/datatransferproject/datatransfer/flickr/FlickrTransferExtension.java b/...kr/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;
   }

diff --git a/...rc/main/java/org/datatransferproject/datatransfer/flickr/photos/FlickrPhotosImporter.java b/...rc/main/java/org/datatransferproject/datatransfer/flickr/photos/FlickrPhotosImporter.java
@@ -59,19 +59,33 @@ 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,
       TemporaryPerJobDataStore jobStore,
       Monitor monitor,
-      TransferServiceConfig serviceConfig) {
+      TransferServiceConfig serviceConfig,
+      IdempotentImportExecutor retryingIdempotentExecutor,
+      boolean enableRetrying) {
     this.jobStore = jobStore;
     this.flickr = new Flickr(appCredentials.getKey(), appCredentials.getSecret(), new REST());
     this.uploader = flickr.getUploader();
     this.connectionProvider = new ConnectionProvider(jobStore);
     this.photosetsInterface = flickr.getPhotosetsInterface();
     this.monitor = monitor;
     this.perUserRateLimiter = serviceConfig.getPerUserRateLimiter();
+    this.retryingIdempotentExecutor = retryingIdempotentExecutor;
+    this.enableRetrying = enableRetrying;
+  }
+
+  public FlickrPhotosImporter(
+      AppCredentials appCredentials,
+      TemporaryPerJobDataStore jobStore,
+      Monitor monitor,
+      TransferServiceConfig serviceConfig) {
+    this(appCredentials, jobStore, monitor, serviceConfig, null /*retryingIdempotentExecutor*/, false  /*enableRetrying*/);
   }
 
   @VisibleForTesting
@@ -112,10 +126,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);