Merge branch 'stable-3.5' into stable-3.8
* stable-3.5:
TasksStorage: Remove synchronized from methods
Place the replaying flag clearing in a finally
Demote delete errors when distributor is enabled
distributor: Reduce log level for no-op consolidations
Change-Id: Iaa6d4a59d5cb26879500f2d36852eaa30efc5d7b
Release-Notes: skip
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ChainedScheduler.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ChainedScheduler.java
index 89b97e9..c61a123 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ChainedScheduler.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ChainedScheduler.java
@@ -85,7 +85,7 @@
super(
threadPool,
stream.iterator(),
- new ForwardingRunner<T>(runner) {
+ new ForwardingRunner<>(runner) {
@Override
public void onDone() {
stream.close();
@@ -109,7 +109,7 @@
try {
runner.run(item);
} catch (RuntimeException e) { // catch to prevent chain from breaking
- logger.atSevere().withCause(e).log("Error while running: " + item);
+ logger.atSevere().withCause(e).log("Error while running: %s", item);
}
if (!scheduledNext) {
runner.onDone();
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 9a53aa3..1f01e2a 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
@@ -27,6 +27,7 @@
import com.google.common.collect.Lists;
import com.google.common.io.Files;
import com.google.common.net.UrlEscapers;
+import com.google.gerrit.common.Nullable;
import com.google.gerrit.entities.AccountGroup;
import com.google.gerrit.entities.BranchNameKey;
import com.google.gerrit.entities.GroupReference;
@@ -59,7 +60,6 @@
import com.google.inject.Injector;
import com.google.inject.Provider;
import com.google.inject.Provides;
-import com.google.inject.Scopes;
import com.google.inject.assistedinject.Assisted;
import com.google.inject.assistedinject.FactoryModuleBuilder;
import com.google.inject.servlet.RequestScoped;
@@ -70,7 +70,9 @@
import com.googlesource.gerrit.plugins.replication.events.ReplicationScheduledEvent;
import java.io.IOException;
import java.net.URISyntaxException;
+import java.util.Collection;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -81,7 +83,7 @@
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
-import org.apache.http.impl.client.CloseableHttpClient;
+import java.util.stream.Collectors;
import org.eclipse.jgit.lib.Constants;
import org.eclipse.jgit.lib.Ref;
import org.eclipse.jgit.lib.Repository;
@@ -193,9 +195,6 @@
.to(AdminApiFactory.DefaultAdminApiFactory.class);
install(new FactoryModuleBuilder().build(GerritRestApi.Factory.class));
- bind(CloseableHttpClient.class)
- .toProvider(HttpClientProvider.class)
- .in(Scopes.SINGLETON);
}
@Provides
@@ -276,6 +275,8 @@
}
private void foreachPushOp(Map<URIish, PushOne> opsMap, Function<PushOne, Void> pushOneFunction) {
+ // Callers may modify the provided opsMap concurrently, hence make a defensive copy of the
+ // values to loop over them.
for (PushOne pushOne : ImmutableList.copyOf(opsMap.values())) {
pushOneFunction.apply(pushOne);
}
@@ -387,32 +388,36 @@
return false;
}
- void schedule(Project.NameKey project, String ref, URIish uri, ReplicationState state) {
- schedule(project, ref, uri, state, false);
+ void schedule(Project.NameKey project, Set<String> refs, URIish uri, ReplicationState state) {
+ schedule(project, refs, uri, state, false);
}
void scheduleFromStorage(
- Project.NameKey project, String ref, URIish uri, ReplicationState state) {
- schedule(project, ref, uri, state, false, true);
+ Project.NameKey project, Set<String> refs, URIish uri, ReplicationState state) {
+ schedule(project, refs, uri, state, false, true);
}
void schedule(
- Project.NameKey project, String ref, URIish uri, ReplicationState state, boolean now) {
- schedule(project, ref, uri, state, now, false);
+ Project.NameKey project, Set<String> refs, URIish uri, ReplicationState state, boolean now) {
+ schedule(project, refs, uri, state, now, false);
}
void schedule(
Project.NameKey project,
- String ref,
+ Set<String> refs,
URIish uri,
ReplicationState state,
boolean now,
boolean fromStorage) {
- if (!shouldReplicate(project, ref, state)) {
- repLog.atFine().log("Not scheduling replication %s:%s => %s", project, ref, uri);
- return;
+ Set<String> refsToSchedule = new HashSet<>();
+ for (String ref : refs) {
+ if (!shouldReplicate(project, ref, state)) {
+ repLog.atFine().log("Not scheduling replication %s:%s => %s", project, ref, uri);
+ continue;
+ }
+ refsToSchedule.add(ref);
}
- repLog.atInfo().log("scheduling replication %s:%s => %s", project, ref, uri);
+ repLog.atInfo().log("scheduling replication %s:%s => %s", project, refs, uri);
if (!config.replicatePermissions()) {
PushOne e;
@@ -443,29 +448,32 @@
PushOne task = getPendingPush(uri);
if (task == null) {
task = opFactory.create(project, uri);
- addRef(task, ref);
- task.addState(ref, state);
+ addRefs(task, ImmutableSet.copyOf(refsToSchedule));
+ task.addState(refsToSchedule, state);
@SuppressWarnings("unused")
ScheduledFuture<?> ignored =
pool.schedule(task, now ? 0 : config.getDelay(), TimeUnit.SECONDS);
pending.put(uri, task);
repLog.atInfo().log(
"scheduled %s:%s => %s to run %s",
- project, ref, task, now ? "now" : "after " + config.getDelay() + "s");
+ project, refsToSchedule, task, now ? "now" : "after " + config.getDelay() + "s");
} else {
- boolean added = addRef(task, ref);
- task.addState(ref, state);
+ boolean added = addRefs(task, ImmutableSet.copyOf(refsToSchedule));
+ task.addState(refsToSchedule, state);
String message = "consolidated %s:%s => %s with an existing pending push";
if (added || !fromStorage) {
- repLog.atInfo().log(message, project, ref, task);
+ repLog.atInfo().log(message, project, refsToSchedule, task);
} else {
- repLog.atFine().log(message, project, ref, task);
+ repLog.atFine().log(message, project, refsToSchedule, task);
}
}
- state.increasePushTaskCount(project.get(), ref);
+ for (String ref : refsToSchedule) {
+ state.increasePushTaskCount(project.get(), ref);
+ }
}
}
+ @Nullable
private PushOne getPendingPush(URIish uri) {
PushOne e = pending.get(uri);
if (e != null && !e.wasCanceled()) {
@@ -496,9 +504,9 @@
pool.schedule(updateHeadFactory.create(uri, project, newHead), 0, TimeUnit.SECONDS);
}
- private boolean addRef(PushOne e, String ref) {
- boolean added = e.addRef(ref);
- postReplicationScheduledEvent(e, ref);
+ private boolean addRefs(PushOne e, ImmutableSet<String> refs) {
+ boolean added = e.addRefBatch(refs);
+ postReplicationScheduledEvent(e, refs);
return added;
}
@@ -543,7 +551,7 @@
// second one fails, it will also be rescheduled and then,
// here, find out replication to its URI is already pending
// for retry (blocking).
- pendingPushOp.addRefs(pushOp.getRefs());
+ pendingPushOp.addRefBatches(pushOp.getRefs());
pendingPushOp.addStates(pushOp.getStates());
pushOp.removeStates();
@@ -562,7 +570,7 @@
pendingPushOp.canceledByReplication();
pending.remove(uri);
- pushOp.addRefs(pendingPushOp.getRefs());
+ pushOp.addRefBatches(pendingPushOp.getRefs());
pushOp.addStates(pendingPushOp.getStates());
pendingPushOp.removeStates();
}
@@ -656,7 +664,7 @@
return true;
}
- boolean matches = (new ReplicationFilter(projects)).matches(project);
+ boolean matches = new ReplicationFilter(projects).matches(project);
if (!matches) {
repLog.atFine().log(
"Skipping replication of project %s; does not match filter", project.get());
@@ -814,6 +822,10 @@
return config.getPushBatchSize();
}
+ boolean replicateNoteDbMetaRefs() {
+ return config.replicateNoteDbMetaRefs();
+ }
+
private static boolean matches(URIish uri, String urlMatch) {
if (urlMatch == null || urlMatch.equals("") || urlMatch.equals("*")) {
return true;
@@ -825,10 +837,10 @@
postReplicationScheduledEvent(pushOp, null);
}
- private void postReplicationScheduledEvent(PushOne pushOp, String inputRef) {
- Set<String> refs = inputRef == null ? pushOp.getRefs() : ImmutableSet.of(inputRef);
+ private void postReplicationScheduledEvent(PushOne pushOp, ImmutableSet<String> inputRefs) {
+ Set<ImmutableSet<String>> refBatches = inputRefs == null ? pushOp.getRefs() : Set.of(inputRefs);
Project.NameKey project = pushOp.getProjectNameKey();
- for (String ref : refs) {
+ for (String ref : flattenSetOfRefBatches(refBatches)) {
ReplicationScheduledEvent event =
new ReplicationScheduledEvent(project.get(), ref, pushOp.getURI());
try {
@@ -841,7 +853,7 @@
private void postReplicationFailedEvent(PushOne pushOp, RemoteRefUpdate.Status status) {
Project.NameKey project = pushOp.getProjectNameKey();
- for (String ref : pushOp.getRefs()) {
+ for (String ref : flattenSetOfRefBatches(pushOp.getRefs())) {
RefReplicatedEvent event =
new RefReplicatedEvent(project.get(), ref, pushOp.getURI(), RefPushResult.FAILED, status);
try {
@@ -851,4 +863,8 @@
}
}
}
+
+ private Set<String> flattenSetOfRefBatches(Set<ImmutableSet<String>> refBatches) {
+ return refBatches.stream().flatMap(Collection::stream).collect(Collectors.toSet());
+ }
}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationConfiguration.java b/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationConfiguration.java
index a74d198..4d72ff0 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationConfiguration.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationConfiguration.java
@@ -39,6 +39,7 @@
private final ImmutableList<String> adminUrls;
private final int poolThreads;
private final boolean createMissingRepos;
+ private final boolean replicateNoteDbMetaRefs;
private final boolean replicatePermissions;
private final boolean replicateProjectDeletions;
private final boolean replicateHiddenProjects;
@@ -71,6 +72,7 @@
"updateRefErrorMaxRetries",
cfg.getInt("replication", "lockErrorMaxRetries", 0));
createMissingRepos = cfg.getBoolean("remote", name, "createMissingRepositories", true);
+ replicateNoteDbMetaRefs = cfg.getBoolean("remote", name, "replicateNoteDbMetaRefs", true);
replicatePermissions = cfg.getBoolean("remote", name, "replicatePermissions", true);
replicateProjectDeletions = cfg.getBoolean("remote", name, "replicateProjectDeletions", false);
replicateHiddenProjects = cfg.getBoolean("remote", name, "replicateHiddenProjects", false);
@@ -177,6 +179,11 @@
return createMissingRepos;
}
+ @Override
+ public boolean replicateNoteDbMetaRefs() {
+ return replicateNoteDbMetaRefs;
+ }
+
public boolean replicateProjectDeletions() {
return replicateProjectDeletions;
}
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 4957a64..79a0683 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationsCollection.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationsCollection.java
@@ -41,6 +41,7 @@
import java.util.List;
import java.util.Objects;
import java.util.Optional;
+import java.util.Set;
import java.util.function.Predicate;
import org.eclipse.jgit.errors.ConfigInvalidException;
import org.eclipse.jgit.transport.URIish;
@@ -157,11 +158,13 @@
}
@Override
- public List<Destination> getDestinations(URIish uri, Project.NameKey project, String ref) {
+ public List<Destination> getDestinations(URIish uri, Project.NameKey project, Set<String> refs) {
List<Destination> dests = new ArrayList<>();
for (Destination dest : getAll(FilterType.ALL)) {
- if (dest.wouldPush(uri, project, ref)) {
- dests.add(dest);
+ for (String ref : refs) {
+ if (dest.wouldPush(uri, project, ref)) {
+ dests.add(dest);
+ }
}
}
return dests;
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/GerritRestApi.java b/src/main/java/com/googlesource/gerrit/plugins/replication/GerritRestApi.java
index fa81dd0..5081e2b 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/GerritRestApi.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/GerritRestApi.java
@@ -18,6 +18,7 @@
import static com.googlesource.gerrit.plugins.replication.ReplicationQueue.repLog;
import static java.nio.charset.StandardCharsets.UTF_8;
+import com.google.gerrit.common.Nullable;
import com.google.gerrit.entities.Project;
import com.google.gerrit.extensions.restapi.Url;
import com.google.inject.Inject;
@@ -109,6 +110,7 @@
return ctx;
}
+ @Nullable
private CredentialsProvider adapt(org.eclipse.jgit.transport.CredentialsProvider cp) {
CredentialItem.Username user = new CredentialItem.Username();
CredentialItem.Password pass = new CredentialItem.Password();
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 c21b837..f9947ae 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/PushOne.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/PushOne.java
@@ -22,15 +22,18 @@
import static java.util.stream.Collectors.toList;
import static java.util.stream.Collectors.toMap;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import com.google.common.base.Throwables;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.LinkedListMultimap;
import com.google.common.collect.ListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
+import com.google.gerrit.common.Nullable;
import com.google.gerrit.entities.Project;
import com.google.gerrit.entities.RefNames;
-import com.google.gerrit.extensions.events.GitReferenceUpdatedListener;
+import com.google.gerrit.extensions.events.GitBatchRefUpdateListener;
import com.google.gerrit.extensions.registration.DynamicItem;
import com.google.gerrit.extensions.restapi.AuthException;
import com.google.gerrit.extensions.restapi.ResourceConflictException;
@@ -51,7 +54,6 @@
import com.google.inject.Inject;
import com.google.inject.assistedinject.Assisted;
import com.googlesource.gerrit.plugins.replication.ReplicationState.RefPushResult;
-import com.jcraft.jsch.JSchException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -85,7 +87,7 @@
import org.eclipse.jgit.transport.URIish;
/**
- * A push to remote operation started by {@link GitReferenceUpdatedListener}.
+ * A push to remote operation started by {@link GitBatchRefUpdateListener}.
*
* <p>Instance members are protected by the lock within PushQueue. Callers must take that lock to
* ensure they are working with a current view of the object.
@@ -120,7 +122,7 @@
private final Project.NameKey projectName;
private final URIish uri;
- private final Set<String> delta = Sets.newHashSetWithExpectedSize(4);
+ private final Set<ImmutableSet<String>> refBatchesToPush = Sets.newHashSetWithExpectedSize(4);
private boolean pushAllRefs;
private Repository git;
private boolean isCollision;
@@ -241,12 +243,16 @@
* config.
*/
protected String getLimitedRefs() {
- Set<String> refs = getRefs();
+ Set<ImmutableSet<String>> refs = getRefs();
int maxRefsToShow = replConfig.getMaxRefsToShow();
if (maxRefsToShow == 0) {
maxRefsToShow = refs.size();
}
- String refsString = refs.stream().limit(maxRefsToShow).collect(Collectors.joining(" "));
+ String refsString =
+ refs.stream()
+ .flatMap(Collection::stream)
+ .limit(maxRefsToShow)
+ .collect(Collectors.joining(" "));
int hiddenRefs = refs.size() - maxRefsToShow;
if (hiddenRefs > 0) {
refsString += " (+" + hiddenRefs + ")";
@@ -281,59 +287,63 @@
return uri;
}
- /** Returns true if the ref was not already included in the push and false otherwise */
- boolean addRef(String ref) {
- if (ALL_REFS.equals(ref)) {
- delta.clear();
+ /** Returns false if all refs were already included in the push, true otherwise */
+ boolean addRefBatch(ImmutableSet<String> refBatch) {
+ if (refBatch.size() == 1 && refBatch.contains(ALL_REFS)) {
+ refBatchesToPush.clear();
boolean pushAllRefsChanged = !pushAllRefs;
pushAllRefs = true;
repLog.atFinest().log("Added all refs for replication to %s", uri);
return pushAllRefsChanged;
}
- if (!pushAllRefs && delta.add(ref)) {
- repLog.atFinest().log("Added ref %s for replication to %s", ref, uri);
+ if (!pushAllRefs && refBatchesToPush.add(refBatch)) {
+ repLog.atFinest().log("Added ref %s for replication to %s", refBatch, uri);
return true;
}
return false;
}
@Override
- public Set<String> getRefs() {
- return pushAllRefs ? Sets.newHashSet(ALL_REFS) : delta;
+ public Set<ImmutableSet<String>> getRefs() {
+ return pushAllRefs ? Set.of(ImmutableSet.of(ALL_REFS)) : refBatchesToPush;
}
- void addRefs(Set<String> refs) {
+ void addRefBatches(Set<ImmutableSet<String>> refBatches) {
if (!pushAllRefs) {
- for (String ref : refs) {
- addRef(ref);
+ for (ImmutableSet<String> refBatch : refBatches) {
+ addRefBatch(refBatch);
}
}
}
- Set<String> setStartedRefs(Set<String> startedRefs) {
- Set<String> notAttemptedRefs = Sets.difference(delta, startedRefs);
+ Set<ImmutableSet<String>> setStartedRefs(Set<ImmutableSet<String>> startedRefs) {
+ Set<ImmutableSet<String>> notAttemptedRefs = Sets.difference(refBatchesToPush, startedRefs);
pushAllRefs = false;
- delta.clear();
- addRefs(startedRefs);
+ refBatchesToPush.clear();
+ addRefBatches(startedRefs);
return notAttemptedRefs;
}
- void notifyNotAttempted(Set<String> notAttemptedRefs) {
- notAttemptedRefs.forEach(
- ref ->
- Arrays.asList(getStatesByRef(ref))
- .forEach(
- state ->
- state.notifyRefReplicated(
- projectName.get(),
- ref,
- uri,
- RefPushResult.NOT_ATTEMPTED,
- RemoteRefUpdate.Status.UP_TO_DATE)));
+ void notifyNotAttempted(Set<ImmutableSet<String>> notAttemptedRefs) {
+ notAttemptedRefs.stream()
+ .flatMap(Collection::stream)
+ .forEach(
+ ref ->
+ Arrays.asList(getStatesByRef(ref))
+ .forEach(
+ state ->
+ state.notifyRefReplicated(
+ projectName.get(),
+ ref,
+ uri,
+ RefPushResult.NOT_ATTEMPTED,
+ RemoteRefUpdate.Status.UP_TO_DATE)));
}
- void addState(String ref, ReplicationState state) {
- stateMap.put(ref, state);
+ void addState(Set<String> refs, ReplicationState state) {
+ for (String ref : refs) {
+ stateMap.put(ref, state);
+ }
}
ListMultimap<String, ReplicationState> getStates() {
@@ -465,10 +475,7 @@
} catch (NotSupportedException e) {
stateLog.error("Cannot replicate to " + uri, e, getStatesAsArray());
} catch (TransportException e) {
- Throwable cause = e.getCause();
- if (cause instanceof JSchException && cause.getMessage().startsWith("UnknownHostKey:")) {
- repLog.atSevere().log("Cannot replicate to %s: %s", uri, cause.getMessage());
- } else if (e instanceof UpdateRefFailureException) {
+ if (e instanceof UpdateRefFailureException) {
updateRefRetryCount++;
repLog.atSevere().log("Cannot replicate to %s due to a lock or write ref failure", uri);
@@ -659,7 +666,7 @@
// to only the references we will update during this operation.
//
Map<String, Ref> n = new HashMap<>();
- for (String src : delta) {
+ for (String src : flattenRefBatchesToPush()) {
Ref r = local.get(src);
if (r != null) {
n.put(src, r);
@@ -721,7 +728,8 @@
private List<RemoteRefUpdate> doPushDelta(Map<String, Ref> local) throws IOException {
List<RemoteRefUpdate> cmds = new ArrayList<>();
boolean noPerms = !pool.isReplicatePermissions();
- for (String src : delta) {
+ Set<String> refs = flattenRefBatchesToPush();
+ for (String src : refs) {
RefSpec spec = matchSrc(src);
if (spec != null) {
// If the ref still exists locally, send it, otherwise delete it.
@@ -744,7 +752,8 @@
private boolean canPushRef(String ref, boolean noPerms) {
return !(noPerms && RefNames.REFS_CONFIG.equals(ref))
- && !ref.startsWith(RefNames.REFS_CACHE_AUTOMERGE);
+ && !ref.startsWith(RefNames.REFS_CACHE_AUTOMERGE)
+ && !(!pool.replicateNoteDbMetaRefs() && RefNames.isNoteDbMetaRef(ref));
}
private Map<String, Ref> listRemote(Transport tn)
@@ -754,6 +763,7 @@
}
}
+ @Nullable
private RefSpec matchSrc(String ref) {
for (RefSpec s : config.getPushRefSpecs()) {
if (s.matchSource(ref)) {
@@ -763,6 +773,7 @@
return null;
}
+ @Nullable
private RefSpec matchDst(String ref) {
for (RefSpec s : config.getPushRefSpecs()) {
if (s.matchDestination(ref)) {
@@ -772,7 +783,8 @@
return null;
}
- private void push(List<RemoteRefUpdate> cmds, RefSpec spec, Ref src) throws IOException {
+ @VisibleForTesting
+ void push(List<RemoteRefUpdate> cmds, RefSpec spec, Ref src) throws IOException {
String dst = spec.getDestination();
boolean force = spec.isForceUpdate();
cmds.add(new RemoteRefUpdate(git, src, dst, force, null, null));
@@ -870,6 +882,10 @@
stateMap.clear();
}
+ private Set<String> flattenRefBatchesToPush() {
+ return refBatchesToPush.stream().flatMap(Collection::stream).collect(Collectors.toSet());
+ }
+
public static class UpdateRefFailureException extends TransportException {
private static final long serialVersionUID = 1L;
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/PushResultProcessing.java b/src/main/java/com/googlesource/gerrit/plugins/replication/PushResultProcessing.java
index 4f33937..6ed47d4 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/PushResultProcessing.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/PushResultProcessing.java
@@ -30,7 +30,15 @@
public interface PushResultProcessing {
public static final PushResultProcessing NO_OP = new PushResultProcessing() {};
- /** Invoked when a ref has been replicated to one node. */
+ /**
+ * Invoked when a ref has been replicated to one node.
+ *
+ * @param project the project name
+ * @param ref the ref name
+ * @param uri the URI
+ * @param status the status of the push
+ * @param refStatus the status for the ref
+ */
default void onRefReplicatedToOneNode(
String project,
String ref,
@@ -38,10 +46,20 @@
RefPushResult status,
RemoteRefUpdate.Status refStatus) {}
- /** Invoked when a ref has been replicated to all nodes */
+ /**
+ * Invoked when refs have been replicated to all nodes.
+ *
+ * @param project the project name
+ * @param ref the ref name
+ * @param nodesCount the number of nodes
+ */
default void onRefReplicatedToAllNodes(String project, String ref, int nodesCount) {}
- /** Invoked when all refs have been replicated to all nodes */
+ /**
+ * Invoked when all refs have been replicated to all nodes.
+ *
+ * @param totalPushTasksCount total number of push tasks
+ */
default void onAllRefsReplicatedToAllNodes(int totalPushTasksCount) {}
/**
@@ -90,7 +108,7 @@
StringBuilder sb = new StringBuilder();
sb.append("Replicate ");
sb.append(project);
- sb.append(" ref ");
+ sb.append(" refs ");
sb.append(ref);
sb.append(" to ");
sb.append(resolveNodeName(uri));
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/RemoteConfiguration.java b/src/main/java/com/googlesource/gerrit/plugins/replication/RemoteConfiguration.java
index 5fe0323..05b4066 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/RemoteConfiguration.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/RemoteConfiguration.java
@@ -105,6 +105,13 @@
int getPushBatchSize();
/**
+ * Whether to replicate the NoteDb meta refs or not.
+ *
+ * @return boolean, true by default
+ */
+ boolean replicateNoteDbMetaRefs();
+
+ /**
* Whether the remote configuration is for a single project only
*
* @return true, when configuration is for a single project, false otherwise
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 2fa6c34..78c1a35 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationDestinations.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationDestinations.java
@@ -20,6 +20,7 @@
import com.googlesource.gerrit.plugins.replication.ReplicationConfig.FilterType;
import java.util.List;
import java.util.Optional;
+import java.util.Set;
import org.eclipse.jgit.transport.URIish;
/** Git destinations currently active for replication. */
@@ -45,14 +46,14 @@
List<Destination> getAll(FilterType filterType);
/**
- * Return the active replication destinations for a uri/project/ref triplet.
+ * Return the active replication destinations for a uri/project/refs triplet.
*
* @param uriish uri of the destinations
* @param project name of the project
- * @param ref ref name
+ * @param refs ref names
* @return the list of active destinations
*/
- List<Destination> getDestinations(URIish uriish, Project.NameKey project, String ref);
+ List<Destination> getDestinations(URIish uriish, Project.NameKey project, Set<String> refs);
/** Returns true if there are no destinations, false otherwise. */
boolean isEmpty();
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationFileBasedConfig.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationFileBasedConfig.java
index 5458b6c..93990f4 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationFileBasedConfig.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationFileBasedConfig.java
@@ -16,6 +16,7 @@
import static com.googlesource.gerrit.plugins.replication.ReplicationQueue.repLog;
import com.google.common.base.Strings;
+import com.google.gerrit.common.Nullable;
import com.google.gerrit.extensions.annotations.PluginData;
import com.google.gerrit.server.config.SitePaths;
import com.google.inject.Inject;
@@ -59,6 +60,7 @@
this.pluginDataDir = pluginDataDir;
}
+ @Nullable
public static String replaceName(String in, String name, boolean keyIsOptional) {
String key = "${name}";
int n = in.indexOf(key);
@@ -72,8 +74,8 @@
}
/**
- * See
- * {@link com.googlesource.gerrit.plugins.replication.ReplicationConfig#isReplicateAllOnPluginStart()}
+ * See {@link
+ * com.googlesource.gerrit.plugins.replication.ReplicationConfig#isReplicateAllOnPluginStart()}
*/
@Override
public boolean isReplicateAllOnPluginStart() {
@@ -81,8 +83,8 @@
}
/**
- * See
- * {@link com.googlesource.gerrit.plugins.replication.ReplicationConfig#isDefaultForceUpdate()}
+ * See {@link
+ * com.googlesource.gerrit.plugins.replication.ReplicationConfig#isDefaultForceUpdate()}
*/
@Override
public boolean isDefaultForceUpdate() {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationModule.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationModule.java
index 75fa5b3..9f331f2 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationModule.java
@@ -19,7 +19,7 @@
import com.google.common.eventbus.EventBus;
import com.google.gerrit.extensions.annotations.Exports;
import com.google.gerrit.extensions.config.CapabilityDefinition;
-import com.google.gerrit.extensions.events.GitReferenceUpdatedListener;
+import com.google.gerrit.extensions.events.GitBatchRefUpdateListener;
import com.google.gerrit.extensions.events.HeadUpdatedListener;
import com.google.gerrit.extensions.events.LifecycleListener;
import com.google.gerrit.extensions.events.ProjectDeletedListener;
@@ -44,6 +44,7 @@
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
+import org.apache.http.impl.client.CloseableHttpClient;
import org.eclipse.jgit.errors.ConfigInvalidException;
import org.eclipse.jgit.storage.file.FileBasedConfig;
import org.eclipse.jgit.transport.SshSessionFactory;
@@ -68,7 +69,7 @@
.annotatedWith(UniqueAnnotations.create())
.to(ReplicationQueue.class);
- DynamicSet.bind(binder(), GitReferenceUpdatedListener.class).to(ReplicationQueue.class);
+ DynamicSet.bind(binder(), GitBatchRefUpdateListener.class).to(ReplicationQueue.class);
DynamicSet.bind(binder(), ProjectDeletedListener.class).to(ReplicationQueue.class);
DynamicSet.bind(binder(), HeadUpdatedListener.class).to(ReplicationQueue.class);
@@ -122,6 +123,7 @@
bind(SshSessionFactory.class).toProvider(ReplicationSshSessionFactoryProvider.class);
bind(TransportFactory.class).to(TransportFactoryImpl.class).in(Scopes.SINGLETON);
+ bind(CloseableHttpClient.class).toProvider(HttpClientProvider.class).in(Scopes.SINGLETON);
}
private FileBasedConfig getReplicationConfig() {
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 4ff7653..aa00634 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
@@ -17,15 +17,17 @@
import static java.util.concurrent.TimeUnit.SECONDS;
import com.google.auto.value.AutoValue;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Queues;
import com.google.gerrit.common.UsedAt;
import com.google.gerrit.entities.Project;
-import com.google.gerrit.extensions.events.GitReferenceUpdatedListener;
+import com.google.gerrit.extensions.events.GitBatchRefUpdateListener;
import com.google.gerrit.extensions.events.HeadUpdatedListener;
import com.google.gerrit.extensions.events.LifecycleListener;
import com.google.gerrit.extensions.events.ProjectDeletedListener;
import com.google.gerrit.extensions.registration.DynamicItem;
import com.google.gerrit.server.events.EventDispatcher;
+import com.google.gerrit.server.extensions.events.GitReferenceUpdated;
import com.google.gerrit.server.git.WorkQueue;
import com.google.gerrit.util.logging.NamedFluentLogger;
import com.google.inject.Inject;
@@ -45,13 +47,14 @@
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
import org.eclipse.jgit.transport.URIish;
/** Manages automatic replication to remote repositories. */
public class ReplicationQueue
implements ObservableQueue,
LifecycleListener,
- GitReferenceUpdatedListener,
+ GitBatchRefUpdateListener,
ProjectDeletedListener,
HeadUpdatedListener {
static final String REPLICATION_LOG_NAME = "replication_log";
@@ -67,7 +70,7 @@
private final ProjectDeletionState.Factory projectDeletionStateFactory;
private volatile boolean running;
private final AtomicBoolean replaying = new AtomicBoolean();
- private final Queue<ReferenceUpdatedEvent> beforeStartupEventsQueue;
+ private final Queue<ReferencesUpdatedEvent> beforeStartupEventsQueue;
private Distributor distributor;
protected enum Prune {
@@ -128,71 +131,89 @@
public void scheduleFullSync(
Project.NameKey project, String urlMatch, ReplicationState state, boolean now) {
- fire(project, urlMatch, PushOne.ALL_REFS, state, now);
+ fire(
+ project,
+ urlMatch,
+ Set.of(new GitReferenceUpdated.UpdatedRef(PushOne.ALL_REFS, null, null, null)),
+ state,
+ now);
}
@Override
- public void onGitReferenceUpdated(GitReferenceUpdatedListener.Event event) {
- fire(event.getProjectName(), event.getRefName());
+ public void onGitBatchRefUpdate(GitBatchRefUpdateListener.Event event) {
+ fire(event.getProjectName(), event.getUpdatedRefs());
}
- private void fire(String projectName, String refName) {
+ private void fire(String projectName, Set<UpdatedRef> updatedRefs) {
ReplicationState state = new ReplicationState(new GitUpdateProcessing(dispatcher.get()));
- fire(Project.nameKey(projectName), null, refName, state, false);
+ fire(Project.nameKey(projectName), null, updatedRefs, state, false);
state.markAllPushTasksScheduled();
}
private void fire(
Project.NameKey project,
String urlMatch,
- String refName,
+ Set<UpdatedRef> updatedRefs,
ReplicationState state,
boolean now) {
if (!running) {
stateLog.warn(
"Replication plugin did not finish startup before event, event replication is postponed",
state);
- beforeStartupEventsQueue.add(ReferenceUpdatedEvent.create(project.get(), refName));
+ beforeStartupEventsQueue.add(ReferencesUpdatedEvent.create(project.get(), updatedRefs));
return;
}
for (Destination cfg : destinations.get().getAll(FilterType.ALL)) {
- pushReference(cfg, project, urlMatch, refName, state, now);
+ pushReferences(
+ cfg,
+ project,
+ urlMatch,
+ updatedRefs.stream().map(UpdatedRef::getRefName).collect(Collectors.toSet()),
+ state,
+ now);
}
}
- private void fireFromStorage(URIish uri, Project.NameKey project, String refName) {
+ private void fireFromStorage(URIish uri, Project.NameKey project, ImmutableSet<String> refNames) {
ReplicationState state = new ReplicationState(new GitUpdateProcessing(dispatcher.get()));
- for (Destination dest : destinations.get().getDestinations(uri, project, refName)) {
- dest.scheduleFromStorage(project, refName, uri, state);
+ for (Destination dest : destinations.get().getDestinations(uri, project, refNames)) {
+ dest.scheduleFromStorage(project, refNames, 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);
+ pushReferences(cfg, project, null, Set.of(refName), null, true);
}
- private void pushReference(
+ private void pushReferences(
Destination cfg,
Project.NameKey project,
String urlMatch,
- String refName,
+ Set<String> refNames,
ReplicationState state,
boolean now) {
boolean withoutState = state == null;
if (withoutState) {
state = new ReplicationState(new GitUpdateProcessing(dispatcher.get()));
}
- if (cfg.wouldPushProject(project) && cfg.wouldPushRef(refName)) {
+ Set<String> refNamesToPush = new HashSet<>();
+ for (String refName : refNames) {
+ if (cfg.wouldPushProject(project) && cfg.wouldPushRef(refName)) {
+ refNamesToPush.add(refName);
+ } else {
+ repLog.atFine().log("Skipping ref %s on project %s", refName, project.get());
+ }
+ }
+ if (!refNamesToPush.isEmpty()) {
for (URIish uri : cfg.getURIs(project, urlMatch)) {
replicationTasksStorage.create(
- ReplicateRefUpdate.create(project.get(), refName, uri, cfg.getRemoteConfigName()));
- cfg.schedule(project, refName, uri, state, now);
+ ReplicateRefUpdate.create(
+ project.get(), refNamesToPush, uri, cfg.getRemoteConfigName()));
+ cfg.schedule(project, refNamesToPush, uri, state, now);
}
- } else {
- repLog.atFine().log("Skipping ref %s on project %s", refName, project.get());
}
if (withoutState) {
state.markAllPushTasksScheduled();
@@ -215,7 +236,7 @@
@Override
public void run(ReplicationTasksStorage.ReplicateRefUpdate u) {
try {
- fireFromStorage(new URIish(u.uri()), Project.nameKey(u.project()), u.ref());
+ fireFromStorage(new URIish(u.uri()), Project.nameKey(u.project()), u.refs());
if (Prune.TRUE.equals(prune)) {
taskNamesByReplicateRefUpdate.remove(u);
}
@@ -240,7 +261,7 @@
@Override
public String toString(ReplicationTasksStorage.ReplicateRefUpdate u) {
- return "Scheduling push to " + String.format("%s:%s", u.project(), u.ref());
+ return "Scheduling push to " + String.format("%s:%s", u.project(), u.refs());
}
});
}
@@ -285,26 +306,31 @@
private void fireBeforeStartupEvents() {
Set<String> eventsReplayed = new HashSet<>();
- for (ReferenceUpdatedEvent event : beforeStartupEventsQueue) {
- String eventKey = String.format("%s:%s", event.projectName(), event.refName());
+ for (ReferencesUpdatedEvent event : beforeStartupEventsQueue) {
+ String eventKey = String.format("%s:%s", event.projectName(), event.getRefNames());
if (!eventsReplayed.contains(eventKey)) {
repLog.atInfo().log("Firing pending task %s", event);
- fire(event.projectName(), event.refName());
+ fire(event.projectName(), event.updatedRefs());
eventsReplayed.add(eventKey);
}
}
}
@AutoValue
- abstract static class ReferenceUpdatedEvent {
+ abstract static class ReferencesUpdatedEvent {
- static ReferenceUpdatedEvent create(String projectName, String refName) {
- return new AutoValue_ReplicationQueue_ReferenceUpdatedEvent(projectName, refName);
+ static ReferencesUpdatedEvent create(String projectName, Set<UpdatedRef> updatedRefs) {
+ return new AutoValue_ReplicationQueue_ReferencesUpdatedEvent(
+ projectName, ImmutableSet.copyOf(updatedRefs));
}
public abstract String projectName();
- public abstract String refName();
+ public abstract ImmutableSet<UpdatedRef> updatedRefs();
+
+ public Set<String> getRefNames() {
+ return updatedRefs().stream().map(UpdatedRef::getRefName).collect(Collectors.toSet());
+ }
}
protected class Distributor implements WorkQueue.CancelableRunnable {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationStateLogger.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationStateLogger.java
index 3e73033..a2bcf2b 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationStateLogger.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationStateLogger.java
@@ -31,19 +31,19 @@
@Override
public void warn(String msg, ReplicationState... states) {
stateWriteErr("Warning: " + msg, states);
- repLog.atWarning().log(msg);
+ repLog.atWarning().log("%s", msg);
}
@Override
public void error(String msg, ReplicationState... states) {
stateWriteErr("Error: " + msg, states);
- repLog.atSevere().log(msg);
+ repLog.atSevere().log("%s", msg);
}
@Override
public void error(String msg, Throwable t, ReplicationState... states) {
stateWriteErr("Error: " + msg, states);
- repLog.atSevere().withCause(t).log(msg);
+ repLog.atSevere().withCause(t).log("%s", msg);
}
private void stateWriteErr(String msg, ReplicationState[] states) {
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 c847f55..2e4b619 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java
@@ -18,15 +18,23 @@
import com.google.auto.value.AutoValue;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableSet;
import com.google.common.flogger.FluentLogger;
import com.google.common.hash.Hashing;
+import com.google.gerrit.common.Nullable;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.TypeAdapter;
+import com.google.gson.TypeAdapterFactory;
+import com.google.gson.reflect.TypeToken;
+import com.google.gson.stream.JsonReader;
+import com.google.gson.stream.JsonToken;
+import com.google.gson.stream.JsonWriter;
import com.google.inject.Inject;
import com.google.inject.ProvisionException;
import com.google.inject.Singleton;
import java.io.IOException;
+import java.net.URISyntaxException;
import java.nio.file.Files;
import java.nio.file.NoSuchFileException;
import java.nio.file.NotDirectoryException;
@@ -79,30 +87,50 @@
public static ReplicateRefUpdate create(Path file, Gson gson) throws IOException {
String json = new String(Files.readAllBytes(file), UTF_8);
- return gson.fromJson(json, ReplicateRefUpdate.class);
+ return create(gson.fromJson(json, ReplicateRefUpdate.class), file.getFileName().toString());
}
- public static ReplicateRefUpdate create(String project, String ref, URIish uri, String remote) {
+ public static ReplicateRefUpdate create(
+ String project, Set<String> refs, URIish uri, String remote) {
return new AutoValue_ReplicationTasksStorage_ReplicateRefUpdate(
- project, ref, uri.toASCIIString(), remote);
+ project,
+ ImmutableSet.copyOf(refs),
+ uri.toASCIIString(),
+ remote,
+ sha1(project, ImmutableSet.copyOf(refs), uri.toASCIIString(), remote));
+ }
+
+ public static ReplicateRefUpdate create(ReplicateRefUpdate u, String filename) {
+ return new AutoValue_ReplicationTasksStorage_ReplicateRefUpdate(
+ u.project(), u.refs(), u.uri(), u.remote(), filename);
}
public abstract String project();
- public abstract String ref();
+ public abstract ImmutableSet<String> refs();
public abstract String uri();
public abstract String remote();
- public String sha1() {
- return ReplicationTasksStorage.sha1(project() + "\n" + ref() + "\n" + uri() + "\n" + remote())
+ public abstract String sha1();
+
+ private static String sha1(String project, Set<String> refs, String uri, String remote) {
+ return ReplicationTasksStorage.sha1(
+ project + "\n" + refs.toString() + "\n" + uri + "\n" + remote)
.name();
}
@Override
public final String toString() {
- return "ref-update " + project() + ":" + ref() + " uri:" + uri() + " remote:" + remote();
+ return "ref-update "
+ + project()
+ + ":"
+ + refs().toString()
+ + " uri:"
+ + uri()
+ + " remote:"
+ + remote();
}
public static TypeAdapter<ReplicateRefUpdate> typeAdapter(Gson gson) {
@@ -130,7 +158,9 @@
runningUpdates = refUpdates.resolve("running");
waitingUpdates = refUpdates.resolve("waiting");
gson =
- new GsonBuilder().registerTypeAdapterFactory(AutoValueTypeAdapterFactory.create()).create();
+ new GsonBuilder()
+ .registerTypeAdapterFactory(new ReplicateRefUpdateTypeAdapterFactory())
+ .create();
}
private boolean isMultiPrimary() {
@@ -141,12 +171,12 @@
return new Task(r).create();
}
- public Set<String> start(UriUpdates uriUpdates) {
- Set<String> startedRefs = new HashSet<>();
+ public Set<ImmutableSet<String>> start(UriUpdates uriUpdates) {
+ Set<ImmutableSet<String>> startedRefs = new HashSet<>();
for (ReplicateRefUpdate update : uriUpdates.getReplicateRefUpdates()) {
Task t = new Task(update);
if (t.start()) {
- startedRefs.add(t.update.ref());
+ startedRefs.add(t.update.refs());
}
}
return startedRefs;
@@ -219,6 +249,104 @@
}
}
+ public static final class ReplicateRefUpdateTypeAdapterFactory implements TypeAdapterFactory {
+ static class ReplicateRefUpdateTypeAdapter<T> extends TypeAdapter<ReplicateRefUpdate> {
+
+ @Override
+ public void write(JsonWriter out, ReplicateRefUpdate value) throws IOException {
+ if (value == null) {
+ out.nullValue();
+ return;
+ }
+ out.beginObject();
+
+ out.name("project");
+ out.value(value.project());
+
+ out.name("refs");
+ out.beginArray();
+ for (String ref : value.refs()) {
+ out.value(ref);
+ }
+ out.endArray();
+
+ out.name("uri");
+ out.value(value.uri());
+
+ out.name("remote");
+ out.value(value.remote());
+
+ out.endObject();
+ }
+
+ @Nullable
+ @Override
+ public ReplicateRefUpdate read(JsonReader in) throws IOException {
+ if (in.peek() == JsonToken.NULL) {
+ in.nextNull();
+ return null;
+ }
+ String project = null;
+ Set<String> refs = new HashSet<>();
+ URIish uri = null;
+ String remote = null;
+
+ String fieldname = null;
+ in.beginObject();
+
+ while (in.hasNext()) {
+ JsonToken token = in.peek();
+
+ if (token.equals(JsonToken.NAME)) {
+ fieldname = in.nextName();
+ }
+
+ switch (fieldname) {
+ case "project":
+ project = in.nextString();
+ break;
+ case "refs":
+ in.beginArray();
+ while (in.hasNext()) {
+ refs.add(in.nextString());
+ }
+ in.endArray();
+ break;
+ case "ref":
+ refs.add(in.nextString());
+ break;
+ case "uri":
+ try {
+ uri = new URIish(in.nextString());
+ } catch (URISyntaxException e) {
+ throw new IOException("Unable to parse remote URI", e);
+ }
+ break;
+ case "remote":
+ remote = in.nextString();
+ break;
+ default:
+ throw new IOException(String.format("Unknown field in stored task: %s", fieldname));
+ }
+ }
+
+ in.endObject();
+ return ReplicateRefUpdate.create(project, refs, uri, remote);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ @Nullable
+ @Override
+ public <T> TypeAdapter<T> create(Gson gson, TypeToken<T> type) {
+ if (type.equals(TypeToken.get(AutoValue_ReplicationTasksStorage_ReplicateRefUpdate.class))
+ || type.equals(TypeToken.get(ReplicateRefUpdate.class))) {
+ return (TypeAdapter<T>) new ReplicateRefUpdateTypeAdapter<T>();
+ }
+ return null;
+ }
+ }
+
@VisibleForTesting
class Task {
public final ReplicateRefUpdate update;
@@ -275,7 +403,8 @@
String message = "Error while deleting task %s";
if (isMultiPrimary() && e instanceof NoSuchFileException) {
logger.atFine().log(
- message + " (expected after recovery from another node's startup with multi-primaries and distributor enabled)",
+ message
+ + " (expected after recovery from another node's startup with multi-primaries and distributor enabled)",
taskKey);
} else {
logger.atSevere().withCause(e).log(message, taskKey);
@@ -283,7 +412,8 @@
}
}
- private boolean rename(Path from, Path to) {
+ @VisibleForTesting
+ boolean rename(Path from, Path to) {
try {
logger.atFine().log("RENAME %s to %s %s", from, to, updateLog());
Files.move(from, to, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
@@ -302,7 +432,7 @@
}
private String updateLog() {
- return String.format("(%s:%s => %s)", update.project(), update.ref(), update.uri());
+ return String.format("(%s:%s => %s)", update.project(), update.refs(), update.uri());
}
}
}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/UriUpdates.java b/src/main/java/com/googlesource/gerrit/plugins/replication/UriUpdates.java
index a9985d2..77a5574 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/UriUpdates.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/UriUpdates.java
@@ -14,6 +14,7 @@
package com.googlesource.gerrit.plugins.replication;
+import com.google.common.collect.ImmutableSet;
import com.google.gerrit.entities.Project;
import java.util.List;
import java.util.Set;
@@ -28,14 +29,15 @@
String getRemoteName();
- Set<String> getRefs();
+ Set<ImmutableSet<String>> getRefs();
default List<ReplicationTasksStorage.ReplicateRefUpdate> getReplicateRefUpdates() {
+ // TODO: keep batch refs together
return getRefs().stream()
.map(
- (ref) ->
+ (refs) ->
ReplicationTasksStorage.ReplicateRefUpdate.create(
- getProjectNameKey().get(), ref, getURI(), getRemoteName()))
+ getProjectNameKey().get(), refs, getURI(), getRemoteName()))
.collect(Collectors.toList());
}
}
diff --git a/src/main/resources/Documentation/cmd-list.md b/src/main/resources/Documentation/cmd-list.md
index b6688e0..b0a8254 100644
--- a/src/main/resources/Documentation/cmd-list.md
+++ b/src/main/resources/Documentation/cmd-list.md
@@ -7,7 +7,7 @@
SYNOPSIS
--------
-```
+```console
ssh -p @SSH_PORT@ @SSH_HOST@ @PLUGIN@ list
[--remote <PATTERN>]
[--detail]
@@ -30,39 +30,39 @@
-------
`--remote <PATTERN>`
-: Only print information for destinations whose remote name matches
- the `PATTERN`.
+: Only print information for destinations whose remote name matches
+the `PATTERN`.
`--detail`
-: Print additional detailed information: AdminUrl, AuthGroup, Project
- and queue (pending and in-flight).
+: Print additional detailed information: AdminUrl, AuthGroup, Project
+and queue (pending and in-flight).
`--json`
-: Output in json format.
+: Output in json format.
EXAMPLES
--------
List all destinations:
-```
+```console
$ ssh -p @SSH_PORT@ @SSH_HOST@ @PLUGIN@ list
```
List all destinations detail information:
-```
+```console
$ ssh -p @SSH_PORT@ @SSH_HOST@ @PLUGIN@ list --detail
```
List all destinations detail information in json format:
-```
+```console
$ ssh -p @SSH_PORT@ @SSH_HOST@ @PLUGIN@ list --detail --json
```
List destinations whose name contains mirror:
-```
+```console
$ ssh -p @SSH_PORT@ @SSH_HOST@ @PLUGIN@ list --remote mirror
$ ssh -p @SSH_PORT@ @SSH_HOST@ @PLUGIN@ list --remote ^.*mirror.*
```
diff --git a/src/main/resources/Documentation/cmd-start.md b/src/main/resources/Documentation/cmd-start.md
index 1a77c72..e1b7341 100644
--- a/src/main/resources/Documentation/cmd-start.md
+++ b/src/main/resources/Documentation/cmd-start.md
@@ -7,7 +7,8 @@
SYNOPSIS
--------
-```
+
+```console
ssh -p @SSH_PORT@ @SSH_HOST@ @PLUGIN@ start
[--now]
[--wait]
@@ -57,14 +58,14 @@
If you get message "Nothing to replicate" while running this command,
it may be caused by several reasons, such as you give a wrong url
-pattern in command options, or the authGroup in the replication.config
+pattern in command options, or the authGroup in the `replication.config`
has no read access for the replicated projects.
If one or several project patterns are supplied, only those projects
conforming to both this/these pattern(s) and those defined in
-replication.config for the target host(s) are queued for replication.
+`replication.config` for the target host(s) are queued for replication.
-The patterns follow the same format as those in replication.config,
+The patterns follow the same format as those in `replication.config`,
where wildcard or regular expression patterns can be given.
Regular expression patterns must match a complete project name to be
considered a match.
@@ -86,58 +87,58 @@
-------
`--now`
-: Start replicating right away without waiting the per remote
- replication delay.
+: Start replicating right away without waiting the per remote
+replication delay.
`--wait`
-: Wait for replication to finish before exiting.
+: Wait for replication to finish before exiting.
`--all`
-: Schedule replication for all projects.
+: Schedule replication for all projects.
`--url <PATTERN>`
-: Replicate only to replication destinations whose configuration
- URL contains the substring `PATTERN`, or whose expanded project
- URL contains `PATTERN`. This can be useful to replicate only to
- a previously down node, which has been brought back online.
+: Replicate only to replication destinations whose configuration
+URL contains the substring `PATTERN`, or whose expanded project
+URL contains `PATTERN`. This can be useful to replicate only to
+a previously down node, which has been brought back online.
EXAMPLES
--------
Replicate every project, to every configured remote:
-```
+```console
$ ssh -p @SSH_PORT@ @SSH_HOST@ @PLUGIN@ start --all
```
Replicate only to `srv2` now that it is back online:
-```
+```console
$ ssh -p @SSH_PORT@ @SSH_HOST@ @PLUGIN@ start --url srv2 --all
```
Replicate only the `tools/gerrit` project, after deleting a ref
locally by hand:
-```
+```console
$ git --git-dir=/home/git/tools/gerrit.git update-ref -d refs/changes/00/100/1
$ ssh -p @SSH_PORT@ @SSH_HOST@ @PLUGIN@ start tools/gerrit
```
Replicate only projects located in the `documentation` subdirectory:
-```
+```console
$ ssh -p @SSH_PORT@ @SSH_HOST@ @PLUGIN@ start documentation/*
```
Replicate projects whose path includes a folder named `vendor` to host replica1:
-```
+```console
$ ssh -p @SSH_PORT@ @SSH_HOST@ @PLUGIN@ start --url replica1 ^(|.*/)vendor(|/.*)
```
Replicate to only one specific destination URL:
-```
+```console
$ ssh -p @SSH_PORT@ @SSH_HOST@ @PLUGIN@ start --url https://example.com/tools/gerrit.git
```
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md
index d401030..8f1ca86 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md
@@ -9,12 +9,12 @@
file. The easiest way to add the host key is to connect once by hand
with the command line:
-```
+```console
sudo su -c 'ssh mirror1.us.some.org echo' gerrit2
```
*NOTE:* make sure the local user's ssh keys format is PEM, here how to generate them:
-```
+```console
ssh-keygen -m PEM -t rsa -C "your_email@example.com"
```
@@ -22,7 +22,7 @@
Next, create `$site_path/etc/replication.config` as a Git-style config
file, for example to replicate in parallel to four different hosts:
-```
+```ini
[remote "host-one"]
url = gerrit2@host-one.example.com:/some/path/${name}.git
@@ -39,7 +39,7 @@
Then reload the replication plugin to pick up the new configuration:
-```
+```console
ssh -p 29418 localhost gerrit plugin reload replication
```
@@ -53,16 +53,16 @@
The replication plugin is designed to allow multiple primaries in a
cluster to efficiently cooperate together via the replication event
persistence subsystem. To enable this cooperation, the directory
-pointed to by the replication.eventsDirectory config key must reside on
+pointed to by the `replication.eventsDirectory` config key must reside on
a shared filesystem, such as NFS. By default, simply pointing multiple
primaries to the same eventsDirectory will enable some cooperation by
preventing the same replication push from being duplicated by more
than one primary.
To further improve cooperation across the cluster, the
-replication.distributionInterval config value can be set. With
+`replication.distributionInterval` config value can be set. With
distribution enabled, the replication queues for all the nodes sharing
-the same eventsDirectory will reflect approximately the same outstanding
+the same `eventsDirectory` will reflect approximately the same outstanding
replication work (i.e. tasks waiting in the queue). Replication pushes
which are running will continue to only be visible in the queue of the
node on which the push is actually happening. This feature helps
@@ -181,7 +181,7 @@
(such as other masters in the same cluster) writing to the same
persistence store. To ensure that updates are seen well before their
replicationDelay expires when the distributor is used, the recommended
- value for this is approximately the smallest remote.NAME.replicationDelay
+ value for this is approximately the smallest `remote.NAME.replicationDelay`
divided by 5.
<a name="replication.updateRefErrorMaxRetries">replication.updateRefErrorMaxRetries</a>
@@ -200,7 +200,7 @@
"failed to lock" is detected as that is the legacy string used by git.
A good value would be 3 retries or less, depending on how often
- you see updateRefError collisions in your server logs. A too highly set
+ you see `updateRefError` collisions in your server logs. A too highly set
value risks keeping around the replication operations in the queue
for a long time, and the number of items in the queue will increase
with time.
@@ -211,7 +211,7 @@
will never succeed.
The issue can also be mitigated somewhat by increasing the
- replicationDelay.
+ `replicationDelay`.
Default: 0 (disabled, i.e. never retry)
@@ -233,7 +233,7 @@
the replication event is discarded from the queue and the remote
destinations may remain out of sync.
- Can be overridden at remote-level by setting replicationMaxRetries.
+ Can be overridden at remote-level by setting `replicationMaxRetries`.
By default, pushes are retried indefinitely.
@@ -247,9 +247,9 @@
it will trigger all replications found under this directory.
For replication to work, is is important that atomic renames be possible
- from within any subdirectory of the eventsDirectory to within any other
- subdirectory of the eventsDirectory. This generally means that the entire
- contents of the eventsDirectory should live on the same filesystem.
+ from within any subdirectory of the `eventsDirectory` to within any other
+ subdirectory of the `eventsDirectory`. This generally means that the entire
+ contents of the `eventsDirectory` should live on the same filesystem.
When not set, defaults to the plugin's data directory.
@@ -280,12 +280,12 @@
remote block, listing different destinations which share the
same settings.
- The adminUrl can be used as an ssh alternative to the url
+ The `adminUrl` can be used as an ssh alternative to the `url`
option, but only related to repository creation. If not
specified, the repository creation tries to follow the default
way through the url value specified.
- It is useful when the remote.NAME.url protocols do not allow
+ It is useful when the `remote.NAME.url` protocols do not allow
repository creation although their usage is mandatory in the
local environment. In that case, an alternative SSH url could
be specified to repository creation.
@@ -326,7 +326,7 @@
contain any code-review or NoteDb information.
Using `gerrit+ssh` for replicating all Gerrit repositories
- would result in failures on the All-Users.git replication and
+ would result in failures on the `All-Users.git` replication and
would not be able to replicate changes magic refs and indexes
across nodes.
@@ -427,7 +427,7 @@
This is a Gerrit specific extension to the Git remote block.
- By default, use replication.maxRetries.
+ By default, use `replication.maxRetries`.
remote.NAME.drainQueueAttempts
: Maximum number of attempts to drain the replication event queue before
@@ -436,7 +436,7 @@
When stopping the plugin, the shutdown will be delayed trying to drain
the event queue.
- The maximum delay is "drainQueueAttempts" * "replicationDelay" seconds.
+ The maximum delay is `drainQueueAttempts * replicationDelay` seconds.
When not set or set to 0, the queue is not drained and the pending
replication events are cancelled.
@@ -455,7 +455,7 @@
remote.NAME.authGroup
: Specifies the name of a group that the remote should use to
- access the repositories. Multiple authGroups may be specified
+ access the repositories. Multiple `authGroups` may be specified
within a single remote block to signify a wider access right.
In the project administration web interface the read access
can be specified for this group to control if a project should
@@ -521,7 +521,7 @@
dashes in the remote repository name. If set to "underscore",
slashes will be replaced with underscores in the repository name.
- Option "basenameOnly" makes `${name}` to be only the basename
+ Option `basenameOnly` makes `${name}` to be only the basename
(the part after the last slash) of the repository path on the
Gerrit server, e.g. `${name}` of `foo/bar/my-repo.git` would
be `my-repo`.
@@ -534,7 +534,7 @@
> to conflicts where commits can be lost between the two repositories
> replicating to the same target `my-repo`.
- By default, "slash", i.e. remote names will contain slashes as
+ By default, `slash`, i.e. remote names will contain slashes as
they do in Gerrit.
<a name="remote.NAME.projects">remote.NAME.projects</a>
@@ -566,7 +566,7 @@
By default, replicates without matching, i.e. replicates
everything to all remotes.
-remote.NAME.slowLatencyThreshold
+<a name="remote.NAME.slowLatencyThreshold">remote.NAME.slowLatencyThreshold</a>
: the time duration after which the replication of a project to this
destination will be considered "slow". A slow project replication
will cause additional metrics to be exposed for further investigation.
@@ -587,6 +587,16 @@
Note that `pushBatchSize` is ignored when *Cluster Replication* is configured
- when `replication.distributionInterval` has value > 0.
+remote.NAME.replicateNoteDbMetaRefs
+: Whether to replicate the NoteDb meta refs (`refs/changes/*/meta`,
+ `refs/changes/*/robot-comments`, `refs/draft-comments/*`,
+ `refs/starred-changes/*`) or not. This setting is useful when the remote
+ replica does not run a Gerrit instance and one wants to turn off replicating
+ NoteDb meta refs to that remote.
+
+ By default, true.
+
+
Directory `replication`
--------------------
The optional directory `$site_path/etc/replication` contains Git-style
@@ -604,7 +614,7 @@
Static configuration in `$site_path/etc/replication.config`:
-```
+```ini
[gerrit]
autoReload = true
replicateOnStartup = false
@@ -617,7 +627,7 @@
* File `$site_path/etc/replication/host-one.config`
- ```
+ ```ini
[remote]
url = gerrit2@host-one.example.com:/some/path/${name}.git
```
@@ -625,7 +635,7 @@
* File `$site_path/etc/replication/pubmirror.config`
- ```
+ ```ini
[remote]
url = mirror1.us.some.org:/pub/git/${name}.git
url = mirror2.us.some.org:/pub/git/${name}.git
@@ -639,7 +649,7 @@
Replication plugin resolves config files to the following configuration:
-```
+```ini
[gerrit]
autoReload = true
replicateOnStartup = false
@@ -682,7 +692,7 @@
Gerrit reads and caches the `~/.ssh/config` at startup, and
supports most SSH configuration options. For example:
-```
+```text
Host host-one.example.com
IdentityFile ~/.ssh/id_hostone
PreferredAuthentications publickey
@@ -693,10 +703,10 @@
PreferredAuthentications publickey
```
-*IdentityFile* and *PreferredAuthentications* must be defined for all the hosts.
+`IdentityFile` and `PreferredAuthentications` must be defined for all the hosts.
Here an example of the minimum `~/.ssh/config` needed:
-```
+```text
Host *
IdentityFile ~/.ssh/id_rsa
PreferredAuthentications publickey
diff --git a/src/main/resources/Documentation/extension-point.md b/src/main/resources/Documentation/extension-point.md
index 83e7e45..f6579fa 100644
--- a/src/main/resources/Documentation/extension-point.md
+++ b/src/main/resources/Documentation/extension-point.md
@@ -1,11 +1,12 @@
@PLUGIN@ extension points
-==============
+=========================
The replication plugin exposes an extension point to allow influencing its behaviour from another plugin or a script.
Extension points can be defined from the replication plugin only when it is loaded as [libModule](../../../Documentation/config-gerrit.html#gerrit.installModule) and
implemented by another plugin by declaring a `provided` dependency from the replication plugin.
-### Install extension libModule
+Install extension libModule
+---------------------------
The replication plugin's extension points are defined in the `c.g.g.p.r.ReplicationExtensionPointModule`
that needs to be configured as libModule.
@@ -15,15 +16,15 @@
Example:
-```
+```ini
[gerrit]
installModule = com.googlesource.gerrit.plugins.replication.ReplicationExtensionPointModule
```
> **NOTE**: Use and configuration of the replication plugin as library module requires a Gerrit server restart and does not support hot plugin install or upgrade.
-
-### Extension points
+Extension points
+----------------
* `com.googlesource.gerrit.plugins.replication.ReplicationPushFilter`
@@ -34,7 +35,7 @@
Example:
- ```
+ ```java
DynamicItem.bind(binder(), ReplicationPushFilter.class).to(ReplicationPushFilterImpl.class);
```
@@ -48,6 +49,6 @@
Example:
- ```
+ ```java
DynamicItem.bind(binder(), AdminApiFactory.class).to(AdminApiFactoryImpl.class);
```
diff --git a/src/main/resources/Documentation/metrics.md b/src/main/resources/Documentation/metrics.md
index ef8b150..f2ac06d 100644
--- a/src/main/resources/Documentation/metrics.md
+++ b/src/main/resources/Documentation/metrics.md
@@ -1,23 +1,28 @@
-# Metrics
+Metrics
+=======
Some metrics are emitted when replication occurs to a remote destination.
The granularity of the metrics recorded is at destination level, however when a particular project replication is flagged
-as slow. This happens when the replication took longer than allowed threshold (see _remote.NAME.slowLatencyThreshold_ in [config.md](config.md))
+as slow. This happens when the replication took longer than allowed threshold (see [`remote.NAME.slowLatencyThreshold`](config.md#remote.NAME.slowLatencyThreshold)).
The reason only slow metrics are published, rather than all, is to contain their number, which, on a big Gerrit installation
could potentially be considerably big.
-### Project level
+Project level
+-------------
-* plugins_replication_latency_slower_than_<threshold>_<destinationName>_<ProjectName> - Time spent pushing <ProjectName> to remote <destinationName> (in ms)
+* `plugins_replication_latency_slower_than_<threshold>_<destinationName>_<ProjectName>` - Time spent pushing `<ProjectName>` to remote `<destinationName>` (in ms)
-### Destination level
+Destination level
+-----------------
-* plugins_replication_replication_delay_<destinationName> - Time spent waiting before pushing to remote <destinationName> (in ms)
-* plugins_replication_replication_retries_<destinationName> - Number of retries when pushing to remote <destinationName>
-* plugins_replication_replication_latency_<destinationName> - Time spent pushing to remote <destinationName> (in ms)
+* `plugins_replication_replication_delay_<destinationName>` - Time spent waiting before pushing to remote `<destinationName>` (in ms)
+* `plugins_replication_replication_retries_<destinationName>` - Number of retries when pushing to remote `<destinationName>`
+* `plugins_replication_replication_latency_<destinationName>` - Time spent pushing to remote `<destinationName>` (in ms)
-### Example
+Example
+-------
+
```
# HELP plugins_replication_replication_delay_destination Generated from Dropwizard metric import (metric=plugins/replication/replication_delay/destination, type=com.codahale.metrics.Histogram)
# TYPE plugins_replication_replication_delay_destination summary
@@ -58,4 +63,4 @@
plugins_replication_latency_slower_than_60_destinationName_projectName{quantile="0.99",} 278.0
plugins_replication_latency_slower_than_60_destinationName_projectName{quantile="0.999",} 278.0
plugins_replication_latency_slower_than_60_destinationName_projectName 1.0
-```
\ No newline at end of file
+```
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/AutoReloadRunnableTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/AutoReloadRunnableTest.java
index e7339d9..725052c 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/AutoReloadRunnableTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/AutoReloadRunnableTest.java
@@ -82,7 +82,7 @@
}
private Provider<ReplicationConfig> newVersionConfigProvider() {
- return new Provider<ReplicationConfig>() {
+ return new Provider<>() {
@Override
public ReplicationConfig get() {
return new ReplicationFileBasedConfig(sitePaths, sitePaths.data_dir) {
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ChainedSchedulerTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ChainedSchedulerTest.java
index 90191f2..242922d 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ChainedSchedulerTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ChainedSchedulerTest.java
@@ -19,9 +19,9 @@
import static java.util.concurrent.TimeUnit.SECONDS;
import com.google.common.collect.ForwardingIterator;
+import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Iterator;
-import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
@@ -367,7 +367,7 @@
WaitingRunner runner = new WaitingRunner();
int batchSize = 5; // how many tasks are started concurrently
- Queue<CountDownLatch> batches = new LinkedList<>();
+ Queue<CountDownLatch> batches = new ArrayDeque<>();
for (int b = 0; b < blockSize; b++) {
batches.add(executeWaitingRunnableBatch(batchSize, executor));
}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/FakeExecutorService.java b/src/test/java/com/googlesource/gerrit/plugins/replication/FakeExecutorService.java
index 2f7059e..ce94187 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/FakeExecutorService.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/FakeExecutorService.java
@@ -14,6 +14,7 @@
package com.googlesource.gerrit.plugins.replication;
+import com.google.gerrit.common.Nullable;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
@@ -103,6 +104,7 @@
return null;
}
+ @Nullable
@Override
public ScheduledFuture<?> scheduleAtFixedRate(
Runnable command, long initialDelay, long period, TimeUnit unit) {
@@ -110,6 +112,7 @@
return null;
}
+ @Nullable
@Override
public ScheduledFuture<?> scheduleWithFixedDelay(
Runnable command, long initialDelay, long delay, TimeUnit unit) {
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ForwardingProxy.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ForwardingProxy.java
index 5e6bde8..c8dcae0 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ForwardingProxy.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ForwardingProxy.java
@@ -14,6 +14,7 @@
package com.googlesource.gerrit.plugins.replication;
+import com.google.gerrit.common.Nullable;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
@@ -55,6 +56,7 @@
return method.invoke(delegate, args);
}
+ @Nullable
protected Method getOverriden(Method method) {
try {
Method implementedByOverrider =
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/PushOneTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/PushOneTest.java
index f9d15f2..21fe85c 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/PushOneTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/PushOneTest.java
@@ -18,6 +18,7 @@
import static org.eclipse.jgit.lib.Ref.Storage.NEW;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -26,6 +27,7 @@
import static org.mockito.Mockito.when;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
import com.google.gerrit.entities.Project;
import com.google.gerrit.extensions.registration.DynamicItem;
import com.google.gerrit.metrics.Timer1;
@@ -71,6 +73,7 @@
import org.eclipse.jgit.util.FS;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
@@ -199,7 +202,7 @@
PushOne pushOne = createPushOne(replicationPushFilter);
- pushOne.addRef(PushOne.ALL_REFS);
+ pushOne.addRefBatch(ImmutableSet.of(PushOne.ALL_REFS));
pushOne.run();
isCallFinished.await(TEST_PUSH_TIMEOUT_SECS, TimeUnit.SECONDS);
@@ -221,7 +224,7 @@
PushOne pushOne = createPushOne(replicationPushFilter);
- pushOne.addRef(PushOne.ALL_REFS);
+ pushOne.addRefBatch(ImmutableSet.of(PushOne.ALL_REFS));
pushOne.run();
isCallFinished.await(10, TimeUnit.SECONDS);
@@ -230,6 +233,86 @@
}
@Test
+ public void shouldPushMetaRefTogetherWithChangeRef() throws InterruptedException, IOException {
+ when(destinationMock.replicateNoteDbMetaRefs()).thenReturn(true);
+ PushOne pushOne = Mockito.spy(createPushOne(null));
+
+ Ref newLocalChangeRef =
+ new ObjectIdRef.Unpeeled(
+ NEW,
+ "refs/changes/11/11111/1",
+ ObjectId.fromString("0000000000000000000000000000000000000002"));
+
+ Ref newLocalChangeMetaRef =
+ new ObjectIdRef.Unpeeled(
+ NEW,
+ "refs/changes/11/11111/meta",
+ ObjectId.fromString("0000000000000000000000000000000000000003"));
+
+ localRefs.add(newLocalChangeRef);
+ localRefs.add(newLocalChangeMetaRef);
+
+ pushOne.addRefBatch(
+ ImmutableSet.of(newLocalChangeRef.getName(), newLocalChangeMetaRef.getName()));
+ pushOne.run();
+
+ isCallFinished.await(10, TimeUnit.SECONDS);
+ verify(transportMock, atLeastOnce()).push(any(), any());
+ verify(pushOne, times(2)).push(any(), any(), any());
+ }
+
+ @Test
+ public void skipPushingMetaRefWhenReplicateNoteDbMetaRefsIsSetToFalse()
+ throws InterruptedException, IOException {
+ when(destinationMock.replicateNoteDbMetaRefs()).thenReturn(false);
+ PushOne pushOne = Mockito.spy(createPushOne(null));
+
+ Ref newLocalChangeRef =
+ new ObjectIdRef.Unpeeled(
+ NEW,
+ "refs/changes/11/11111/1",
+ ObjectId.fromString("0000000000000000000000000000000000000002"));
+
+ Ref newLocalChangeMetaRef =
+ new ObjectIdRef.Unpeeled(
+ NEW,
+ "refs/changes/11/11111/meta",
+ ObjectId.fromString("0000000000000000000000000000000000000003"));
+
+ localRefs.add(newLocalChangeRef);
+ localRefs.add(newLocalChangeMetaRef);
+
+ pushOne.addRefBatch(
+ ImmutableSet.of(newLocalChangeRef.getName(), newLocalChangeMetaRef.getName()));
+ pushOne.run();
+
+ isCallFinished.await(10, TimeUnit.SECONDS);
+ verify(transportMock, atLeastOnce()).push(any(), any());
+ verify(pushOne, times(1)).push(any(), any(), any());
+ }
+
+ @Test
+ public void shouldNotAttemptDuplicateRemoteRefUpdate() throws InterruptedException, IOException {
+ PushOne pushOne = Mockito.spy(createPushOne(null));
+
+ Ref newLocalChangeRef =
+ new ObjectIdRef.Unpeeled(
+ NEW,
+ "refs/changes/11/11111/1",
+ ObjectId.fromString("0000000000000000000000000000000000000002"));
+
+ localRefs.add(newLocalChangeRef);
+
+ pushOne.addRefBatch(ImmutableSet.of(newLocalChangeRef.getName()));
+ pushOne.addRefBatch(ImmutableSet.of(newLocalChangeRef.getName()));
+ pushOne.run();
+
+ isCallFinished.await(10, TimeUnit.SECONDS);
+ verify(transportMock, times(1)).push(any(), any());
+ verify(pushOne, times(1)).push(any(), any(), any());
+ }
+
+ @Test
public void shouldPushInSingleOperationWhenPushBatchSizeIsNotConfigured()
throws InterruptedException, IOException {
replicateTwoRefs(createPushOne(null));
@@ -268,7 +351,7 @@
when(gitRepositoryManagerMock.openRepository(projectNameKey))
.thenThrow(new RepositoryNotFoundException("not found"));
PushOne pushOne = createPushOne(null);
- pushOne.addRef(PushOne.ALL_REFS);
+ pushOne.addRefBatch(ImmutableSet.of(PushOne.ALL_REFS));
pushOne.setToRetry();
pushOne.run();
assertThat(pushOne.isRetrying()).isFalse();
@@ -280,7 +363,7 @@
NEW, "bar", ObjectId.fromString("0000000000000000000000000000000000000001"));
localRefs.add(barLocalRef);
- pushOne.addRef(PushOne.ALL_REFS);
+ pushOne.addRefBatch(ImmutableSet.of(PushOne.ALL_REFS));
pushOne.run();
isCallFinished.await(TEST_PUSH_TIMEOUT_SECS, TimeUnit.SECONDS);
@@ -335,7 +418,7 @@
@Override
public Callable<Object> answer(InvocationOnMock invocation) throws Throwable {
Callable<Object> originalCall = (Callable<Object>) invocation.getArguments()[0];
- return new Callable<Object>() {
+ return new Callable<>() {
@Override
public Object call() throws Exception {
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDaemon.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDaemon.java
index b6a3ed1..3bc86c7 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDaemon.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDaemon.java
@@ -22,6 +22,7 @@
import com.google.gerrit.acceptance.UseLocalDisk;
import com.google.gerrit.acceptance.WaitUtil;
import com.google.gerrit.acceptance.testsuite.project.ProjectOperations;
+import com.google.gerrit.common.Nullable;
import com.google.gerrit.entities.Project;
import com.google.gerrit.extensions.api.changes.NotifyHandling;
import com.google.gerrit.extensions.events.ProjectDeletedListener;
@@ -245,6 +246,7 @@
return true;
}
+ @Nullable
protected Ref checkedGetRef(Repository repo, String branchName) {
try {
return repo.getRefDatabase().exactRef(branchName);
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDistributorIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDistributorIT.java
index 3d0dfc1..dfcf250 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDistributorIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDistributorIT.java
@@ -24,6 +24,7 @@
import com.google.gerrit.server.git.WorkQueue;
import java.time.Duration;
import java.util.List;
+import java.util.Set;
import java.util.stream.Collectors;
import org.eclipse.jgit.lib.Ref;
import org.eclipse.jgit.lib.RefUpdate;
@@ -61,7 +62,7 @@
Project.NameKey targetProject = createTestProject(project + replica);
ReplicationTasksStorage.ReplicateRefUpdate ref =
ReplicationTasksStorage.ReplicateRefUpdate.create(
- project.get(), newBranch, new URIish(getProjectUri(targetProject)), remote);
+ project.get(), Set.of(newBranch), new URIish(getProjectUri(targetProject)), remote);
createBranch(project, master, newBranch);
setReplicationDestination(remote, replica, ALL_PROJECTS);
reloadConfig();
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationEventsIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationEventsIT.java
index 4cd55f9..b32829c 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationEventsIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationEventsIT.java
@@ -384,6 +384,7 @@
}
}
+ @SuppressWarnings("deprecation")
private boolean equals(ReplicationScheduledEvent scheduledEvent, Object other) {
if (!(other instanceof ReplicationScheduledEvent)) {
return false;
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationFanoutIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationFanoutIT.java
index 1c1f983..5f80e8c 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationFanoutIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationFanoutIT.java
@@ -191,7 +191,7 @@
Pattern refmaskPattern = Pattern.compile(refRegex);
return tasksStorage
.streamWaiting()
- .filter(task -> refmaskPattern.matcher(task.ref()).matches())
+ .filter(task -> task.refs().stream().anyMatch(ref -> refmaskPattern.matcher(ref).matches()))
.collect(toList());
}
}
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 eb2b999..9285c58 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java
@@ -111,6 +111,7 @@
Result pushResult = createChange();
RevCommit sourceCommit = pushResult.getCommit();
String sourceRef = pushResult.getPatchSet().refName();
+ String metaRef = sourceRef.substring(0, sourceRef.lastIndexOf('/') + 1).concat("meta");
try (Repository repo = repoManager.openRepository(targetProject)) {
waitUntil(() -> checkedGetRef(repo, sourceRef) != null);
@@ -118,6 +119,9 @@
Ref targetBranchRef = getRef(repo, sourceRef);
assertThat(targetBranchRef).isNotNull();
assertThat(targetBranchRef.getObjectId()).isEqualTo(sourceCommit.getId());
+
+ Ref targetBranchMetaRef = getRef(repo, metaRef);
+ assertThat(targetBranchMetaRef).isNotNull();
}
}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStorageDaemon.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStorageDaemon.java
index f549f47..b6d14e8 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStorageDaemon.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStorageDaemon.java
@@ -61,7 +61,7 @@
Pattern refmaskPattern = Pattern.compile(refRegex);
return tasksStorage
.streamWaiting()
- .filter(task -> refmaskPattern.matcher(task.ref()).matches())
+ .filter(task -> refmaskPattern.matcher(task.refs().toArray()[0].toString()).matches())
.collect(toList());
}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStorageIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStorageIT.java
index e2e1e21..9390798 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStorageIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStorageIT.java
@@ -34,6 +34,7 @@
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import java.util.regex.Pattern;
import java.util.stream.Stream;
import org.eclipse.jgit.transport.URIish;
@@ -96,7 +97,7 @@
.forEach(
(update) -> {
try {
- UriUpdates uriUpdates = TestUriUpdates.create(update);
+ UriUpdates uriUpdates = new TestUriUpdates(update);
tasksStorage.start(uriUpdates);
tasksStorage.finish(uriUpdates);
} catch (URISyntaxException e) {
@@ -125,7 +126,7 @@
.forEach(
(update) -> {
try {
- UriUpdates uriUpdates = TestUriUpdates.create(update);
+ UriUpdates uriUpdates = new TestUriUpdates(update);
tasksStorage.start(uriUpdates);
tasksStorage.finish(uriUpdates);
} catch (URISyntaxException e) {
@@ -211,7 +212,7 @@
.forEach(
(task) -> {
assertThat(task.uri()).isEqualTo(expectedURI);
- assertThat(task.ref()).isEqualTo(PushOne.ALL_REFS);
+ assertThat(task.refs()).isEqualTo(Set.of(PushOne.ALL_REFS));
});
}
@@ -236,7 +237,7 @@
.forEach(
(task) -> {
assertThat(task.uri()).isEqualTo(expectedURI);
- assertThat(task.ref()).isEqualTo(PushOne.ALL_REFS);
+ assertThat(task.refs()).isEqualTo(Set.of(PushOne.ALL_REFS));
});
}
@@ -351,14 +352,14 @@
String changeRef, String remote) {
return tasksStorage
.streamWaiting()
- .filter(task -> changeRef.equals(task.ref()))
+ .filter(task -> task.refs().stream().anyMatch(ref -> changeRef.equals(ref)))
.filter(task -> remote.equals(task.remote()));
}
private Stream<ReplicateRefUpdate> changeReplicationTasksForRemote(
Stream<ReplicateRefUpdate> updates, String changeRef, String remote) {
return updates
- .filter(task -> changeRef.equals(task.ref()))
+ .filter(task -> task.refs().stream().anyMatch(ref -> changeRef.equals(ref)))
.filter(task -> remote.equals(task.remote()));
}
}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageMPTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageMPTest.java
index 5cfb2d0..cf5168f 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageMPTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageMPTest.java
@@ -24,6 +24,7 @@
import java.net.URISyntaxException;
import java.nio.file.FileSystem;
import java.nio.file.Path;
+import java.util.Set;
import org.eclipse.jgit.transport.URIish;
import org.junit.After;
import org.junit.Before;
@@ -36,7 +37,9 @@
protected static final URIish URISH =
ReplicationTasksStorageTest.getUrish("http://example.com/" + PROJECT + ".git");
protected static final ReplicateRefUpdate REF_UPDATE =
- ReplicateRefUpdate.create(PROJECT, REF, URISH, REMOTE);
+ ReplicateRefUpdate.create(PROJECT, Set.of(REF), URISH, REMOTE);
+ protected static final ReplicateRefUpdate STORED_REF_UPDATE =
+ ReplicateRefUpdate.create(REF_UPDATE, REF_UPDATE.sha1());
protected static final UriUpdates URI_UPDATES = getUriUpdates(REF_UPDATE);
protected ReplicationTasksStorage nodeA;
@@ -66,7 +69,7 @@
nodeA.create(REF_UPDATE);
nodeB.create(REF_UPDATE);
- assertThatStream(persistedView.streamWaiting()).containsExactly(REF_UPDATE);
+ assertThatStream(persistedView.streamWaiting()).containsExactly(STORED_REF_UPDATE);
}
@Test
@@ -74,7 +77,7 @@
nodeA.create(REF_UPDATE);
nodeB.start(URI_UPDATES);
- assertThatStream(persistedView.streamRunning()).containsExactly(REF_UPDATE);
+ assertThatStream(persistedView.streamRunning()).containsExactly(STORED_REF_UPDATE);
nodeB.finish(URI_UPDATES);
assertNoIncompleteTasks(persistedView);
@@ -86,10 +89,10 @@
nodeA.start(URI_UPDATES);
nodeA.reset(URI_UPDATES);
- assertThatStream(persistedView.streamWaiting()).containsExactly(REF_UPDATE);
+ assertThatStream(persistedView.streamWaiting()).containsExactly(STORED_REF_UPDATE);
nodeB.start(URI_UPDATES);
- assertThatStream(persistedView.streamRunning()).containsExactly(REF_UPDATE);
+ assertThatStream(persistedView.streamRunning()).containsExactly(STORED_REF_UPDATE);
nodeB.finish(URI_UPDATES);
assertNoIncompleteTasks(persistedView);
@@ -103,10 +106,10 @@
nodeB.start(URI_UPDATES);
nodeB.reset(URI_UPDATES);
- assertThatStream(persistedView.streamWaiting()).containsExactly(REF_UPDATE);
+ assertThatStream(persistedView.streamWaiting()).containsExactly(STORED_REF_UPDATE);
nodeB.start(URI_UPDATES);
- assertThatStream(persistedView.streamRunning()).containsExactly(REF_UPDATE);
+ assertThatStream(persistedView.streamRunning()).containsExactly(STORED_REF_UPDATE);
nodeB.finish(URI_UPDATES);
assertNoIncompleteTasks(persistedView);
@@ -121,7 +124,7 @@
nodeB.reset(URI_UPDATES);
nodeA.start(URI_UPDATES);
- assertThatStream(persistedView.streamRunning()).containsExactly(REF_UPDATE);
+ assertThatStream(persistedView.streamRunning()).containsExactly(STORED_REF_UPDATE);
nodeA.finish(URI_UPDATES);
assertNoIncompleteTasks(persistedView);
@@ -133,10 +136,10 @@
nodeA.start(URI_UPDATES);
nodeB.recoverAll();
- assertThatStream(persistedView.streamWaiting()).containsExactly(REF_UPDATE);
+ assertThatStream(persistedView.streamWaiting()).containsExactly(STORED_REF_UPDATE);
nodeB.start(URI_UPDATES);
- assertThatStream(persistedView.streamRunning()).containsExactly(REF_UPDATE);
+ assertThatStream(persistedView.streamRunning()).containsExactly(STORED_REF_UPDATE);
nodeA.finish(URI_UPDATES);
// Bug: https://crbug.com/gerrit/12973
@@ -153,10 +156,10 @@
nodeB.recoverAll();
nodeA.finish(URI_UPDATES);
- assertThatStream(persistedView.streamWaiting()).containsExactly(REF_UPDATE);
+ assertThatStream(persistedView.streamWaiting()).containsExactly(STORED_REF_UPDATE);
nodeB.start(URI_UPDATES);
- assertThatStream(persistedView.streamRunning()).containsExactly(REF_UPDATE);
+ assertThatStream(persistedView.streamRunning()).containsExactly(STORED_REF_UPDATE);
nodeB.finish(URI_UPDATES);
assertNoIncompleteTasks(persistedView);
@@ -169,7 +172,7 @@
nodeB.recoverAll();
nodeB.start(URI_UPDATES);
- assertThatStream(persistedView.streamRunning()).containsExactly(REF_UPDATE);
+ assertThatStream(persistedView.streamRunning()).containsExactly(STORED_REF_UPDATE);
nodeB.finish(URI_UPDATES);
ReplicationTasksStorageTest.assertNoIncompleteTasks(persistedView);
@@ -182,14 +185,14 @@
public void multipleNodesCanReplicateSameRef() {
nodeA.create(REF_UPDATE);
nodeA.start(URI_UPDATES);
- assertThatStream(persistedView.streamRunning()).containsExactly(REF_UPDATE);
+ assertThatStream(persistedView.streamRunning()).containsExactly(STORED_REF_UPDATE);
nodeA.finish(URI_UPDATES);
assertNoIncompleteTasks(persistedView);
nodeB.create(REF_UPDATE);
nodeB.start(URI_UPDATES);
- assertThatStream(persistedView.streamRunning()).containsExactly(REF_UPDATE);
+ assertThatStream(persistedView.streamRunning()).containsExactly(STORED_REF_UPDATE);
nodeB.finish(URI_UPDATES);
assertNoIncompleteTasks(persistedView);
@@ -200,16 +203,16 @@
nodeA.create(REF_UPDATE);
nodeB.create(REF_UPDATE);
- assertThat(nodeA.start(URI_UPDATES)).containsExactly(REF_UPDATE.ref());
- assertThatStream(persistedView.streamRunning()).containsExactly(REF_UPDATE);
+ assertThat(nodeA.start(URI_UPDATES)).containsExactly(REF_UPDATE.refs());
+ assertThatStream(persistedView.streamRunning()).containsExactly(STORED_REF_UPDATE);
assertThat(nodeB.start(URI_UPDATES)).isEmpty();
- assertThatStream(persistedView.streamRunning()).containsExactly(REF_UPDATE);
+ assertThatStream(persistedView.streamRunning()).containsExactly(STORED_REF_UPDATE);
}
public static UriUpdates getUriUpdates(ReplicationTasksStorage.ReplicateRefUpdate refUpdate) {
try {
- return TestUriUpdates.create(refUpdate);
+ return new TestUriUpdates(refUpdate);
} catch (URISyntaxException e) {
throw new RuntimeException("Cannot instantiate UriUpdates object", e);
}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTaskMPTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTaskMPTest.java
index 202cac9..481fa9c 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTaskMPTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTaskMPTest.java
@@ -24,6 +24,7 @@
import com.googlesource.gerrit.plugins.replication.ReplicationTasksStorage.ReplicateRefUpdate;
import java.nio.file.FileSystem;
import java.nio.file.Path;
+import java.util.Set;
import org.eclipse.jgit.transport.URIish;
import org.junit.After;
import org.junit.Before;
@@ -36,7 +37,7 @@
protected static final URIish URISH =
ReplicationTasksStorageTest.getUrish("http://example.com/" + PROJECT + ".git");
protected static final ReplicateRefUpdate REF_UPDATE =
- ReplicateRefUpdate.create(PROJECT, REF, URISH, REMOTE);
+ ReplicateRefUpdate.create(PROJECT, Set.of(REF), URISH, REMOTE);
protected FileSystem fileSystem;
protected Path storageSite;
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTaskTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTaskTest.java
index a2e5e4d..43c0b3e 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTaskTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTaskTest.java
@@ -14,17 +14,28 @@
package com.googlesource.gerrit.plugins.replication;
+import static com.google.gerrit.testing.GerritJUnit.assertThrows;
+import static java.nio.charset.StandardCharsets.UTF_8;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.hash.Hashing;
import com.google.common.jimfs.Configuration;
import com.google.common.jimfs.Jimfs;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.JsonSyntaxException;
import com.googlesource.gerrit.plugins.replication.ReplicationTasksStorage.ReplicateRefUpdate;
+import com.googlesource.gerrit.plugins.replication.ReplicationTasksStorage.ReplicateRefUpdateTypeAdapterFactory;
import com.googlesource.gerrit.plugins.replication.ReplicationTasksStorage.Task;
import java.net.URISyntaxException;
import java.nio.file.FileSystem;
import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Set;
+import org.eclipse.jgit.lib.ObjectId;
import org.eclipse.jgit.transport.URIish;
import org.junit.After;
import org.junit.Before;
@@ -36,7 +47,7 @@
protected static final String REMOTE = "myDest";
protected static final URIish URISH = getUrish("http://example.com/" + PROJECT + ".git");
protected static final ReplicateRefUpdate REF_UPDATE =
- ReplicateRefUpdate.create(PROJECT, REF, URISH, REMOTE);
+ ReplicateRefUpdate.create(PROJECT, Set.of(REF), URISH, REMOTE);
protected ReplicationTasksStorage tasksStorage;
protected FileSystem fileSystem;
@@ -200,8 +211,10 @@
@Test
public void canHaveTwoWaitingTasksForDifferentRefs() throws Exception {
- Task updateA = tasksStorage.new Task(ReplicateRefUpdate.create(PROJECT, "refA", URISH, REMOTE));
- Task updateB = tasksStorage.new Task(ReplicateRefUpdate.create(PROJECT, "refB", URISH, REMOTE));
+ Task updateA =
+ tasksStorage.new Task(ReplicateRefUpdate.create(PROJECT, Set.of("refA"), URISH, REMOTE));
+ Task updateB =
+ tasksStorage.new Task(ReplicateRefUpdate.create(PROJECT, Set.of("refB"), URISH, REMOTE));
updateA.create();
updateB.create();
assertIsWaiting(updateA);
@@ -210,8 +223,10 @@
@Test
public void canHaveTwoRunningTasksForDifferentRefs() throws Exception {
- Task updateA = tasksStorage.new Task(ReplicateRefUpdate.create(PROJECT, "refA", URISH, REMOTE));
- Task updateB = tasksStorage.new Task(ReplicateRefUpdate.create(PROJECT, "refB", URISH, REMOTE));
+ Task updateA =
+ tasksStorage.new Task(ReplicateRefUpdate.create(PROJECT, Set.of("refA"), URISH, REMOTE));
+ Task updateB =
+ tasksStorage.new Task(ReplicateRefUpdate.create(PROJECT, Set.of("refB"), URISH, REMOTE));
updateA.create();
updateB.create();
updateA.start();
@@ -226,12 +241,12 @@
tasksStorage
.new Task(
ReplicateRefUpdate.create(
- "projectA", REF, getUrish("http://example.com/projectA.git"), REMOTE));
+ "projectA", Set.of(REF), getUrish("http://example.com/projectA.git"), REMOTE));
Task updateB =
tasksStorage
.new Task(
ReplicateRefUpdate.create(
- "projectB", REF, getUrish("http://example.com/projectB.git"), REMOTE));
+ "projectB", Set.of(REF), getUrish("http://example.com/projectB.git"), REMOTE));
updateA.create();
updateB.create();
assertIsWaiting(updateA);
@@ -244,12 +259,12 @@
tasksStorage
.new Task(
ReplicateRefUpdate.create(
- "projectA", REF, getUrish("http://example.com/projectA.git"), REMOTE));
+ "projectA", Set.of(REF), getUrish("http://example.com/projectA.git"), REMOTE));
Task updateB =
tasksStorage
.new Task(
ReplicateRefUpdate.create(
- "projectB", REF, getUrish("http://example.com/projectB.git"), REMOTE));
+ "projectB", Set.of(REF), getUrish("http://example.com/projectB.git"), REMOTE));
updateA.create();
updateB.create();
updateA.start();
@@ -260,8 +275,10 @@
@Test
public void canHaveTwoWaitingTasksForDifferentRemotes() throws Exception {
- Task updateA = tasksStorage.new Task(ReplicateRefUpdate.create(PROJECT, REF, URISH, "remoteA"));
- Task updateB = tasksStorage.new Task(ReplicateRefUpdate.create(PROJECT, REF, URISH, "remoteB"));
+ Task updateA =
+ tasksStorage.new Task(ReplicateRefUpdate.create(PROJECT, Set.of(REF), URISH, "remoteA"));
+ Task updateB =
+ tasksStorage.new Task(ReplicateRefUpdate.create(PROJECT, Set.of(REF), URISH, "remoteB"));
updateA.create();
updateB.create();
assertIsWaiting(updateA);
@@ -270,8 +287,10 @@
@Test
public void canHaveTwoRunningTasksForDifferentRemotes() throws Exception {
- Task updateA = tasksStorage.new Task(ReplicateRefUpdate.create(PROJECT, REF, URISH, "remoteA"));
- Task updateB = tasksStorage.new Task(ReplicateRefUpdate.create(PROJECT, REF, URISH, "remoteB"));
+ Task updateA =
+ tasksStorage.new Task(ReplicateRefUpdate.create(PROJECT, Set.of(REF), URISH, "remoteA"));
+ Task updateB =
+ tasksStorage.new Task(ReplicateRefUpdate.create(PROJECT, Set.of(REF), URISH, "remoteB"));
updateA.create();
updateB.create();
updateA.start();
@@ -346,6 +365,130 @@
assertIsWaiting(persistedView);
}
+ @Test
+ public void writeReplicateRefUpdateTypeAdapter() throws Exception {
+ Gson gson =
+ new GsonBuilder()
+ .registerTypeAdapterFactory(new ReplicateRefUpdateTypeAdapterFactory())
+ .create();
+ ReplicateRefUpdate update =
+ ReplicateRefUpdate.create(
+ "someProject",
+ ImmutableSet.of("ref1"),
+ new URIish("git://host1/someRepo.git"),
+ "someRemote");
+ assertEquals(
+ gson.toJson(update),
+ "{\"project\":\"someProject\",\"refs\":[\"ref1\"],\"uri\":\"git://host1/someRepo.git\",\"remote\":\"someRemote\"}");
+ ReplicateRefUpdate update2 =
+ ReplicateRefUpdate.create(
+ "someProject",
+ ImmutableSet.of("ref1", "ref2"),
+ new URIish("git://host1/someRepo.git"),
+ "someRemote");
+ assertEquals(
+ gson.toJson(update2),
+ "{\"project\":\"someProject\",\"refs\":[\"ref1\",\"ref2\"],\"uri\":\"git://host1/someRepo.git\",\"remote\":\"someRemote\"}");
+ }
+
+ @Test
+ public void ReadReplicateRefUpdateTypeAdapter() throws Exception {
+ Gson gson =
+ new GsonBuilder()
+ .registerTypeAdapterFactory(new ReplicateRefUpdateTypeAdapterFactory())
+ .create();
+ ReplicateRefUpdate update =
+ ReplicateRefUpdate.create(
+ "someProject",
+ ImmutableSet.of("ref1"),
+ new URIish("git://host1/someRepo.git"),
+ "someRemote");
+ assertEquals(
+ gson.fromJson(
+ "{\"project\":\"someProject\",\"refs\":[\"ref1\"],\"uri\":\"git://host1/someRepo.git\",\"remote\":\"someRemote\"}",
+ ReplicateRefUpdate.class),
+ update);
+ ReplicateRefUpdate update2 =
+ ReplicateRefUpdate.create(
+ "someProject",
+ ImmutableSet.of("ref1", "ref2"),
+ new URIish("git://host1/someRepo.git"),
+ "someRemote");
+ ReplicateRefUpdate restoredUpdate2 =
+ gson.fromJson(
+ "{\"project\":\"someProject\",\"refs\":[\"ref1\",\"ref2\"],\"uri\":\"git://host1/someRepo.git\",\"remote\":\"someRemote\"}",
+ ReplicateRefUpdate.class);
+ // ReplicateRefUpdate.sha() might be different for the compared objects, since
+ // the order of refs() might differ.
+ assertEquals(update2.project(), restoredUpdate2.project());
+ assertEquals(update2.uri(), restoredUpdate2.uri());
+ assertEquals(update2.remote(), restoredUpdate2.remote());
+ assertEquals(update2.refs(), restoredUpdate2.refs());
+ }
+
+ @Test
+ public void ReplicateRefUpdateTypeAdapter_FailWithUnknownField() throws Exception {
+ Gson gson =
+ new GsonBuilder()
+ .registerTypeAdapterFactory(new ReplicateRefUpdateTypeAdapterFactory())
+ .create();
+ assertThrows(
+ JsonSyntaxException.class,
+ () -> gson.fromJson("{\"unknownKey\":\"someValue\"}", ReplicateRefUpdate.class));
+ }
+
+ @Test
+ public void ReadOldFormatReplicateRefUpdateTypeAdapter() throws Exception {
+ Gson gson =
+ new GsonBuilder()
+ .registerTypeAdapterFactory(new ReplicateRefUpdateTypeAdapterFactory())
+ .create();
+ ReplicateRefUpdate update =
+ ReplicateRefUpdate.create(
+ "someProject",
+ ImmutableSet.of("ref1"),
+ new URIish("git://host1/someRepo.git"),
+ "someRemote");
+ assertEquals(
+ gson.fromJson(
+ "{\"project\":\"someProject\",\"ref\":\"ref1\",\"uri\":\"git://host1/someRepo.git\",\"remote\":\"someRemote\"}",
+ ReplicateRefUpdate.class),
+ update);
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void schedulingTaskFromOldFormatTasksIsSuccessful() throws Exception {
+ Gson gson =
+ new GsonBuilder()
+ .registerTypeAdapterFactory(new ReplicateRefUpdateTypeAdapterFactory())
+ .create();
+ ReplicateRefUpdate update =
+ gson.fromJson(
+ "{\"project\":\"someProject\",\"ref\":\"ref1\",\"uri\":\"git://host1/someRepo.git\",\"remote\":\"someRemote\"}",
+ ReplicateRefUpdate.class);
+
+ String oldTaskKey =
+ ObjectId.fromRaw(
+ Hashing.sha1()
+ .hashString("someProject\nref1\ngit://host1/someRepo.git\nsomeRemote", UTF_8)
+ .asBytes())
+ .name();
+ String json = gson.toJson(update) + "\n";
+ Path tmp =
+ Files.createTempFile(
+ Files.createDirectories(fileSystem.getPath("replication_site").resolve("building")),
+ oldTaskKey,
+ null);
+ Files.write(tmp, json.getBytes(UTF_8));
+
+ Task task = tasksStorage.new Task(update);
+ task.rename(tmp, task.waiting);
+
+ task.start();
+ assertIsRunning(task);
+ }
+
protected static void assertIsWaiting(Task task) {
assertTrue(task.isWaiting());
}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTest.java
index bb47ef8..6e02573 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTest.java
@@ -28,6 +28,7 @@
import java.net.URISyntaxException;
import java.nio.file.FileSystem;
import java.nio.file.Path;
+import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.eclipse.jgit.transport.URIish;
@@ -38,10 +39,17 @@
public class ReplicationTasksStorageTest {
protected static final String PROJECT = "myProject";
protected static final String REF = "myRef";
+ protected static final String REF_2 = "myRef2";
protected static final String REMOTE = "myDest";
protected static final URIish URISH = getUrish("http://example.com/" + PROJECT + ".git");
protected static final ReplicateRefUpdate REF_UPDATE =
- ReplicateRefUpdate.create(PROJECT, REF, URISH, REMOTE);
+ ReplicateRefUpdate.create(PROJECT, Set.of(REF), URISH, REMOTE);
+ protected static final ReplicateRefUpdate STORED_REF_UPDATE =
+ ReplicateRefUpdate.create(REF_UPDATE, REF_UPDATE.sha1());
+ protected static final ReplicateRefUpdate REFS_UPDATE =
+ ReplicateRefUpdate.create(PROJECT, Set.of(REF, REF_2), URISH, REMOTE);
+ protected static final ReplicateRefUpdate STORED_REFS_UPDATE =
+ ReplicateRefUpdate.create(REFS_UPDATE, REFS_UPDATE.sha1());
protected ReplicationTasksStorage storage;
protected FileSystem fileSystem;
@@ -53,7 +61,7 @@
fileSystem = Jimfs.newFileSystem(Configuration.unix());
storageSite = fileSystem.getPath("replication_site");
storage = new ReplicationTasksStorage(storageSite);
- uriUpdates = TestUriUpdates.create(REF_UPDATE);
+ uriUpdates = new TestUriUpdates(REF_UPDATE);
}
@After
@@ -70,7 +78,7 @@
@Test
public void canListWaitingUpdate() throws Exception {
storage.create(REF_UPDATE);
- assertThatStream(storage.streamWaiting()).containsExactly(REF_UPDATE);
+ assertThatStream(storage.streamWaiting()).containsExactly(STORED_REF_UPDATE);
}
@Test
@@ -85,10 +93,20 @@
@Test
public void canStartWaitingUpdate() throws Exception {
storage.create(REF_UPDATE);
- assertThat(storage.start(uriUpdates)).containsExactly(REF_UPDATE.ref());
+ assertThat(storage.start(uriUpdates)).containsExactly(REF_UPDATE.refs());
assertThatStream(storage.streamWaiting()).isEmpty();
assertFalse(storage.isWaiting(uriUpdates));
- assertThatStream(storage.streamRunning()).containsExactly(REF_UPDATE);
+ assertThatStream(storage.streamRunning()).containsExactly(STORED_REF_UPDATE);
+ }
+
+ @Test
+ public void canStartWaitingUpdateWithMultipleRefs() throws Exception {
+ TestUriUpdates updates = new TestUriUpdates(REFS_UPDATE);
+ storage.create(REFS_UPDATE);
+ assertThat(storage.start(updates)).containsExactly(REFS_UPDATE.refs());
+ assertThatStream(storage.streamWaiting()).isEmpty();
+ assertFalse(storage.isWaiting(updates));
+ assertThatStream(storage.streamRunning()).containsExactly(STORED_REFS_UPDATE);
}
@Test
@@ -107,14 +125,14 @@
assertThatStream(persistedView.streamWaiting()).isEmpty();
storage.create(REF_UPDATE);
- assertThatStream(storage.streamWaiting()).containsExactly(REF_UPDATE);
- assertThatStream(persistedView.streamWaiting()).containsExactly(REF_UPDATE);
+ assertThatStream(storage.streamWaiting()).containsExactly(STORED_REF_UPDATE);
+ assertThatStream(persistedView.streamWaiting()).containsExactly(STORED_REF_UPDATE);
storage.start(uriUpdates);
assertThatStream(storage.streamWaiting()).isEmpty();
assertThatStream(persistedView.streamWaiting()).isEmpty();
- assertThatStream(storage.streamRunning()).containsExactly(REF_UPDATE);
- assertThatStream(persistedView.streamRunning()).containsExactly(REF_UPDATE);
+ assertThatStream(storage.streamRunning()).containsExactly(STORED_REF_UPDATE);
+ assertThatStream(persistedView.streamRunning()).containsExactly(STORED_REF_UPDATE);
storage.finish(uriUpdates);
assertThatStream(storage.streamRunning()).isEmpty();
@@ -126,7 +144,7 @@
String key = storage.create(REF_UPDATE);
String secondKey = storage.create(REF_UPDATE);
assertEquals(key, secondKey);
- assertThatStream(storage.streamWaiting()).containsExactly(REF_UPDATE);
+ assertThatStream(storage.streamWaiting()).containsExactly(STORED_REF_UPDATE);
}
@Test
@@ -134,7 +152,7 @@
ReplicateRefUpdate updateB =
ReplicateRefUpdate.create(
PROJECT,
- REF,
+ Set.of(REF),
getUrish("ssh://example.com/" + PROJECT + ".git"), // uses ssh not http
REMOTE);
@@ -142,7 +160,7 @@
String keyB = storage.create(updateB);
assertThatStream(storage.streamWaiting()).hasSize(2);
assertTrue(storage.isWaiting(uriUpdates));
- assertTrue(storage.isWaiting(TestUriUpdates.create(updateB)));
+ assertTrue(storage.isWaiting(new TestUriUpdates(updateB)));
assertNotEquals(keyA, keyB);
}
@@ -155,7 +173,8 @@
"project/with/a/strange/name key=a-value=1, --option1 \"OPTION_VALUE_1\" --option-2 <option_VALUE-2> --option-without-value";
Project.NameKey project = Project.nameKey(strangeValidName);
URIish expanded = Destination.getURI(template, project, "slash", false);
- ReplicateRefUpdate update = ReplicateRefUpdate.create(strangeValidName, REF, expanded, REMOTE);
+ ReplicateRefUpdate update =
+ ReplicateRefUpdate.create(strangeValidName, Set.of(REF), expanded, REMOTE);
storage.create(update);
assertThat(new URIish(update.uri())).isEqualTo(expanded);
@@ -170,20 +189,21 @@
ReplicateRefUpdate updateB =
ReplicateRefUpdate.create(
PROJECT,
- REF,
+ Set.of(REF),
getUrish("ssh://example.com/" + PROJECT + ".git"), // uses ssh not http
REMOTE);
- UriUpdates uriUpdatesB = TestUriUpdates.create(updateB);
+ UriUpdates uriUpdatesB = new TestUriUpdates(updateB);
storage.create(REF_UPDATE);
storage.create(updateB);
+ ReplicateRefUpdate storedUpdateB = ReplicateRefUpdate.create(updateB, updateB.sha1());
storage.start(uriUpdates);
- assertThatStream(storage.streamWaiting()).containsExactly(updateB);
- assertThatStream(storage.streamRunning()).containsExactly(REF_UPDATE);
+ assertThatStream(storage.streamWaiting()).containsExactly(storedUpdateB);
+ assertThatStream(storage.streamRunning()).containsExactly(STORED_REF_UPDATE);
storage.start(uriUpdatesB);
assertThatStream(storage.streamWaiting()).isEmpty();
- assertThatStream(storage.streamRunning()).containsExactly(REF_UPDATE, updateB);
+ assertThatStream(storage.streamRunning()).containsExactly(STORED_REF_UPDATE, storedUpdateB);
}
@Test
@@ -191,17 +211,18 @@
ReplicateRefUpdate updateB =
ReplicateRefUpdate.create(
PROJECT,
- REF,
+ Set.of(REF),
getUrish("ssh://example.com/" + PROJECT + ".git"), // uses ssh not http
REMOTE);
- UriUpdates uriUpdatesB = TestUriUpdates.create(updateB);
+ UriUpdates uriUpdatesB = new TestUriUpdates(updateB);
storage.create(REF_UPDATE);
storage.create(updateB);
storage.start(uriUpdates);
storage.start(uriUpdatesB);
storage.finish(uriUpdates);
- assertThatStream(storage.streamRunning()).containsExactly(updateB);
+ assertThatStream(storage.streamRunning())
+ .containsExactly(ReplicateRefUpdate.create(updateB, updateB.sha1()));
storage.finish(uriUpdatesB);
assertThatStream(storage.streamRunning()).isEmpty();
@@ -212,7 +233,7 @@
ReplicateRefUpdate updateB =
ReplicateRefUpdate.create(
PROJECT,
- REF,
+ Set.of(REF),
getUrish("ssh://example.com/" + PROJECT + ".git"), // uses ssh not http
REMOTE);
@@ -222,35 +243,38 @@
storage.create(updateB);
assertThatStream(storage.streamWaiting()).hasSize(2);
assertTrue(storage.isWaiting(uriUpdates));
- assertTrue(storage.isWaiting(TestUriUpdates.create(updateB)));
+ assertTrue(storage.isWaiting(new TestUriUpdates(updateB)));
}
@Test
public void canCreateMulipleRefsForSameUri() throws Exception {
- ReplicateRefUpdate refA = ReplicateRefUpdate.create(PROJECT, "refA", URISH, REMOTE);
- ReplicateRefUpdate refB = ReplicateRefUpdate.create(PROJECT, "refB", URISH, REMOTE);
+ ReplicateRefUpdate refA = ReplicateRefUpdate.create(PROJECT, Set.of("refA"), URISH, REMOTE);
+ ReplicateRefUpdate refB = ReplicateRefUpdate.create(PROJECT, Set.of("refB"), URISH, REMOTE);
String keyA = storage.create(refA);
String keyB = storage.create(refB);
assertThatStream(storage.streamWaiting()).hasSize(2);
assertNotEquals(keyA, keyB);
- assertTrue(storage.isWaiting(TestUriUpdates.create(refA)));
- assertTrue(storage.isWaiting(TestUriUpdates.create(refB)));
+ assertTrue(storage.isWaiting(new TestUriUpdates(refA)));
+ assertTrue(storage.isWaiting(new TestUriUpdates(refB)));
}
@Test
public void canFinishMulipleRefsForSameUri() throws Exception {
- ReplicateRefUpdate refUpdateA = ReplicateRefUpdate.create(PROJECT, "refA", URISH, REMOTE);
- ReplicateRefUpdate refUpdateB = ReplicateRefUpdate.create(PROJECT, "refB", URISH, REMOTE);
- UriUpdates uriUpdatesA = TestUriUpdates.create(refUpdateA);
- UriUpdates uriUpdatesB = TestUriUpdates.create(refUpdateB);
+ ReplicateRefUpdate refUpdateA =
+ ReplicateRefUpdate.create(PROJECT, Set.of("refA"), URISH, REMOTE);
+ ReplicateRefUpdate refUpdateB =
+ ReplicateRefUpdate.create(PROJECT, Set.of("refB"), URISH, REMOTE);
+ UriUpdates uriUpdatesA = new TestUriUpdates(refUpdateA);
+ UriUpdates uriUpdatesB = new TestUriUpdates(refUpdateB);
storage.create(refUpdateA);
storage.create(refUpdateB);
storage.start(uriUpdatesA);
storage.start(uriUpdatesB);
storage.finish(uriUpdatesA);
- assertThatStream(storage.streamRunning()).containsExactly(refUpdateB);
+ assertThatStream(storage.streamRunning())
+ .containsExactly(ReplicateRefUpdate.create(refUpdateB, refUpdateB.sha1()));
storage.finish(uriUpdatesB);
assertThatStream(storage.streamRunning()).isEmpty();
@@ -262,7 +286,7 @@
storage.start(uriUpdates);
storage.reset(uriUpdates);
- assertThatStream(storage.streamWaiting()).containsExactly(REF_UPDATE);
+ assertThatStream(storage.streamWaiting()).containsExactly(STORED_REF_UPDATE);
assertThatStream(storage.streamRunning()).isEmpty();
}
@@ -273,7 +297,7 @@
storage.reset(uriUpdates);
storage.start(uriUpdates);
- assertThatStream(storage.streamRunning()).containsExactly(REF_UPDATE);
+ assertThatStream(storage.streamRunning()).containsExactly(STORED_REF_UPDATE);
assertThatStream(storage.streamWaiting()).isEmpty();
assertFalse(storage.isWaiting(uriUpdates));
@@ -293,7 +317,7 @@
storage.start(uriUpdates);
storage.recoverAll();
- assertThatStream(storage.streamWaiting()).containsExactly(REF_UPDATE);
+ assertThatStream(storage.streamWaiting()).containsExactly(STORED_REF_UPDATE);
assertThatStream(storage.streamRunning()).isEmpty();
assertTrue(storage.isWaiting(uriUpdates));
}
@@ -305,7 +329,7 @@
storage.recoverAll();
storage.start(uriUpdates);
- assertThatStream(storage.streamRunning()).containsExactly(REF_UPDATE);
+ assertThatStream(storage.streamRunning()).containsExactly(STORED_REF_UPDATE);
assertThatStream(storage.streamWaiting()).isEmpty();
assertFalse(storage.isWaiting(uriUpdates));
@@ -318,17 +342,18 @@
ReplicateRefUpdate updateB =
ReplicateRefUpdate.create(
PROJECT,
- REF,
+ Set.of(REF),
getUrish("ssh://example.com/" + PROJECT + ".git"), // uses ssh not http
REMOTE);
- UriUpdates uriUpdatesB = TestUriUpdates.create(updateB);
+ UriUpdates uriUpdatesB = new TestUriUpdates(updateB);
storage.create(REF_UPDATE);
storage.create(updateB);
storage.start(uriUpdates);
storage.start(uriUpdatesB);
storage.recoverAll();
- assertThatStream(storage.streamWaiting()).containsExactly(REF_UPDATE, updateB);
+ assertThatStream(storage.streamWaiting())
+ .containsExactly(STORED_REF_UPDATE, ReplicateRefUpdate.create(updateB, updateB.sha1()));
}
@Test
@@ -336,22 +361,23 @@
ReplicateRefUpdate updateB =
ReplicateRefUpdate.create(
PROJECT,
- REF,
+ Set.of(REF),
getUrish("ssh://example.com/" + PROJECT + ".git"), // uses ssh not http
REMOTE);
- UriUpdates uriUpdatesB = TestUriUpdates.create(updateB);
+ UriUpdates uriUpdatesB = new TestUriUpdates(updateB);
storage.create(REF_UPDATE);
storage.create(updateB);
storage.start(uriUpdates);
storage.start(uriUpdatesB);
storage.recoverAll();
+ ReplicateRefUpdate storedUpdateB = ReplicateRefUpdate.create(updateB, updateB.sha1());
storage.start(uriUpdates);
- assertThatStream(storage.streamRunning()).containsExactly(REF_UPDATE);
- assertThatStream(storage.streamWaiting()).containsExactly(updateB);
+ assertThatStream(storage.streamRunning()).containsExactly(STORED_REF_UPDATE);
+ assertThatStream(storage.streamWaiting()).containsExactly(storedUpdateB);
storage.start(uriUpdatesB);
- assertThatStream(storage.streamRunning()).containsExactly(REF_UPDATE, updateB);
+ assertThatStream(storage.streamRunning()).containsExactly(STORED_REF_UPDATE, storedUpdateB);
assertThatStream(storage.streamWaiting()).isEmpty();
storage.finish(uriUpdates);
@@ -378,10 +404,10 @@
ReplicateRefUpdate updateB =
ReplicateRefUpdate.create(
PROJECT,
- REF,
+ Set.of(REF),
getUrish("ssh://example.com/" + PROJECT + ".git"), // uses ssh not http
REMOTE);
- UriUpdates uriUpdatesB = TestUriUpdates.create(updateB);
+ UriUpdates uriUpdatesB = new TestUriUpdates(updateB);
storage.create(REF_UPDATE);
storage.create(updateB);
storage.start(uriUpdates);
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/TestDispatcher.java b/src/test/java/com/googlesource/gerrit/plugins/replication/TestDispatcher.java
index 901200b..5881ea7 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/TestDispatcher.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/TestDispatcher.java
@@ -22,14 +22,14 @@
import com.google.gerrit.server.events.EventDispatcher;
import com.google.gerrit.server.events.ProjectEvent;
import com.google.gerrit.server.events.RefEvent;
-import java.util.LinkedList;
+import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
public class TestDispatcher implements EventDispatcher {
- private final List<ProjectEvent> projectEvents = new LinkedList<>();
- private final List<RefEvent> refEvents = new LinkedList<>();
- private final List<Event> events = new LinkedList<>();
+ private final List<ProjectEvent> projectEvents = new ArrayList<>();
+ private final List<RefEvent> refEvents = new ArrayList<>();
+ private final List<Event> events = new ArrayList<>();
@Override
public void postEvent(Change change, ChangeEvent event) {} // Not used in replication
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/TestUriUpdates.java b/src/test/java/com/googlesource/gerrit/plugins/replication/TestUriUpdates.java
index f61114e..f6eec83 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/TestUriUpdates.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/TestUriUpdates.java
@@ -14,38 +14,43 @@
package com.googlesource.gerrit.plugins.replication;
-import com.google.auto.value.AutoValue;
+import com.google.common.collect.ImmutableSet;
import com.google.gerrit.entities.Project;
import com.googlesource.gerrit.plugins.replication.ReplicationTasksStorage.ReplicateRefUpdate;
import java.net.URISyntaxException;
-import java.util.Collections;
import java.util.Set;
import org.eclipse.jgit.transport.URIish;
-@AutoValue
-public abstract class TestUriUpdates implements UriUpdates {
- public static TestUriUpdates create(ReplicateRefUpdate update) throws URISyntaxException {
- return create(
- Project.nameKey(update.project()),
- new URIish(update.uri()),
- update.remote(),
- Collections.singleton(update.ref()));
- }
+public class TestUriUpdates implements UriUpdates {
+ private final Project.NameKey project;
+ private final URIish uri;
+ private final String remote;
+ private final Set<ImmutableSet<String>> refs;
- public static TestUriUpdates create(
- Project.NameKey project, URIish uri, String remote, Set<String> refs) {
- return new AutoValue_TestUriUpdates(project, uri, remote, refs);
+ public TestUriUpdates(ReplicateRefUpdate update) throws URISyntaxException {
+ project = Project.nameKey(update.project());
+ uri = new URIish(update.uri());
+ remote = update.remote();
+ refs = Set.of(update.refs());
}
@Override
- public abstract Project.NameKey getProjectNameKey();
+ public Project.NameKey getProjectNameKey() {
+ return project;
+ }
@Override
- public abstract URIish getURI();
+ public URIish getURI() {
+ return uri;
+ }
@Override
- public abstract String getRemoteName();
+ public String getRemoteName() {
+ return remote;
+ }
@Override
- public abstract Set<String> getRefs();
+ public Set<ImmutableSet<String>> getRefs() {
+ return refs;
+ }
}
