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

src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java

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.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 @@
     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 @@
     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 @@
     }
 
     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 @@
     }
   }
 
+  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 @@
   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.

src/main/java/com/googlesource/gerrit/plugins/replication/PushOne.java

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 @@
 
   @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 @@
   }
 
   boolean wasCanceled() {
-    return canceled;
+    return canceled || canceledWhileRunning.get();
   }
 
   URIish getURI() {