Call synchronous fetch for all refs in batch if any is marked as sync
The list of refs provided to the `batch-ref` endpoint can include both
async and sync refs. If any ref in batch matches `replication.syncRefs`
param all refs are fetched synchronously.
By fetching all references synchronously when any reference matches the
replication.syncRefs parameter, we ensure that the entire batch
operation is atomic. This helps maintain the consistency and integrity of
changes.
Bug: Issue 305984261
Change-Id: I66a5c7da6aa3a85eace4b8e34556f3e7502c6b0c
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClient.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClient.java
index 8531b0a..e682b71 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClient.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClient.java
@@ -17,7 +17,6 @@
import static com.google.common.net.HttpHeaders.CONTENT_TYPE;
import static com.google.gson.FieldNamingPolicy.LOWER_CASE_WITH_UNDERSCORES;
import static java.util.Objects.requireNonNull;
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
import com.google.common.base.Strings;
import com.google.common.flogger.FluentLogger;
@@ -46,7 +45,6 @@
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.List;
-import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
import org.apache.http.HttpHeaders;
@@ -153,62 +151,27 @@
return executeRequest(post, bearerTokenProvider.get(), targetUri);
}
- private Map<Boolean, List<String>> partitionRefsToAsyncAndSync(List<String> refsInBatch) {
- return refsInBatch.stream()
- .collect(
- Collectors.partitioningBy(
- refName -> !syncRefsFilter.match(refName),
- Collectors.mapping(
- refName ->
- String.format(
- "{\"label\":\"%s\", \"ref_name\": \"%s\", \"async\":%s}",
- instanceId, refName, !syncRefsFilter.match(refName)),
- Collectors.toList())));
+ private Boolean containsSyncFetchRef(List<String> refsInBatch) {
+ return refsInBatch.stream().anyMatch(syncRefsFilter::match);
}
@Override
public HttpResult callBatchFetch(
NameKey project, List<String> refsInBatch, URIish targetUri, long startTimeNanos)
throws IOException {
- Map<Boolean, List<String>> refsPartitionedInAsyncAndSync =
- partitionRefsToAsyncAndSync(refsInBatch);
- boolean async = true;
- List<String> asyncRefs = refsPartitionedInAsyncAndSync.get(async);
- List<String> syncRefs = refsPartitionedInAsyncAndSync.get(!async);
-
- if (asyncRefs.isEmpty() && syncRefs.isEmpty()) {
- throw new IllegalArgumentException(
- "At least one ref should be provided during a batch-fetch operation");
- }
+ boolean callAsync = !containsSyncFetchRef(refsInBatch);
+ String msgBody =
+ refsInBatch.stream()
+ .map(
+ refName ->
+ String.format(
+ "{\"label\":\"%s\", \"ref_name\": \"%s\", \"async\":%s}",
+ instanceId, refName, callAsync))
+ .collect(Collectors.joining(","));
String url = formatUrl(targetUri.toString(), project, "batch-fetch");
-
- if (asyncRefs.isEmpty()) {
- HttpPost syncPost =
- createPostRequest(url, "[" + String.join(",", syncRefs) + "]", startTimeNanos);
- return executeRequest(syncPost, bearerTokenProvider.get(), targetUri);
- }
- if (syncRefs.isEmpty()) {
- HttpPost asyncPost =
- createPostRequest(url, "[" + String.join(",", asyncRefs) + "]", startTimeNanos);
- return executeRequest(asyncPost, bearerTokenProvider.get(), targetUri);
- }
-
- // first execute for async refs, then for sync
- HttpPost asyncPost =
- createPostRequest(url, "[" + String.join(",", asyncRefs) + "]", startTimeNanos);
- HttpResult asyncResult = executeRequest(asyncPost, bearerTokenProvider.get(), targetUri);
-
- if (asyncResult.isSuccessful()) {
- HttpPost syncPost =
- createPostRequest(
- url,
- "[" + String.join(",", syncRefs) + "]",
- MILLISECONDS.toNanos(System.currentTimeMillis()));
- return executeRequest(syncPost, bearerTokenProvider.get(), targetUri);
- }
-
- return asyncResult;
+ HttpPost post = createPostRequest(url, "[" + msgBody + "]", startTimeNanos);
+ return executeRequest(post, bearerTokenProvider.get(), targetUri);
}
private HttpPost createPostRequest(String url, String msgBody, long startTimeNanos) {
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md
index ed0c5e7..acc636b 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md
@@ -615,7 +615,11 @@
over, this value should be set to `true` to leverage the
performance improvements introduced by the `batch-apply-object` API.
- By default, false.
+> *NOTE*: if any ref from a single batch matches `replication.syncRefs`
+> filter, all refs in that batch are going to be fetched synchronously as
+> a single git fetch operation.
+>
+> By default, false.
Directory `replication`
--------------------
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClientBase.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClientBase.java
index 087920b..e9e28ed 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClientBase.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClientBase.java
@@ -360,8 +360,7 @@
}
@Test
- public void shouldExecuteOneFetchCallForAsyncAndOneForSyncRefsDuringBatchFetch()
- throws Exception {
+ public void shouldMarkAllRefsAsSyncWhenTheBatchContainsAtLeastOneSyncRef() throws Exception {
when(config.getStringList("replication", null, "syncRefs"))
.thenReturn(new String[] {"^refs\\/heads\\/test"});
@@ -380,48 +379,20 @@
source);
objectUnderTest.callBatchFetch(Project.nameKey("test_repo"), refs, new URIish(api));
- verify(httpClient, times(2)).execute(httpPostCaptor.capture(), any());
-
- List<HttpPost> httpPosts = httpPostCaptor.getAllValues();
- String expectedSyncPayload =
- "[{\"label\":\"Replication\", \"ref_name\": \""
- + refs.get(1)
- + "\", \"async\":false}"
- + "]";
- String expectedAsyncPayload =
- "[{\"label\":\"Replication\", \"ref_name\": \"" + refName + "\", \"async\":true}]";
-
- assertThat(readPayload(httpPosts.get(0))).isEqualTo(expectedAsyncPayload);
- assertThat(readPayload(httpPosts.get(1))).isEqualTo(expectedSyncPayload);
- }
-
- @Test
- public void shouldNotExecuteSyncFetchCallWhenAsyncCallFailsDuringBatchFetch() throws Exception {
- when(config.getStringList("replication", null, "syncRefs"))
- .thenReturn(new String[] {"^refs\\/heads\\/test"});
- when(httpClient.execute(any(), any())).thenReturn(new HttpResult(500, Optional.of("BOOM")));
- syncRefsFilter = new SyncRefsFilter(replicationConfig);
- String testRef = RefNames.REFS_HEADS + "test";
- List<String> refs = List.of(refName, testRef);
- objectUnderTest =
- new FetchRestApiClient(
- credentials,
- httpClientFactory,
- replicationConfig,
- syncRefsFilter,
- pluginName,
- instanceId,
- bearerTokenProvider,
- source);
- objectUnderTest.callBatchFetch(Project.nameKey("test_repo"), refs, new URIish(api));
-
verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any());
- HttpPost httpPost = httpPostCaptor.getValue();
- String expectedAsyncPayload =
- "[{\"label\":\"Replication\", \"ref_name\": \"" + refName + "\", \"async\":true}]";
+ HttpPost httpPosts = httpPostCaptor.getValue();
+ String expectedSyncPayload =
+ "["
+ + "{\"label\":\"Replication\", \"ref_name\": \""
+ + refName
+ + "\", \"async\":false},"
+ + "{\"label\":\"Replication\", \"ref_name\": \""
+ + refs.get(1)
+ + "\", \"async\":false}"
+ + "]";
- assertThat(readPayload(httpPost)).isEqualTo(expectedAsyncPayload);
+ assertThat(readPayload(httpPosts)).isEqualTo(expectedSyncPayload);
}
@Test
