Implement inMemeoryIdempotentExecutorExtension and update loader and worker (#1210)

* Bump version after publishing and remove the end /'s from the repository URLs

* Implement inMemeoryIdempotentExecutorExtension and update loader and worker

* Revert "Merge branch 'skippable2' of https://github.com/google/data-transfer-project into skippable2"

This reverts commit 21dfbd9, reversing
changes made to b90bf5b.

---------

Co-authored-by: William Morland <wmorland@fb.com>

Author: kateyeo

portability-spi-transfer/src/main/java/org/datatransferproject/spi/transfer/idempotentexecutor/IdempotentImportExecutorLoader.java

@@ -7,18 +7,17 @@
 
 public class IdempotentImportExecutorLoader {
 
-  public static IdempotentImportExecutor load(ExtensionContext extensionContext) {
+  public static IdempotentImportExecutorExtension load(ExtensionContext extensionContext) {
     ImmutableList.Builder<IdempotentImportExecutorExtension> builder = ImmutableList.builder();
     ServiceLoader.load(IdempotentImportExecutorExtension.class)
         .iterator()
         .forEachRemaining(builder::add);
     ImmutableList<IdempotentImportExecutorExtension> executors = builder.build();
     if (executors.isEmpty()) {
-      return new InMemoryIdempotentImportExecutor(extensionContext.getMonitor());
+      return new InMemoryIdempotentImportExecutorExtension();
     } else if (executors.size() == 1) {
       IdempotentImportExecutorExtension extension = executors.get(0);
-      extension.initialize();
-      return extension.getIdempotentImportExecutor(extensionContext);
+      return extension;
     } else {
       throw new IllegalStateException("Cannot load multiple IdempotentImportExecutors");
     }

portability-spi-transfer/src/main/java/org/datatransferproject/spi/transfer/idempotentexecutor/InMemoryIdempotentImportExecutorExtension.java

@@ -0,0 +1,20 @@
+package org.datatransferproject.spi.transfer.idempotentexecutor;
+
+import org.datatransferproject.api.launcher.ExtensionContext;
+import org.datatransferproject.api.launcher.Monitor;
+
+/**
+ * ImMemory Implementation of IdempotentImportExecutor.
+ */
+public class InMemoryIdempotentImportExecutorExtension
+    implements IdempotentImportExecutorExtension {
+
+  @Override
+  public IdempotentImportExecutor getIdempotentImportExecutor(ExtensionContext extensionContext) {
+    return new InMemoryIdempotentImportExecutor(extensionContext.getMonitor());
+  }
+
+  @Override
+  public void initialize() {
+  }
+}

portability-transfer/src/main/java/org/datatransferproject/transfer/WorkerMain.java

@@ -44,6 +44,7 @@
 import org.datatransferproject.spi.transfer.extension.TransferExtension;
 import org.datatransferproject.spi.transfer.hooks.JobHooks;
 import org.datatransferproject.spi.transfer.idempotentexecutor.IdempotentImportExecutor;
+import org.datatransferproject.spi.transfer.idempotentexecutor.IdempotentImportExecutorExtension;
 import org.datatransferproject.spi.transfer.idempotentexecutor.IdempotentImportExecutorLoader;
 import org.datatransferproject.spi.transfer.provider.TransferCompatibilityProvider;
 import org.datatransferproject.spi.transfer.security.SecurityExtension;
@@ -103,10 +104,14 @@ public void initialize() {
         SecurityExtensionLoader.getSecurityExtension(extensionContext);
     monitor.info(() -> "Using SecurityExtension: " + securityExtension.getClass().getName());
 
-    IdempotentImportExecutor idempotentImportExecutor =
+    IdempotentImportExecutorExtension idempotentImportExecutorExtension =
         IdempotentImportExecutorLoader.load(extensionContext);
+
+    extensionContext.registerService(
+        IdempotentImportExecutorExtension.class, idempotentImportExecutorExtension);
+
     monitor.info(
-        () -> "Using IdempotentImportExecutor: " + idempotentImportExecutor.getClass().getName());
+        () -> "Using IdempotentImportExecutor: " + idempotentImportExecutorExtension.getClass().getName());
 
     // TODO: make configurable
     SymmetricKeyGenerator symmetricKeyGenerator = new AesSymmetricKeyGenerator(monitor);
@@ -122,7 +127,7 @@ public void initialize() {
                   cloudExtension,
                   transferExtensions,
                   securityExtension,
-                  idempotentImportExecutor,
+                  idempotentImportExecutorExtension.getIdempotentImportExecutor(extensionContext),
                   symmetricKeyGenerator,
                   jobHooks,
                   new TransferCompatibilityProvider()));