Cancel pending replications upon shutdown

When the replication plugin is stopped or reloaded,
mark all the current replications as cancelled.
Having pending and unmanaged replication tasks is quite
dangerous for two reasons:

1. The result of the replication isn't accounted and
   properly managed, because the associated objects won't
   be there anymore (pending replications, runway, ...)

2. The destination configuration could have changed completely
   (auto-reload use-case) or even not exist anymore: the
   replication event would thus perform an unwanted and
   unconfigured operation.

With regards to the replication events that are already
on the runway, there is no value in cancelling them. Just flag
them as cancelled so that, once they finish, can be clearly
recognized.

Change-Id: Iabc17d375011cbc61f8c642ae07d3d018b49fc69

2 files changed, 38 insertions, 6 deletions

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java b/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
index 9aab0f6..e8929d5 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
@@ -69,6 +69,7 @@ import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
 import org.apache.commons.io.FilenameUtils;
 import org.eclipse.jgit.lib.Constants;
 import org.eclipse.jgit.lib.Ref;
@@ -217,12 +218,32 @@ public class Destination {
     if (pool != null) {
       repLog.warn("Cancelling replication events");
 
+      foreachPushOp(
+          pending,
+          push -> {
+            push.cancel();
+            return null;
+          });
+      pending.clear();
+      foreachPushOp(
+          inFlight,
+          push -> {
+            push.setCanceledWhileRunning();
+            return null;
+          });
+      inFlight.clear();
       cnt = pool.shutdownNow().size();
       pool = null;
     }
     return cnt;
   }
 
+  private void foreachPushOp(Map<URIish, PushOne> opsMap, Function<PushOne, Void> pushOneFunction) {
+    for (PushOne pushOne : ImmutableList.copyOf(opsMap.values())) {
+      pushOneFunction.apply(pushOne);
+    }
+  }
+
   private boolean shouldReplicate(ProjectControl ctl) throws PermissionBackendException {
     if (!config.replicateHiddenProjects() && ctl.getProject().getState() == ProjectState.HIDDEN) {
       return false;
@@ -309,7 +330,7 @@ public class Destination {
     if (!config.replicatePermissions()) {
       PushOne e;
       synchronized (stateLock) {
-        e = pending.get(uri);
+        e = getPendingPush(uri);
       }
       if (e == null) {
         try (Repository git = gitManager.openRepository(project)) {
@@ -332,7 +353,7 @@ public class Destination {
     }
 
     synchronized (stateLock) {
-      PushOne e = pending.get(uri);
+      PushOne e = getPendingPush(uri);
       if (e == null) {
         e = opFactory.create(project, uri);
         addRef(e, ref);
@@ -348,6 +369,14 @@ public class Destination {
     }
   }
 
+  private PushOne getPendingPush(URIish uri) {
+    PushOne e = pending.get(uri);
+    if (e != null && !e.wasCanceled()) {
+      return e;
+    }
+    return null;
+  }
+
   void pushWasCanceled(PushOne pushOp) {
     synchronized (stateLock) {
       URIish uri = pushOp.getURI();
@@ -384,7 +413,7 @@ public class Destination {
   void reschedule(PushOne pushOp, RetryReason reason) {
     synchronized (stateLock) {
       URIish uri = pushOp.getURI();
-      PushOne pendingPushOp = pending.get(uri);
+      PushOne pendingPushOp = getPendingPush(uri);
 
       if (pendingPushOp != null) {
         // There is one PushOp instance already pending to same URI.
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/PushOne.java b/src/main/java/com/googlesource/gerrit/plugins/replication/PushOne.java
index 889e94d..9fe497b 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/PushOne.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/PushOne.java
@@ -149,14 +149,17 @@ class PushOne implements ProjectRunnable, CanceledWhileRunning {
 
   @Override
   public void cancel() {
-    repLog.info("Replication {} was canceled", getURI());
+    repLog.info("Replication [{}] to {} was canceled", IdGenerator.format(id), getURI());
     canceledByReplication();
     pool.pushWasCanceled(this);
   }
 
   @Override
   public void setCanceledWhileRunning() {
-    repLog.info("Replication {} was canceled while being executed", getURI());
+    repLog.info(
+        "Replication [{}] to {} was canceled while being executed",
+        IdGenerator.format(id),
+        getURI());
     canceledWhileRunning.set(true);
   }
 
@@ -200,7 +203,7 @@ class PushOne implements ProjectRunnable, CanceledWhileRunning {
   }
 
   boolean wasCanceled() {
-    return canceled;
+    return canceled || canceledWhileRunning.get();
   }
 
   URIish getURI() {