Merge branch 'stable-3.0'
* stable-3.0:
Clean up ReplicationQueue tests
Convert ReferenceUpdateEvent class to AutoValue
Fix issue with dropping events on start
Change-Id: I5a4bc84a829600865fdd551152219cd1bbe5aa91
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 5bc7320..3e17167 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
@@ -14,7 +14,9 @@
package com.googlesource.gerrit.plugins.replication;
+import com.google.auto.value.AutoValue;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Queues;
import com.google.gerrit.extensions.events.GitReferenceUpdatedListener;
import com.google.gerrit.extensions.events.HeadUpdatedListener;
import com.google.gerrit.extensions.events.LifecycleListener;
@@ -30,6 +32,7 @@
import com.googlesource.gerrit.plugins.replication.ReplicationTasksStorage.ReplicateRefUpdate;
import java.util.HashSet;
import java.util.Optional;
+import java.util.Queue;
import java.util.Set;
import org.eclipse.jgit.transport.URIish;
import org.slf4j.Logger;
@@ -52,6 +55,7 @@
private final ReplicationTasksStorage replicationTasksStorage;
private volatile boolean running;
private volatile boolean replaying;
+ private final Queue<ReferenceUpdatedEvent> beforeStartupEventsQueue;
@Inject
ReplicationQueue(
@@ -65,6 +69,7 @@
destinations = rd;
stateLog = sl;
replicationTasksStorage = rts;
+ beforeStartupEventsQueue = Queues.newConcurrentLinkedQueue();
}
@Override
@@ -73,6 +78,7 @@
destinations.get().startup(workQueue);
running = true;
firePendingEvents();
+ fireBeforeStartupEvents();
}
}
@@ -125,7 +131,10 @@
private void onGitReferenceUpdated(String projectName, String refName) {
ReplicationState state = new ReplicationState(new GitUpdateProcessing(dispatcher.get()));
if (!running) {
- stateLog.warn("Replication plugin did not finish startup before event", state);
+ stateLog.warn(
+ "Replication plugin did not finish startup before event, event replication is postponed",
+ state);
+ beforeStartupEventsQueue.add(ReferenceUpdatedEvent.create(projectName, refName));
return;
}
@@ -173,4 +182,28 @@
destinations.get().getURIs(Optional.empty(), p, FilterType.ALL).entries().stream()
.forEach(e -> e.getKey().scheduleUpdateHead(e.getValue(), p, event.getNewHeadName()));
}
+
+ private void fireBeforeStartupEvents() {
+ Set<String> eventsReplayed = new HashSet<>();
+ for (ReferenceUpdatedEvent event : beforeStartupEventsQueue) {
+ String eventKey = String.format("%s:%s", event.projectName(), event.refName());
+ if (!eventsReplayed.contains(eventKey)) {
+ repLog.info("Firing pending task {}", event);
+ onGitReferenceUpdated(event.projectName(), event.refName());
+ eventsReplayed.add(eventKey);
+ }
+ }
+ }
+
+ @AutoValue
+ abstract static class ReferenceUpdatedEvent {
+
+ static ReferenceUpdatedEvent create(String projectName, String refName) {
+ return new AutoValue_ReplicationQueue_ReferenceUpdatedEvent(projectName, refName);
+ }
+
+ public abstract String projectName();
+
+ public abstract String refName();
+ }
}
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 d39cf79..eed5158 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java
@@ -306,8 +306,26 @@
}
}
- private Project.NameKey createTestProject(String name) throws Exception {
- return projectOperations.newProject().name(name).create();
+ @Test
+ public void shouldNotDropEventsWhenStarting() throws Exception {
+ Project.NameKey targetProject = createTestProject(project + "replica");
+
+ setReplicationDestination("foo", "replica", ALL_PROJECTS);
+ reloadConfig();
+
+ replicationQueueStop();
+ Result pushResult = createChange();
+ replicationQueueStart();
+
+ RevCommit sourceCommit = pushResult.getCommit();
+ String sourceRef = pushResult.getPatchSet().refName();
+
+ try (Repository repo = repoManager.openRepository(targetProject)) {
+ waitUntil(() -> checkedGetRef(repo, sourceRef) != null);
+ Ref targetBranchRef = getRef(repo, sourceRef);
+ assertThat(targetBranchRef).isNotNull();
+ assertThat(targetBranchRef.getObjectId()).isEqualTo(sourceCommit.getId());
+ }
}
private Ref getRef(Repository repo, String branchName) throws IOException {
@@ -348,11 +366,35 @@
}
private void reloadConfig() {
- plugin.getSysInjector().getInstance(AutoReloadConfigDecorator.class).reload();
+ getAutoReloadConfigDecoratorInstance().reload();
}
private void shutdownDestinations() {
- plugin.getSysInjector().getInstance(DestinationsCollection.class).shutdown();
+ getInstance(DestinationsCollection.class).shutdown();
+ }
+
+ private void replicationQueueStart() {
+ getReplicationQueueInstance().start();
+ }
+
+ private void replicationQueueStop() {
+ getReplicationQueueInstance().stop();
+ }
+
+ private AutoReloadConfigDecorator getAutoReloadConfigDecoratorInstance() {
+ return getInstance(AutoReloadConfigDecorator.class);
+ }
+
+ private ReplicationQueue getReplicationQueueInstance() {
+ return getInstance(ReplicationQueue.class);
+ }
+
+ private <T> T getInstance(Class<T> classObj) {
+ return plugin.getSysInjector().getInstance(classObj);
+ }
+
+ private Project.NameKey createTestProject(String name) throws Exception {
+ return projectOperations.newProject().name(name).create();
}
private List<ReplicateRefUpdate> listReplicationTasks(String refRegex) {
