Merge branch 'stable-3.1' into stable-3.2
* stable-3.1:
Improve readability of shouldFirePendingOnlyToStoredUri test
Fix flakiness in ReplicationIT for pending events firing
Only fire the specified pending event URI
Change-Id: Ie83763e4a9fe13522f356b569fc2360fa5883224
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 adc3133..8f04c74 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
@@ -609,6 +609,10 @@
return names;
}
+ boolean wouldPush(URIish uri, Project.NameKey project, String ref) {
+ return matches(uri, project) && wouldPushProject(project) && wouldPushRef(ref);
+ }
+
boolean wouldPushProject(Project.NameKey project) {
if (!shouldReplicate(project)) {
repLog.atFine().log("Skipping replication of project %s", project.get());
@@ -662,6 +666,16 @@
return config.replicateProjectDeletions();
}
+ private boolean matches(URIish uri, Project.NameKey project) {
+ for (URIish configUri : config.getRemoteConfig().getURIs()) {
+ URIish projectUri = getURI(configUri, project);
+ if (uri.equals(projectUri)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
List<URIish> getURIs(Project.NameKey project, String urlMatch) {
List<URIish> r = Lists.newArrayListWithCapacity(config.getRemoteConfig().getURIs().size());
for (URIish configUri : config.getRemoteConfig().getURIs()) {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationsCollection.java b/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationsCollection.java
index ecfbb8e..ac14657 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationsCollection.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationsCollection.java
@@ -38,6 +38,7 @@
import com.googlesource.gerrit.plugins.replication.Destination.Factory;
import com.googlesource.gerrit.plugins.replication.ReplicationConfig.FilterType;
import java.net.URISyntaxException;
+import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
@@ -157,6 +158,17 @@
}
@Override
+ public List<Destination> getDestinations(URIish uri, Project.NameKey project, String ref) {
+ List<Destination> dests = new ArrayList<>();
+ for (Destination dest : getAll(FilterType.ALL)) {
+ if (dest.wouldPush(uri, project, ref)) {
+ dests.add(dest);
+ }
+ }
+ return dests;
+ }
+
+ @Override
public boolean isEmpty() {
return destinations.isEmpty();
}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationDestinations.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationDestinations.java
index b191d7d..18ccc66 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationDestinations.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationDestinations.java
@@ -44,6 +44,16 @@
*/
List<Destination> getAll(FilterType filterType);
+ /**
+ * Return the active replication destinations for a uri/project/ref triplet.
+ *
+ * @param uriish uri of the destinations
+ * @param project name of the project
+ * @param ref ref name
+ * @return the list of active destinations
+ */
+ List<Destination> getDestinations(URIish uriish, Project.NameKey project, String ref);
+
/** @return true if there are no destinations, false otherwise. */
boolean isEmpty();
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
index 21e4227..e685237 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
@@ -34,6 +34,7 @@
import com.googlesource.gerrit.plugins.replication.PushResultProcessing.GitUpdateProcessing;
import com.googlesource.gerrit.plugins.replication.ReplicationConfig.FilterType;
import com.googlesource.gerrit.plugins.replication.ReplicationTasksStorage.ReplicateRefUpdate;
+import java.net.URISyntaxException;
import java.util.HashSet;
import java.util.Optional;
import java.util.Queue;
@@ -154,6 +155,14 @@
}
}
+ private void fire(URIish uri, Project.NameKey project, String refName) {
+ ReplicationState state = new ReplicationState(new GitUpdateProcessing(dispatcher.get()));
+ for (Destination dest : destinations.get().getDestinations(uri, project, refName)) {
+ dest.schedule(project, refName, uri, state);
+ }
+ state.markAllPushTasksScheduled();
+ }
+
@UsedAt(UsedAt.Project.COLLABNET)
public void pushReference(Destination cfg, Project.NameKey project, String refName) {
pushReference(cfg, project, null, refName, null, true, false);
@@ -190,14 +199,16 @@
private void firePendingEvents() {
replaying = true;
try {
- Set<String> eventsReplayed = new HashSet<>();
replaying = true;
for (ReplicationTasksStorage.ReplicateRefUpdate t : replicationTasksStorage.listWaiting()) {
- String eventKey = String.format("%s:%s", t.project, t.ref);
- if (!eventsReplayed.contains(eventKey)) {
- repLog.atInfo().log("Firing pending task %s", eventKey);
- fire(t.project, t.ref, true);
- eventsReplayed.add(eventKey);
+ if (t == null) {
+ repLog.atWarning().log("Encountered null replication event in ReplicationTasksStorage");
+ continue;
+ }
+ try {
+ fire(new URIish(t.uri), Project.nameKey(t.project), t.ref);
+ } catch (URISyntaxException e) {
+ repLog.atSevere().withCause(e).log("Encountered malformed URI for persisted event %s", t);
}
}
} finally {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java
index 506b175..bfd0886 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java
@@ -112,6 +112,11 @@
this.disableDeleteForTesting = deleteDisabled;
}
+ @VisibleForTesting
+ public void delete(ReplicateRefUpdate r) {
+ new Task(r).delete();
+ }
+
public synchronized void start(PushOne push) {
for (String ref : push.getRefs()) {
new Task(new ReplicateRefUpdate(push, ref)).start();
@@ -264,6 +269,15 @@
}
}
+ public void delete() {
+ try {
+ Files.deleteIfExists(waiting);
+ Files.deleteIfExists(running);
+ } catch (IOException e) {
+ logger.atSevere().withCause(e).log("Error while deleting task %s", taskKey);
+ }
+ }
+
private void rename(Path from, Path to) {
try {
logger.atFine().log("RENAME %s to %s %s", from, to, updateLog());
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java
index 653e5a1..08b8c65 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java
@@ -48,6 +48,7 @@
import java.util.Optional;
import java.util.function.Supplier;
import java.util.regex.Pattern;
+import java.util.stream.Stream;
import org.eclipse.jgit.lib.Constants;
import org.eclipse.jgit.lib.Ref;
import org.eclipse.jgit.lib.Repository;
@@ -513,6 +514,46 @@
return pushOne == null ? false : pushOne.isRetrying();
}
+ @Test
+ public void shouldFirePendingOnlyToStoredUri() throws Exception {
+ String suffix1 = "replica1";
+ String suffix2 = "replica2";
+ Project.NameKey target1 = createTestProject(project + suffix1);
+ Project.NameKey target2 = createTestProject(project + suffix2);
+ String remote1 = "foo1";
+ String remote2 = "foo2";
+ setReplicationDestination(remote1, suffix1, ALL_PROJECTS, Integer.MAX_VALUE, false);
+ setReplicationDestination(remote2, suffix2, ALL_PROJECTS, Integer.MAX_VALUE, false);
+ reloadConfig();
+
+ String changeRef = createChange().getPatchSet().refName();
+
+ tasksStorage.disableDeleteForTesting(false);
+ changeReplicationTasksForRemote(changeRef, remote1).forEach(tasksStorage::delete);
+ tasksStorage.disableDeleteForTesting(true);
+
+ setReplicationDestination(remote1, suffix1, ALL_PROJECTS);
+ setReplicationDestination(remote2, suffix2, ALL_PROJECTS);
+ reloadConfig();
+
+ assertThat(changeReplicationTasksForRemote(changeRef, remote2).count()).isEqualTo(1);
+ assertThat(changeReplicationTasksForRemote(changeRef, remote1).count()).isEqualTo(0);
+
+ assertThat(isPushCompleted(target2, changeRef, TEST_TIMEOUT)).isEqualTo(true);
+ assertThat(isPushCompleted(target1, changeRef, TEST_TIMEOUT)).isEqualTo(false);
+ }
+
+ public boolean isPushCompleted(Project.NameKey project, String ref, Duration timeOut) {
+ try (Repository repo = repoManager.openRepository(project)) {
+ WaitUtil.waitUntil(() -> checkedGetRef(repo, ref) != null, timeOut);
+ return true;
+ } catch (InterruptedException e) {
+ return false;
+ } catch (Exception e) {
+ throw new RuntimeException("Cannot open repo for project" + project, e);
+ }
+ }
+
private Ref getRef(Repository repo, String branchName) throws IOException {
return repo.getRefDatabase().exactRef(branchName);
}
@@ -528,30 +569,47 @@
private void setReplicationDestination(
String remoteName, String replicaSuffix, Optional<String> project) throws IOException {
- setReplicationDestination(remoteName, Arrays.asList(replicaSuffix), project, false);
+ setReplicationDestination(
+ remoteName, Arrays.asList(replicaSuffix), project, TEST_REPLICATION_DELAY, false);
}
private void setReplicationDestination(
String remoteName, String replicaSuffix, Optional<String> project, boolean mirror)
throws IOException {
- setReplicationDestination(remoteName, Arrays.asList(replicaSuffix), project, mirror);
+ setReplicationDestination(
+ remoteName, Arrays.asList(replicaSuffix), project, TEST_REPLICATION_DELAY, mirror);
+ }
+
+ private void setReplicationDestination(
+ String remoteName,
+ String replicaSuffix,
+ Optional<String> project,
+ int replicationDelay,
+ boolean mirror)
+ throws IOException {
+ setReplicationDestination(
+ remoteName, Arrays.asList(replicaSuffix), project, replicationDelay, mirror);
}
private void setReplicationDestination(
String remoteName, List<String> replicaSuffixes, Optional<String> project)
throws IOException {
- setReplicationDestination(remoteName, replicaSuffixes, project, false);
+ setReplicationDestination(remoteName, replicaSuffixes, project, TEST_REPLICATION_DELAY, false);
}
private void setReplicationDestination(
- String remoteName, List<String> replicaSuffixes, Optional<String> project, boolean mirror)
+ String remoteName,
+ List<String> replicaSuffixes,
+ Optional<String> project,
+ int replicationDelay,
+ boolean mirror)
throws IOException {
List<String> replicaUrls =
replicaSuffixes.stream()
.map(suffix -> gitPath.resolve("${name}" + suffix + ".git").toString())
.collect(toList());
config.setStringList("remote", remoteName, "url", replicaUrls);
- config.setInt("remote", remoteName, "replicationDelay", TEST_REPLICATION_DELAY);
+ config.setInt("remote", remoteName, "replicationDelay", replicationDelay);
config.setInt("remote", remoteName, "replicationRetry", TEST_REPLICATION_RETRY);
config.setBoolean("remote", remoteName, "mirror", mirror);
project.ifPresent(prj -> config.setString("remote", remoteName, "projects", prj));
@@ -597,6 +655,13 @@
return plugin.getSysInjector().getInstance(classObj);
}
+ private Stream<ReplicateRefUpdate> changeReplicationTasksForRemote(
+ String changeRef, String remote) {
+ return tasksStorage.list().stream()
+ .filter(task -> changeRef.equals(task.ref))
+ .filter(task -> remote.equals(task.remote));
+ }
+
private Project.NameKey createTestProject(String name) throws Exception {
return projectOperations.newProject().name(name).create();
}
