Merge branch 'stable-3.2' into master * stable-3.2: Stop occupying threads during retry phase Fix issue with indexing tasks queue Change-Id: I9d0c1a3275d7501e4dc833c1f4dff00ed7d63a53
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/Forwarder.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/Forwarder.java index 49cdc8b..bb47f11 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/Forwarder.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/Forwarder.java
@@ -15,6 +15,7 @@ package com.ericsson.gerrit.plugins.highavailability.forwarder; import com.google.gerrit.server.events.Event; +import java.util.concurrent.CompletableFuture; /** Forward indexing, stream events and cache evictions to the other master */ public interface Forwarder { @@ -24,9 +25,10 @@ * * @param accountId the account to index. * @param indexEvent the details of the index event. - * @return true if successful, otherwise false. + * @return {@link CompletableFuture} of true if successful, otherwise {@link CompletableFuture} of + * false. */ - boolean indexAccount(int accountId, IndexEvent indexEvent); + CompletableFuture<Boolean> indexAccount(int accountId, IndexEvent indexEvent); /** * Forward a change indexing event to the other master. @@ -34,67 +36,75 @@ * @param projectName the project of the change to index. * @param changeId the change to index. * @param indexEvent the details of the index event. - * @return true if successful, otherwise false. + * @return {@link CompletableFuture} of true if successful, otherwise {@link CompletableFuture} of + * false. */ - boolean indexChange(String projectName, int changeId, IndexEvent indexEvent); + CompletableFuture<Boolean> indexChange(String projectName, int changeId, IndexEvent indexEvent); /** * Forward a delete change from index event to the other master. * * @param changeId the change to remove from the index. * @param indexEvent the details of the index event. - * @return rue if successful, otherwise false. + * @return {@link CompletableFuture} of true if successful, otherwise {@link CompletableFuture} of + * false. */ - boolean deleteChangeFromIndex(int changeId, IndexEvent indexEvent); + CompletableFuture<Boolean> deleteChangeFromIndex(int changeId, IndexEvent indexEvent); /** * Forward a group indexing event to the other master. * * @param uuid the group to index. * @param indexEvent the details of the index event. - * @return true if successful, otherwise false. + * @return {@link CompletableFuture} of true if successful, otherwise {@link CompletableFuture} of + * false. */ - boolean indexGroup(String uuid, IndexEvent indexEvent); + CompletableFuture<Boolean> indexGroup(String uuid, IndexEvent indexEvent); /** * Forward a project indexing event to the other master. * * @param projectName the project to index. * @param indexEvent the details of the index event. - * @return true if successful, otherwise false. + * @return {@link CompletableFuture} of true if successful, otherwise {@link CompletableFuture} of + * false. */ - boolean indexProject(String projectName, IndexEvent indexEvent); + CompletableFuture<Boolean> indexProject(String projectName, IndexEvent indexEvent); /** * Forward a stream event to the other master. * * @param event the event to forward. - * @return true if successful, otherwise false. + * @return {@link CompletableFuture} of true if successful, otherwise {@link CompletableFuture} of + * false. */ - boolean send(Event event); + CompletableFuture<Boolean> send(Event event); /** * Forward a cache eviction event to the other master. * * @param cacheName the name of the cache to evict an entry from. * @param key the key identifying the entry to evict from the cache. - * @return true if successful, otherwise false. + * @return {@link CompletableFuture} of true if successful, otherwise {@link CompletableFuture} of + * false. */ - boolean evict(String cacheName, Object key); + CompletableFuture<Boolean> evict(String cacheName, Object key); /** * Forward an addition to the project list cache to the other master. * * @param projectName the name of the project to add to the project list cache - * @return true if successful, otherwise false. + * @return {@link CompletableFuture} of true if successful, otherwise {@link CompletableFuture} of + * false. */ - boolean addToProjectList(String projectName); + CompletableFuture<Boolean> addToProjectList(String projectName); /** * Forward a removal from the project list cache to the other master. * * @param projectName the name of the project to remove from the project list cache - * @return true if successful, otherwise false. + * @return {@link CompletableFuture} of true if successful, otherwise {@link CompletableFuture} of + * false. */ - boolean removeFromProjectList(String projectName); + CompletableFuture<Boolean> removeFromProjectList(String projectName); }
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarder.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarder.java index 3ab9878..d404ab2 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarder.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarder.java
@@ -30,10 +30,8 @@ import com.google.inject.Inject; import com.google.inject.Provider; import java.io.IOException; -import java.util.List; import java.util.Set; import java.util.concurrent.CompletableFuture; -import java.util.stream.Collectors; import javax.net.ssl.SSLException; import org.apache.http.HttpException; import org.apache.http.client.ClientProtocolException; @@ -51,6 +49,7 @@ private final Configuration cfg; private final Provider<Set<PeerInfo>> peerInfoProvider; private final Gson gson; + private final RestForwarderScheduler scheduler; @Inject RestForwarder( @@ -58,21 +57,24 @@ @PluginName String pluginName, Configuration cfg, Provider<Set<PeerInfo>> peerInfoProvider, - @EventGson Gson gson) { + @EventGson Gson gson, + RestForwarderScheduler scheduler) { this.httpSession = httpClient; this.pluginRelativePath = Joiner.on("/").join("plugins", pluginName); this.cfg = cfg; this.peerInfoProvider = peerInfoProvider; this.gson = gson; + this.scheduler = scheduler; } @Override - public boolean indexAccount(final int accountId, IndexEvent event) { + public CompletableFuture<Boolean> indexAccount(final int accountId, IndexEvent event) { return execute(RequestMethod.POST, "index account", "index/account", accountId, event); } @Override - public boolean indexChange(String projectName, int changeId, IndexEvent event) { + public CompletableFuture<Boolean> indexChange( + String projectName, int changeId, IndexEvent event) { return execute( RequestMethod.POST, "index change", @@ -82,13 +84,13 @@ } @Override - public boolean deleteChangeFromIndex(final int changeId, IndexEvent event) { + public CompletableFuture<Boolean> deleteChangeFromIndex(final int changeId, IndexEvent event) { return execute( RequestMethod.DELETE, "delete change", "index/change", buildIndexEndpoint(changeId), event); } @Override - public boolean indexGroup(final String uuid, IndexEvent event) { + public CompletableFuture<Boolean> indexGroup(final String uuid, IndexEvent event) { return execute(RequestMethod.POST, "index group", "index/group", uuid, event); } @@ -102,24 +104,24 @@ } @Override - public boolean indexProject(String projectName, IndexEvent event) { + public CompletableFuture<Boolean> indexProject(String projectName, IndexEvent event) { return execute( RequestMethod.POST, "index project", "index/project", Url.encode(projectName), event); } @Override - public boolean send(final Event event) { + public CompletableFuture<Boolean> send(final Event event) { return execute(RequestMethod.POST, "send event", "event", event.type, event); } @Override - public boolean evict(final String cacheName, final Object key) { + public CompletableFuture<Boolean> evict(final String cacheName, final Object key) { String json = gson.toJson(key); return execute(RequestMethod.POST, "invalidate cache " + cacheName, "cache", cacheName, json); } @Override - public boolean addToProjectList(String projectName) { + public CompletableFuture<Boolean> addToProjectList(String projectName) { return execute( RequestMethod.POST, "Update project_list, add ", @@ -128,7 +130,7 @@ } @Override - public boolean removeFromProjectList(String projectName) { + public CompletableFuture<Boolean> removeFromProjectList(String projectName) { return execute( RequestMethod.DELETE, "Update project_list, remove ", @@ -140,19 +142,19 @@ return Joiner.on("/").join("cache", Constants.PROJECT_LIST); } - private boolean execute(RequestMethod method, String action, String endpoint, Object id) { + private CompletableFuture<Boolean> execute( + RequestMethod method, String action, String endpoint, Object id) { return execute(method, action, endpoint, id, null); } - private boolean execute( + private CompletableFuture<Boolean> execute( RequestMethod method, String action, String endpoint, Object id, Object payload) { - List<CompletableFuture<Boolean>> futures = - peerInfoProvider.get().stream() - .map(peer -> createRequest(method, peer, action, endpoint, id, payload)) - .map(request -> CompletableFuture.supplyAsync(request::execute)) - .collect(Collectors.toList()); - CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); - return futures.stream().allMatch(CompletableFuture::join); + return peerInfoProvider.get().stream() + .map(peer -> createRequest(method, peer, action, endpoint, id, payload)) + .map(scheduler::execute) + .reduce( + CompletableFuture.completedFuture(true), + (a, b) -> a.thenCombine(b, (left, right) -> left && right)); } private Request createRequest( @@ -178,7 +180,7 @@ }; } - private abstract class Request { + protected abstract class Request { private final String action; private final Object key; private final String destination; @@ -191,42 +193,36 @@ this.destination = destination; } - boolean execute() { - log.atFine().log("Executing %s %s towards %s", action, key, destination); - for (; ; ) { - try { - execCnt++; - tryOnce(); - log.atFine().log("%s %s towards %s OK", action, key, destination); - return true; - } catch (ForwardingException e) { - int maxTries = cfg.http().maxTries(); - log.atFine().withCause(e).log( - "Failed to %s %s on %s [%d/%d]", action, key, destination, execCnt, maxTries); - if (!e.isRecoverable()) { - log.atSevere().withCause(e).log( - "%s %s towards %s failed with unrecoverable error; giving up", - action, key, destination); - return false; - } - if (execCnt >= maxTries) { - log.atSevere().log( - "Failed to %s %s on %s after %d tries; giving up", - action, key, destination, maxTries); - return false; - } + @Override + public String toString() { + return String.format("%s:%s => %s (try #%d)", action, key, destination, execCnt); + } - log.atFine().log("Retrying to %s %s on %s", action, key, destination); - try { - Thread.sleep(cfg.http().retryInterval()); - } catch (InterruptedException ie) { - log.atSevere().withCause(ie).log( - "%s %s towards %s was interrupted; giving up", action, key, destination); - Thread.currentThread().interrupt(); - return false; - } + boolean execute() throws ForwardingException { + log.atFine().log("Executing %s %s towards %s", action, key, destination); + try { + execCnt++; + tryOnce(); + log.atFine().log("%s %s towards %s OK", action, key, destination); + return true; + } catch (ForwardingException e) { + int maxTries = cfg.http().maxTries(); + log.atFine().withCause(e).log( + "Failed to %s %s on %s [%d/%d]", action, key, destination, execCnt, maxTries); + if (!e.isRecoverable()) { + log.atSevere().withCause(e).log( + "%s %s towards %s failed with unrecoverable error; giving up", + action, key, destination); + throw e; + } + if (execCnt >= maxTries) { + log.atSevere().log( + "Failed to %s %s on %s after %d tries; giving up", + action, key, destination, maxTries); + throw e; } } + return false; } void tryOnce() throws ForwardingException {
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarderScheduler.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarderScheduler.java new file mode 100644 index 0000000..ebbd940 --- /dev/null +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarderScheduler.java
@@ -0,0 +1,146 @@ +// Copyright (C) 2020 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.ericsson.gerrit.plugins.highavailability.forwarder.rest; + +import com.ericsson.gerrit.plugins.highavailability.Configuration; +import com.ericsson.gerrit.plugins.highavailability.peers.PeerInfo; +import com.google.common.annotations.VisibleForTesting; +import com.google.common.flogger.FluentLogger; +import com.google.gerrit.server.git.WorkQueue; +import com.google.inject.Inject; +import com.google.inject.Provider; +import com.google.inject.Singleton; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.function.Function; +import java.util.function.Supplier; + +@Singleton +public class RestForwarderScheduler { + private static final FluentLogger log = FluentLogger.forEnclosingClass(); + private final ScheduledExecutorService executor; + private final long retryIntervalMs; + + public class CompletablePromise<V> extends CompletableFuture<V> { + private Future<V> future; + + public CompletablePromise(Future<V> future) { + this.future = future; + executor.execute(this::tryToComplete); + } + + private void tryToComplete() { + if (future.isDone()) { + try { + complete(future.get()); + } catch (InterruptedException e) { + completeExceptionally(e); + } catch (ExecutionException e) { + completeExceptionally(e.getCause()); + } + return; + } + + if (future.isCancelled()) { + cancel(true); + return; + } + + executor.execute(this::tryToComplete); + } + } + + @Inject + public RestForwarderScheduler( + WorkQueue workQueue, Configuration cfg, Provider<Set<PeerInfo>> peerInfoProvider) { + int executorSize = peerInfoProvider.get().size() * cfg.index().threadPoolSize(); + retryIntervalMs = cfg.index().retryInterval(); + this.executor = workQueue.createQueue(executorSize, "RestForwarderScheduler"); + } + + @VisibleForTesting + public RestForwarderScheduler(ScheduledExecutorService executor) { + this.executor = executor; + retryIntervalMs = 0; + } + + public CompletableFuture<Boolean> execute(RestForwarder.Request request) { + return execute(request, 0); + } + + public CompletableFuture<Boolean> execute(RestForwarder.Request request, long delayMs) { + return supplyAsync( + request.toString(), + () -> { + try { + if (!request.execute()) { + log.atWarning().log( + "Rescheduling %s for retry after %d msec", request, retryIntervalMs); + return execute(request, retryIntervalMs); + } + return CompletableFuture.completedFuture(true); + } catch (ForwardingException e) { + log.atSevere().withCause(e).log("Forwarding of %s has failed", request); + return CompletableFuture.completedFuture(false); + } + }, + executor, + delayMs); + } + + private CompletableFuture<Boolean> supplyAsync( + String taskName, + Supplier<CompletableFuture<Boolean>> fn, + ScheduledExecutorService executor, + long delayMs) { + BooleanAsyncSupplier asyncSupplier = new BooleanAsyncSupplier(taskName, fn); + executor.schedule(asyncSupplier, delayMs, TimeUnit.MILLISECONDS); + return asyncSupplier.future(); + } + + static class BooleanAsyncSupplier implements Runnable { + private CompletableFuture<CompletableFuture<Boolean>> dep; + private Supplier<CompletableFuture<Boolean>> fn; + private String taskName; + + BooleanAsyncSupplier(String taskName, Supplier<CompletableFuture<Boolean>> fn) { + this.taskName = taskName; + this.dep = new CompletableFuture<>(); + this.fn = fn; + } + + public CompletableFuture<Boolean> future() { + return dep.thenCompose(Function.identity()); + } + + @Override + public void run() { + try { + dep.complete(fn.get()); + } catch (Throwable ex) { + dep.completeExceptionally(ex); + } + } + + @Override + public String toString() { + return taskName; + } + } +}
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandler.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandler.java index a352e38..b311db9 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandler.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandler.java
@@ -28,6 +28,7 @@ import com.google.inject.Inject; import java.util.Collections; import java.util.Set; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -155,7 +156,7 @@ this, () -> { queuedTasks.remove(this); - execute(); + return execute(); }, this::reschedule); } @@ -169,7 +170,7 @@ } } - abstract void execute(); + abstract CompletableFuture<Boolean> execute(); } class IndexChangeTask extends IndexTask { @@ -183,8 +184,8 @@ } @Override - public void execute() { - forwarder.indexChange(projectName, changeId, indexEvent); + public CompletableFuture<Boolean> execute() { + return forwarder.indexChange(projectName, changeId, indexEvent); } @Override @@ -216,8 +217,8 @@ } @Override - public void execute() { - forwarder.deleteChangeFromIndex(changeId, indexEvent); + public CompletableFuture<Boolean> execute() { + return forwarder.deleteChangeFromIndex(changeId, indexEvent); } @Override @@ -248,8 +249,8 @@ } @Override - public void execute() { - forwarder.indexAccount(accountId, indexEvent); + public CompletableFuture<Boolean> execute() { + return forwarder.indexAccount(accountId, indexEvent); } @Override @@ -280,8 +281,8 @@ } @Override - public void execute() { - forwarder.indexGroup(groupUUID, indexEvent); + public CompletableFuture<Boolean> execute() { + return forwarder.indexGroup(groupUUID, indexEvent); } @Override @@ -312,8 +313,8 @@ } @Override - public void execute() { - forwarder.indexProject(projectName, indexEvent); + public CompletableFuture<Boolean> execute() { + return forwarder.indexProject(projectName, indexEvent); } @Override
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventLocks.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventLocks.java index 4617e08..4434c34 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventLocks.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventLocks.java
@@ -20,6 +20,7 @@ import com.google.common.flogger.FluentLogger; import com.google.common.util.concurrent.Striped; import com.google.inject.Inject; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; @@ -38,15 +39,16 @@ } public void withLock( - IndexTask id, VoidFunction function, VoidFunction lockAcquireTimeoutCallback) { + IndexTask id, IndexCallFunction function, VoidFunction lockAcquireTimeoutCallback) { Lock idLock = getLock(id); try { if (idLock.tryLock(waitTimeout, TimeUnit.MILLISECONDS)) { - try { - function.invoke(); - } finally { - idLock.unlock(); - } + function + .invoke() + .whenComplete( + (result, error) -> { + idLock.unlock(); + }); } else { lockAcquireTimeoutCallback.invoke(); } @@ -64,4 +66,9 @@ public interface VoidFunction { void invoke(); } + + @FunctionalInterface + public interface IndexCallFunction { + CompletableFuture<?> invoke(); + } }
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexModule.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexModule.java index 9d0d03d..4a25fc8 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexModule.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexModule.java
@@ -36,7 +36,9 @@ .toProvider(ForwardedIndexExecutorProvider.class); bind(IndexEventLocks.class).in(Scopes.SINGLETON); listener().to(IndexExecutorProvider.class); - DynamicSet.bind(binder(), ChangeIndexedListener.class).to(IndexEventHandler.class); + DynamicSet.bind(binder(), ChangeIndexedListener.class) + .to(IndexEventHandler.class) + .in(Scopes.SINGLETON); DynamicSet.bind(binder(), AccountIndexedListener.class).to(IndexEventHandler.class); DynamicSet.bind(binder(), GroupIndexedListener.class).to(IndexEventHandler.class); DynamicSet.bind(binder(), ProjectIndexedListener.class).to(IndexEventHandler.class);
diff --git a/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarderTest.java b/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarderTest.java index 40ef88e..046d6ad 100644 --- a/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarderTest.java +++ b/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarderTest.java
@@ -37,6 +37,8 @@ import com.google.inject.Provider; import java.io.IOException; import java.util.Set; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import javax.net.ssl.SSLException; import org.junit.Before; import org.junit.Test; @@ -80,121 +82,170 @@ private static final String EVENT_ENDPOINT = Joiner.on("/").join(URL, PLUGINS, PLUGIN_NAME, "event", event.type); + private static final long TEST_TIMEOUT = 10; + private static final TimeUnit TEST_TIMEOUT_UNITS = TimeUnit.SECONDS; + private RestForwarder forwarder; private HttpSession httpSessionMock; private Gson gson = new Gson(); + private Configuration configMock; + Provider<Set<PeerInfo>> peersMock; @SuppressWarnings("unchecked") @Before public void setUp() { httpSessionMock = mock(HttpSession.class); - Configuration configMock = mock(Configuration.class, Answers.RETURNS_DEEP_STUBS); + configMock = mock(Configuration.class, Answers.RETURNS_DEEP_STUBS); when(configMock.http().maxTries()).thenReturn(3); when(configMock.http().retryInterval()).thenReturn(10); - Provider<Set<PeerInfo>> peersMock = mock(Provider.class); + peersMock = mock(Provider.class); when(peersMock.get()).thenReturn(ImmutableSet.of(new PeerInfo(URL))); forwarder = new RestForwarder( - httpSessionMock, PLUGIN_NAME, configMock, peersMock, gson); // TODO: Create provider + httpSessionMock, + PLUGIN_NAME, + configMock, + peersMock, + gson, // TODO: Create provider + new RestForwarderScheduler(Executors.newScheduledThreadPool(1))); } @Test public void testIndexAccountOK() throws Exception { when(httpSessionMock.post(eq(INDEX_ACCOUNT_ENDPOINT), any())) .thenReturn(new HttpResult(SUCCESSFUL, EMPTY_MSG)); - assertThat(forwarder.indexAccount(ACCOUNT_NUMBER, new IndexEvent())).isTrue(); + assertThat( + forwarder + .indexAccount(ACCOUNT_NUMBER, new IndexEvent()) + .get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isTrue(); } @Test public void testIndexAccountFailed() throws Exception { when(httpSessionMock.post(eq(INDEX_ACCOUNT_ENDPOINT), any())) .thenReturn(new HttpResult(FAILED, EMPTY_MSG)); - assertThat(forwarder.indexAccount(ACCOUNT_NUMBER, new IndexEvent())).isFalse(); + assertThat( + forwarder + .indexAccount(ACCOUNT_NUMBER, new IndexEvent()) + .get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isFalse(); } @Test public void testIndexAccountThrowsException() throws Exception { doThrow(new IOException()).when(httpSessionMock).post(eq(INDEX_ACCOUNT_ENDPOINT), any()); - assertThat(forwarder.indexAccount(ACCOUNT_NUMBER, new IndexEvent())).isFalse(); + assertThat( + forwarder + .indexAccount(ACCOUNT_NUMBER, new IndexEvent()) + .get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isFalse(); } @Test public void testIndexGroupOK() throws Exception { when(httpSessionMock.post(eq(INDEX_GROUP_ENDPOINT), any())) .thenReturn(new HttpResult(SUCCESSFUL, EMPTY_MSG)); - assertThat(forwarder.indexGroup(UUID, new IndexEvent())).isTrue(); + assertThat(forwarder.indexGroup(UUID, new IndexEvent()).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isTrue(); } @Test public void testIndexGroupFailed() throws Exception { when(httpSessionMock.post(eq(INDEX_GROUP_ENDPOINT), any())) .thenReturn(new HttpResult(FAILED, EMPTY_MSG)); - assertThat(forwarder.indexGroup(UUID, new IndexEvent())).isFalse(); + assertThat(forwarder.indexGroup(UUID, new IndexEvent()).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isFalse(); } @Test public void testIndexGroupThrowsException() throws Exception { doThrow(new IOException()).when(httpSessionMock).post(eq(INDEX_GROUP_ENDPOINT), any()); - assertThat(forwarder.indexGroup(UUID, new IndexEvent())).isFalse(); + assertThat(forwarder.indexGroup(UUID, new IndexEvent()).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isFalse(); } @Test public void testIndexChangeOK() throws Exception { when(httpSessionMock.post(eq(INDEX_CHANGE_ENDPOINT), any())) .thenReturn(new HttpResult(SUCCESSFUL, EMPTY_MSG)); - assertThat(forwarder.indexChange(PROJECT_NAME, CHANGE_NUMBER, new IndexEvent())).isTrue(); + assertThat( + forwarder + .indexChange(PROJECT_NAME, CHANGE_NUMBER, new IndexEvent()) + .get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isTrue(); } @Test public void testIndexChangeFailed() throws Exception { when(httpSessionMock.post(eq(INDEX_CHANGE_ENDPOINT), any())) .thenReturn(new HttpResult(FAILED, EMPTY_MSG)); - assertThat(forwarder.indexChange(PROJECT_NAME, CHANGE_NUMBER, new IndexEvent())).isFalse(); + assertThat( + forwarder + .indexChange(PROJECT_NAME, CHANGE_NUMBER, new IndexEvent()) + .get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isFalse(); } @Test public void testIndexChangeThrowsException() throws Exception { doThrow(new IOException()).when(httpSessionMock).post(eq(INDEX_CHANGE_ENDPOINT), any()); - assertThat(forwarder.indexChange(PROJECT_NAME, CHANGE_NUMBER, new IndexEvent())).isFalse(); + assertThat( + forwarder + .indexChange(PROJECT_NAME, CHANGE_NUMBER, new IndexEvent()) + .get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isFalse(); } @Test public void testChangeDeletedFromIndexOK() throws Exception { when(httpSessionMock.delete(eq(DELETE_CHANGE_ENDPOINT))) .thenReturn(new HttpResult(SUCCESSFUL, EMPTY_MSG)); - assertThat(forwarder.deleteChangeFromIndex(CHANGE_NUMBER, new IndexEvent())).isTrue(); + assertThat( + forwarder + .deleteChangeFromIndex(CHANGE_NUMBER, new IndexEvent()) + .get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isTrue(); } @Test public void testChangeDeletedFromIndexFailed() throws Exception { when(httpSessionMock.delete(eq(DELETE_CHANGE_ENDPOINT))) .thenReturn(new HttpResult(FAILED, EMPTY_MSG)); - assertThat(forwarder.deleteChangeFromIndex(CHANGE_NUMBER, new IndexEvent())).isFalse(); + assertThat( + forwarder + .deleteChangeFromIndex(CHANGE_NUMBER, new IndexEvent()) + .get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isFalse(); } @Test public void testChangeDeletedFromThrowsException() throws Exception { doThrow(new IOException()).when(httpSessionMock).delete(eq(DELETE_CHANGE_ENDPOINT)); - assertThat(forwarder.deleteChangeFromIndex(CHANGE_NUMBER, new IndexEvent())).isFalse(); + assertThat( + forwarder + .deleteChangeFromIndex(CHANGE_NUMBER, new IndexEvent()) + .get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isFalse(); } @Test public void testEventSentOK() throws Exception { when(httpSessionMock.post(EVENT_ENDPOINT, event)) .thenReturn(new HttpResult(SUCCESSFUL, EMPTY_MSG)); - assertThat(forwarder.send(event)).isTrue(); + assertThat(forwarder.send(event).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)).isTrue(); } @Test public void testEventSentFailed() throws Exception { when(httpSessionMock.post(EVENT_ENDPOINT, event)).thenReturn(new HttpResult(FAILED, EMPTY_MSG)); - assertThat(forwarder.send(event)).isFalse(); + assertThat(forwarder.send(event).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)).isFalse(); } @Test public void testEventSentThrowsException() throws Exception { doThrow(new IOException()).when(httpSessionMock).post(EVENT_ENDPOINT, event); - assertThat(forwarder.send(event)).isFalse(); + assertThat(forwarder.send(event).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)).isFalse(); } @Test @@ -203,7 +254,8 @@ String keyJson = gson.toJson(key); when(httpSessionMock.post(buildCacheEndpoint(Constants.PROJECTS), keyJson)) .thenReturn(new HttpResult(SUCCESSFUL, EMPTY_MSG)); - assertThat(forwarder.evict(Constants.PROJECTS, key)).isTrue(); + assertThat(forwarder.evict(Constants.PROJECTS, key).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isTrue(); } @Test @@ -212,7 +264,8 @@ String keyJson = gson.toJson(key); when(httpSessionMock.post(buildCacheEndpoint(Constants.ACCOUNTS), keyJson)) .thenReturn(new HttpResult(SUCCESSFUL, EMPTY_MSG)); - assertThat(forwarder.evict(Constants.ACCOUNTS, key)).isTrue(); + assertThat(forwarder.evict(Constants.ACCOUNTS, key).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isTrue(); } @Test @@ -221,7 +274,8 @@ String keyJson = gson.toJson(key); String endpoint = buildCacheEndpoint(Constants.GROUPS); when(httpSessionMock.post(endpoint, keyJson)).thenReturn(new HttpResult(SUCCESSFUL, EMPTY_MSG)); - assertThat(forwarder.evict(Constants.GROUPS, key)).isTrue(); + assertThat(forwarder.evict(Constants.GROUPS, key).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isTrue(); } @Test @@ -230,7 +284,9 @@ String keyJson = gson.toJson(key); when(httpSessionMock.post(buildCacheEndpoint(Constants.GROUPS_BYINCLUDE), keyJson)) .thenReturn(new HttpResult(SUCCESSFUL, EMPTY_MSG)); - assertThat(forwarder.evict(Constants.GROUPS_BYINCLUDE, key)).isTrue(); + assertThat( + forwarder.evict(Constants.GROUPS_BYINCLUDE, key).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isTrue(); } @Test @@ -239,7 +295,8 @@ String keyJson = gson.toJson(key); when(httpSessionMock.post(buildCacheEndpoint(Constants.GROUPS_MEMBERS), keyJson)) .thenReturn(new HttpResult(SUCCESSFUL, EMPTY_MSG)); - assertThat(forwarder.evict(Constants.GROUPS_MEMBERS, key)).isTrue(); + assertThat(forwarder.evict(Constants.GROUPS_MEMBERS, key).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isTrue(); } @Test @@ -248,7 +305,8 @@ String keyJson = gson.toJson(key); when(httpSessionMock.post(buildCacheEndpoint(Constants.PROJECTS), keyJson)) .thenReturn(new HttpResult(FAILED, EMPTY_MSG)); - assertThat(forwarder.evict(Constants.PROJECTS, key)).isFalse(); + assertThat(forwarder.evict(Constants.PROJECTS, key).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isFalse(); } @Test @@ -258,7 +316,8 @@ doThrow(new IOException()) .when(httpSessionMock) .post(buildCacheEndpoint(Constants.PROJECTS), keyJson); - assertThat(forwarder.evict(Constants.PROJECTS, key)).isFalse(); + assertThat(forwarder.evict(Constants.PROJECTS, key).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isFalse(); } private static String buildCacheEndpoint(String name) { @@ -270,7 +329,8 @@ String projectName = PROJECT_TO_ADD; when(httpSessionMock.post(buildProjectListCacheEndpoint(projectName), null)) .thenReturn(new HttpResult(SUCCESSFUL, EMPTY_MSG)); - assertThat(forwarder.addToProjectList(projectName)).isTrue(); + assertThat(forwarder.addToProjectList(projectName).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isTrue(); } @Test @@ -278,7 +338,8 @@ String projectName = PROJECT_TO_ADD; when(httpSessionMock.post(buildProjectListCacheEndpoint(projectName), null)) .thenReturn(new HttpResult(FAILED, EMPTY_MSG)); - assertThat(forwarder.addToProjectList(projectName)).isFalse(); + assertThat(forwarder.addToProjectList(projectName).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isFalse(); } @Test @@ -287,7 +348,8 @@ doThrow(new IOException()) .when(httpSessionMock) .post(buildProjectListCacheEndpoint(projectName), null); - assertThat(forwarder.addToProjectList(projectName)).isFalse(); + assertThat(forwarder.addToProjectList(projectName).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isFalse(); } @Test @@ -295,7 +357,8 @@ String projectName = PROJECT_TO_DELETE; when(httpSessionMock.delete(buildProjectListCacheEndpoint(projectName))) .thenReturn(new HttpResult(SUCCESSFUL, EMPTY_MSG)); - assertThat(forwarder.removeFromProjectList(projectName)).isTrue(); + assertThat(forwarder.removeFromProjectList(projectName).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isTrue(); } @Test @@ -303,7 +366,8 @@ String projectName = PROJECT_TO_DELETE; when(httpSessionMock.delete(buildProjectListCacheEndpoint(projectName))) .thenReturn(new HttpResult(FAILED, EMPTY_MSG)); - assertThat(forwarder.removeFromProjectList(projectName)).isFalse(); + assertThat(forwarder.removeFromProjectList(projectName).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isFalse(); } @Test @@ -312,7 +376,8 @@ doThrow(new IOException()) .when(httpSessionMock) .delete((buildProjectListCacheEndpoint(projectName))); - assertThat(forwarder.removeFromProjectList(projectName)).isFalse(); + assertThat(forwarder.removeFromProjectList(projectName).get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isFalse(); } private static String buildProjectListCacheEndpoint(String projectName) { @@ -320,41 +385,57 @@ } @Test - public void testRetryOnErrorThenSuccess() throws IOException { + public void testRetryOnErrorThenSuccess() throws Exception { when(httpSessionMock.post(anyString(), anyString())) .thenReturn(new HttpResult(false, ERROR)) .thenReturn(new HttpResult(false, ERROR)) .thenReturn(new HttpResult(true, SUCCESS)); - assertThat(forwarder.evict(Constants.PROJECT_LIST, new Object())).isTrue(); + assertThat( + forwarder + .evict(Constants.PROJECT_LIST, new Object()) + .get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isTrue(); } @Test - public void testRetryOnIoExceptionThenSuccess() throws IOException { + public void testRetryOnIoExceptionThenSuccess() throws Exception { when(httpSessionMock.post(anyString(), anyString())) .thenThrow(new IOException()) .thenThrow(new IOException()) .thenReturn(new HttpResult(true, SUCCESS)); - assertThat(forwarder.evict(Constants.PROJECT_LIST, new Object())).isTrue(); + assertThat( + forwarder + .evict(Constants.PROJECT_LIST, new Object()) + .get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isTrue(); } @Test - public void testNoRetryAfterNonRecoverableException() throws IOException { + public void testNoRetryAfterNonRecoverableException() throws Exception { when(httpSessionMock.post(anyString(), anyString())) .thenThrow(new SSLException("Non Recoverable")) .thenReturn(new HttpResult(true, SUCCESS)); - assertThat(forwarder.evict(Constants.PROJECT_LIST, new Object())).isFalse(); + assertThat( + forwarder + .evict(Constants.PROJECT_LIST, new Object()) + .get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isFalse(); } @Test - public void testFailureAfterMaxTries() throws IOException { + public void testFailureAfterMaxTries() throws Exception { when(httpSessionMock.post(anyString(), anyString())) .thenReturn(new HttpResult(false, ERROR)) .thenReturn(new HttpResult(false, ERROR)) .thenReturn(new HttpResult(false, ERROR)); - assertThat(forwarder.evict(Constants.PROJECT_LIST, new Object())).isFalse(); + assertThat( + forwarder + .evict(Constants.PROJECT_LIST, new Object()) + .get(TEST_TIMEOUT, TEST_TIMEOUT_UNITS)) + .isFalse(); } }
diff --git a/src/test/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandlerTest.java b/src/test/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandlerTest.java index 9d8f015..db4c613 100644 --- a/src/test/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandlerTest.java +++ b/src/test/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandlerTest.java
@@ -43,6 +43,7 @@ import java.util.List; import java.util.Optional; import java.util.concurrent.Callable; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; @@ -105,6 +106,13 @@ when(configuration.http()).thenReturn(http); when(http.maxTries()).thenReturn(Configuration.Http.DEFAULT_MAX_TRIES); when(http.retryInterval()).thenReturn(Configuration.Http.DEFAULT_RETRY_INTERVAL); + when(forwarder.indexAccount(eq(ACCOUNT_ID), any())) + .thenReturn(CompletableFuture.completedFuture(true)); + when(forwarder.deleteChangeFromIndex(eq(CHANGE_ID), any())) + .thenReturn(CompletableFuture.completedFuture(true)); + when(forwarder.indexGroup(eq(UUID), any())).thenReturn(CompletableFuture.completedFuture(true)); + when(forwarder.indexChange(eq(PROJECT_NAME), eq(CHANGE_ID), any())) + .thenReturn(CompletableFuture.completedFuture(true)); idLocks = new IndexEventLocks(configuration); setUpIndexEventHandler(currCtx);