Use failsafe for indexing at the receiving side Replace custom coded async-retry with failsafe. Change-Id: If5ec4a1edffbc8da67842d46cebf565187a52716
diff --git a/BUILD b/BUILD index e11005d..4f10eab 100644 --- a/BUILD +++ b/BUILD
@@ -50,6 +50,7 @@ "@global-refdb//jar", "@wiremock//jar", "@jgroups//jar", - "@commons-net//jar" + "@commons-net//jar", + "@failsafe//jar", ], )
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexAccountHandler.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexAccountHandler.java index b1a595c..b2f8141 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexAccountHandler.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexAccountHandler.java
@@ -19,6 +19,7 @@ import com.google.inject.Inject; import com.google.inject.Singleton; import java.util.Optional; +import java.util.concurrent.CompletableFuture; /** * Index an account using {@link AccountIndexer}. This class is meant to be used on the receiving @@ -36,13 +37,14 @@ } @Override - protected void doIndex(Account.Id id, Optional<IndexEvent> indexEvent) { + protected CompletableFuture<Boolean> doIndex(Account.Id id, Optional<IndexEvent> indexEvent) { indexer.index(id); log.atFine().log("Account %s successfully indexed", id); + return CompletableFuture.completedFuture(true); } @Override - protected void doDelete(Account.Id id, Optional<IndexEvent> indexEvent) { + protected CompletableFuture<Boolean> doDelete(Account.Id id, Optional<IndexEvent> indexEvent) { throw new UnsupportedOperationException("Delete from account index not supported"); } }
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexBatchChangeHandler.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexBatchChangeHandler.java index dee8876..bf48f76 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexBatchChangeHandler.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexBatchChangeHandler.java
@@ -14,14 +14,13 @@ package com.ericsson.gerrit.plugins.highavailability.forwarder; -import com.ericsson.gerrit.plugins.highavailability.Configuration; import com.ericsson.gerrit.plugins.highavailability.index.ChangeCheckerImpl.Factory; import com.ericsson.gerrit.plugins.highavailability.index.ForwardedBatchIndexExecutor; import com.google.gerrit.server.index.change.ChangeIndexer; import com.google.gerrit.server.util.OneOffRequestContext; import com.google.inject.Inject; import com.google.inject.Singleton; -import java.util.concurrent.ScheduledExecutorService; +import dev.failsafe.FailsafeExecutor; @Singleton public class ForwardedIndexBatchChangeHandler extends ForwardedIndexChangeHandler { @@ -29,10 +28,9 @@ @Inject ForwardedIndexBatchChangeHandler( ChangeIndexer indexer, - Configuration config, - @ForwardedBatchIndexExecutor ScheduledExecutorService indexExecutor, + @ForwardedBatchIndexExecutor FailsafeExecutor<Boolean> indexExecutor, OneOffRequestContext oneOffCtx, Factory changeCheckerFactory) { - super(indexer, config, indexExecutor, oneOffCtx, changeCheckerFactory); + super(indexer, indexExecutor, oneOffCtx, changeCheckerFactory); } }
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexChangeHandler.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexChangeHandler.java index ab8895b..c232af2 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexChangeHandler.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexChangeHandler.java
@@ -14,8 +14,6 @@ package com.ericsson.gerrit.plugins.highavailability.forwarder; -import com.ericsson.gerrit.plugins.highavailability.Configuration; -import com.ericsson.gerrit.plugins.highavailability.Configuration.Index; import com.ericsson.gerrit.plugins.highavailability.index.ChangeChecker; import com.ericsson.gerrit.plugins.highavailability.index.ChangeCheckerImpl; import com.ericsson.gerrit.plugins.highavailability.index.ForwardedIndexExecutor; @@ -30,12 +28,11 @@ import com.google.gerrit.server.util.OneOffRequestContext; import com.google.inject.Inject; import com.google.inject.Singleton; +import dev.failsafe.FailsafeExecutor; import java.io.IOException; -import java.time.Duration; import java.util.List; import java.util.Optional; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.CompletableFuture; /** * Index a change using {@link ChangeIndexer}. This class is meant to be used on the receiving side @@ -46,36 +43,35 @@ @Singleton public class ForwardedIndexChangeHandler extends ForwardedIndexingHandler<String> { private final ChangeIndexer indexer; - private final ScheduledExecutorService indexExecutor; + private final FailsafeExecutor<Boolean> indexExecutor; private final OneOffRequestContext oneOffCtx; - private final Duration retryInterval; - private final int maxTries; private final ChangeCheckerImpl.Factory changeCheckerFactory; @Inject ForwardedIndexChangeHandler( ChangeIndexer indexer, - Configuration config, - @ForwardedIndexExecutor ScheduledExecutorService indexExecutor, + @ForwardedIndexExecutor FailsafeExecutor<Boolean> indexExecutor, OneOffRequestContext oneOffCtx, ChangeCheckerImpl.Factory changeCheckerFactory) { this.indexer = indexer; this.indexExecutor = indexExecutor; this.oneOffCtx = oneOffCtx; this.changeCheckerFactory = changeCheckerFactory; - - Index indexConfig = config.index(); - this.retryInterval = indexConfig != null ? indexConfig.retryInterval() : Duration.ZERO; - this.maxTries = indexConfig != null ? indexConfig.maxTries() : 0; } @Override - protected void doIndex(String id, Optional<IndexEvent> indexEvent) throws IOException { - doIndex(id, indexEvent, 0); + protected CompletableFuture<Boolean> doIndex(String id, Optional<IndexEvent> indexEvent) + throws IOException { + return indexExecutor.getAsync( + () -> { + try (ManualRequestContext ctx = oneOffCtx.open()) { + Context.setForwardedEvent(true); + return indexOnce(id, indexEvent); + } + }); } - private void doIndex(String id, Optional<IndexEvent> indexEvent, int retryCount) - throws IOException { + private boolean indexOnce(String id, Optional<IndexEvent> indexEvent) throws Exception { try { ChangeChecker checker = changeCheckerFactory.create(id); Optional<ChangeNotes> changeNotes; @@ -90,33 +86,25 @@ reindex(notes); if (checker.isChangeUpToDate(indexEvent)) { - if (retryCount > 0) { - log.atWarning().log( - "Change %s has been eventually indexed after %d attempt(s)", id, retryCount); - } else { - log.atFine().log("Change %s successfully indexed", id); - } - } else { - log.atWarning().log( - "Change %s seems too old compared to the event timestamp (event-Ts=%s >> change-Ts=%s)", - id, indexEvent, checker); - rescheduleIndex(id, indexEvent, retryCount + 1); + log.atFine().log("Change %s successfully indexed", id); + return true; } - } else { + log.atWarning().log( - "Change %s not present yet in local Git repository (event=%s) after %d attempt(s)", - id, indexEvent, retryCount); - if (!rescheduleIndex(id, indexEvent, retryCount + 1)) { - log.atSevere().log( - "Change %s could not be found in the local Git repository (event=%s)", - id, indexEvent); - } + "Change %s seems too old compared to the event timestamp (event-Ts=%s >> change-Ts=%s)", + id, indexEvent, checker); + return false; } + + log.atWarning().log( + "Change %s not present yet in local Git repository (event=%s)", id, indexEvent); + return false; + } catch (Exception e) { if (isCausedByNoSuchChangeException(e)) { indexer.delete(parseChangeId(id)); log.atWarning().withCause(e).log("Error trying to index Change %s. Deleted from index", id); - return; + return true; } throw e; @@ -128,34 +116,12 @@ indexer.index(notes.getChange()); } - private boolean rescheduleIndex(String id, Optional<IndexEvent> indexEvent, int retryCount) { - if (retryCount > maxTries) { - log.atSevere().log( - "Change %s could not be indexed after %d retries. Change index could be stale.", - id, retryCount); - return false; - } - - log.atWarning().log( - "Retrying for the #%d time to index Change %s after %s", retryCount, id, retryInterval); - indexExecutor.schedule( - () -> { - try (ManualRequestContext ctx = oneOffCtx.open()) { - Context.setForwardedEvent(true); - doIndex(id, indexEvent, retryCount); - } catch (Exception e) { - log.atWarning().withCause(e).log("Change %s could not be indexed", id); - } - }, - retryInterval.toMillis(), - TimeUnit.MILLISECONDS); - return true; - } - @Override - protected void doDelete(String id, Optional<IndexEvent> indexEvent) throws IOException { + protected CompletableFuture<Boolean> doDelete(String id, Optional<IndexEvent> indexEvent) + throws IOException { indexer.delete(parseChangeId(id)); log.atFine().log("Change %s successfully deleted from index", id); + return CompletableFuture.completedFuture(true); } private static Change.Id parseChangeId(String id) {
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexGroupHandler.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexGroupHandler.java index ab31659..99ac369 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexGroupHandler.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexGroupHandler.java
@@ -19,6 +19,7 @@ import com.google.inject.Inject; import com.google.inject.Singleton; import java.util.Optional; +import java.util.concurrent.CompletableFuture; /** * Index a group using {@link GroupIndexer}. This class is meant to be used on the receiving side of @@ -36,13 +37,16 @@ } @Override - protected void doIndex(AccountGroup.UUID uuid, Optional<IndexEvent> indexEvent) { + protected CompletableFuture<Boolean> doIndex( + AccountGroup.UUID uuid, Optional<IndexEvent> indexEvent) { indexer.index(uuid); log.atFine().log("Group %s successfully indexed", uuid); + return CompletableFuture.completedFuture(true); } @Override - protected void doDelete(AccountGroup.UUID uuid, Optional<IndexEvent> indexEvent) { + protected CompletableFuture<Boolean> doDelete( + AccountGroup.UUID uuid, Optional<IndexEvent> indexEvent) { throw new UnsupportedOperationException("Delete from group index not supported"); } }
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexProjectHandler.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexProjectHandler.java index 20ffefd..b16cb6e 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexProjectHandler.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexProjectHandler.java
@@ -19,6 +19,7 @@ import com.google.inject.Inject; import com.google.inject.Singleton; import java.util.Optional; +import java.util.concurrent.CompletableFuture; /** * Index a project using {@link ProjectIndexer}. This class is meant to be used on the receiving @@ -36,13 +37,16 @@ } @Override - protected void doIndex(Project.NameKey projectName, Optional<IndexEvent> indexEvent) { + protected CompletableFuture<Boolean> doIndex( + Project.NameKey projectName, Optional<IndexEvent> indexEvent) { indexer.index(projectName); log.atFine().log("Project %s successfully indexed", projectName); + return CompletableFuture.completedFuture(true); } @Override - protected void doDelete(Project.NameKey projectName, Optional<IndexEvent> indexEvent) { + protected CompletableFuture<Boolean> doDelete( + Project.NameKey projectName, Optional<IndexEvent> indexEvent) { throw new UnsupportedOperationException("Delete from project index not supported"); } }
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexingHandler.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexingHandler.java index 44e35a8..6ce925a 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexingHandler.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexingHandler.java
@@ -19,6 +19,7 @@ import java.util.Collections; import java.util.Optional; import java.util.Set; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; /** @@ -41,9 +42,11 @@ } } - protected abstract void doIndex(T id, Optional<IndexEvent> indexEvent) throws IOException; + protected abstract CompletableFuture<Boolean> doIndex(T id, Optional<IndexEvent> indexEvent) + throws IOException; - protected abstract void doDelete(T id, Optional<IndexEvent> indexEvent) throws IOException; + protected abstract CompletableFuture<Boolean> doDelete(T id, Optional<IndexEvent> indexEvent) + throws IOException; /** * Index an item in the local node, indexing will not be forwarded to the other node. @@ -53,29 +56,27 @@ * @param indexEvent The index event details. * @throws IOException If an error occur while indexing. */ - public void index(T id, Operation operation, Optional<IndexEvent> indexEvent) throws IOException { + public CompletableFuture<Boolean> index( + T id, Operation operation, Optional<IndexEvent> indexEvent) throws IOException { log.atFine().log("%s %s %s", operation, id, indexEvent); if (inFlightIndexing.add(id)) { try { Context.setForwardedEvent(true); switch (operation) { case INDEX: - doIndex(id, indexEvent); - break; + return doIndex(id, indexEvent); case DELETE: - doDelete(id, indexEvent); - break; + return doDelete(id, indexEvent); default: log.atSevere().log("unexpected operation: %s", operation); - break; + return CompletableFuture.completedFuture(false); } } finally { Context.unsetForwardedEvent(); inFlightIndexing.remove(id); } - } else { - throw new InFlightIndexedException( - String.format("Indexing for %s %s %s already in flight", operation, id, indexEvent)); } + throw new InFlightIndexedException( + String.format("Indexing for %s %s %s already in flight", operation, id, indexEvent)); } }
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexExecutorProvider.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexExecutorProvider.java index f5bc85e..64fe3b1 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexExecutorProvider.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexExecutorProvider.java
@@ -15,16 +15,19 @@ package com.ericsson.gerrit.plugins.highavailability.index; import com.ericsson.gerrit.plugins.highavailability.Configuration; -import com.ericsson.gerrit.plugins.highavailability.ExecutorProvider; -import com.google.gerrit.server.git.WorkQueue; import com.google.inject.Inject; import com.google.inject.Singleton; @Singleton -class ForwardedBatchIndexExecutorProvider extends ExecutorProvider { +class ForwardedBatchIndexExecutorProvider extends ForwardedIndexExecutorProvider { @Inject - ForwardedBatchIndexExecutorProvider(WorkQueue workQueue, Configuration config) { - super(workQueue, config.index().batchThreadPoolSize(), "Forwarded-BatchIndex-Event"); + ForwardedBatchIndexExecutorProvider(Configuration cfg) { + super(cfg); + } + + @Override + protected int threadPoolSize(Configuration cfg) { + return cfg.index().batchThreadPoolSize(); } }
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedIndexExecutorProvider.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedIndexExecutorProvider.java index 2112dbe..8819d74 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedIndexExecutorProvider.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedIndexExecutorProvider.java
@@ -15,16 +15,39 @@ package com.ericsson.gerrit.plugins.highavailability.index; import com.ericsson.gerrit.plugins.highavailability.Configuration; -import com.ericsson.gerrit.plugins.highavailability.ExecutorProvider; -import com.google.gerrit.server.git.WorkQueue; import com.google.inject.Inject; +import com.google.inject.Provider; import com.google.inject.Singleton; +import dev.failsafe.Failsafe; +import dev.failsafe.FailsafeExecutor; +import dev.failsafe.RetryPolicy; +import java.io.IOException; +import java.util.concurrent.Executors; @Singleton -class ForwardedIndexExecutorProvider extends ExecutorProvider { +public class ForwardedIndexExecutorProvider implements Provider<FailsafeExecutor<Boolean>> { + + private final Configuration cfg; @Inject - ForwardedIndexExecutorProvider(WorkQueue workQueue, Configuration config) { - super(workQueue, config.index().threadPoolSize(), "Forwarded-Index-Event"); + public ForwardedIndexExecutorProvider(Configuration cfg) { + this.cfg = cfg; + } + + @Override + public FailsafeExecutor<Boolean> get() { + RetryPolicy<Boolean> retryPolicy = + RetryPolicy.<Boolean>builder() + .withMaxAttempts(cfg.index().maxTries()) + .withDelay(cfg.index().retryInterval()) + .handleResult(false) + .abortOn(IOException.class) + .build(); + // TODO: the executor shall be created by workQueue.createQueue(...) + return Failsafe.with(retryPolicy).with(Executors.newScheduledThreadPool(threadPoolSize(cfg))); + } + + protected int threadPoolSize(Configuration cfg) { + return cfg.index().threadPoolSize(); } }
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 b973d52..208bc86 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
@@ -21,19 +21,24 @@ import com.google.gerrit.extensions.registration.DynamicSet; import com.google.inject.AbstractModule; import com.google.inject.Scopes; +import com.google.inject.TypeLiteral; import com.google.inject.assistedinject.FactoryModuleBuilder; -import java.util.concurrent.ScheduledExecutorService; +import dev.failsafe.FailsafeExecutor; public class IndexModule extends AbstractModule { @Override protected void configure() { - bind(ScheduledExecutorService.class) + bind(new TypeLiteral<FailsafeExecutor<Boolean>>() {}) .annotatedWith(ForwardedIndexExecutor.class) - .toProvider(ForwardedIndexExecutorProvider.class); - bind(ScheduledExecutorService.class) + .toProvider(ForwardedIndexExecutorProvider.class) + .in(Scopes.SINGLETON); + + bind(new TypeLiteral<FailsafeExecutor<Boolean>>() {}) .annotatedWith(ForwardedBatchIndexExecutor.class) - .toProvider(ForwardedBatchIndexExecutorProvider.class); + .toProvider(ForwardedBatchIndexExecutorProvider.class) + .in(Scopes.SINGLETON); + DynamicSet.bind(binder(), ChangeIndexedListener.class) .to(IndexEventHandler.class) .in(Scopes.SINGLETON);
diff --git a/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexAccountHandlerTest.java b/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexAccountHandlerTest.java index 0e9cee8..e549e50 100644 --- a/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexAccountHandlerTest.java +++ b/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexAccountHandlerTest.java
@@ -16,6 +16,7 @@ import static com.google.common.truth.Truth.assertThat; import static com.google.gerrit.testing.GerritJUnit.assertThrows; +import static java.util.concurrent.TimeUnit.SECONDS; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.verify; @@ -46,7 +47,7 @@ @Test public void testSuccessfulIndexing() throws Exception { - handler.index(id, Operation.INDEX, Optional.empty()); + handler.index(id, Operation.INDEX, Optional.empty()).get(10, SECONDS); verify(indexerMock).index(id); } @@ -55,7 +56,7 @@ UnsupportedOperationException thrown = assertThrows( UnsupportedOperationException.class, - () -> handler.index(id, Operation.DELETE, Optional.empty())); + () -> handler.index(id, Operation.DELETE, Optional.empty()).get(10, SECONDS)); assertThat(thrown).hasMessageThat().contains("Delete from account index not supported"); } @@ -73,7 +74,7 @@ .index(id); assertThat(Context.isForwardedEvent()).isFalse(); - handler.index(id, Operation.INDEX, Optional.empty()); + handler.index(id, Operation.INDEX, Optional.empty()).get(10, SECONDS); assertThat(Context.isForwardedEvent()).isFalse(); verify(indexerMock).index(id); @@ -92,7 +93,9 @@ assertThat(Context.isForwardedEvent()).isFalse(); IOException thrown = - assertThrows(IOException.class, () -> handler.index(id, Operation.INDEX, Optional.empty())); + assertThrows( + IOException.class, + () -> handler.index(id, Operation.INDEX, Optional.empty()).get(10, SECONDS)); assertThat(thrown).hasMessageThat().isEqualTo("someMessage"); assertThat(Context.isForwardedEvent()).isFalse();
diff --git a/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexChangeHandlerTest.java b/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexChangeHandlerTest.java index 105f312..fa70958 100644 --- a/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexChangeHandlerTest.java +++ b/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexChangeHandlerTest.java
@@ -16,6 +16,7 @@ import static com.google.common.truth.Truth.assertThat; import static com.google.gerrit.testing.GerritJUnit.assertThrows; +import static java.util.concurrent.TimeUnit.SECONDS; import static org.mockito.Answers.RETURNS_DEEP_STUBS; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.atLeast; @@ -28,15 +29,17 @@ import com.ericsson.gerrit.plugins.highavailability.forwarder.ForwardedIndexingHandler.Operation; import com.ericsson.gerrit.plugins.highavailability.index.ChangeChecker; import com.ericsson.gerrit.plugins.highavailability.index.ChangeCheckerImpl; +import com.ericsson.gerrit.plugins.highavailability.index.ForwardedIndexExecutorProvider; import com.google.gerrit.entities.Change; import com.google.gerrit.server.index.change.ChangeIndexer; import com.google.gerrit.server.notedb.ChangeNotes; import com.google.gerrit.server.util.OneOffRequestContext; +import dev.failsafe.FailsafeExecutor; import java.io.IOException; +import java.time.Duration; import java.time.Instant; import java.util.Optional; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ExecutionException; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -73,38 +76,40 @@ id = Change.id(TEST_CHANGE_NUMBER); Change change = new Change(null, id, null, null, Instant.now()); when(changeNotes.getChange()).thenReturn(change); + when(configMock.index().threadPoolSize()).thenReturn(4); when(configMock.index().maxTries()).thenReturn(3); + when(configMock.index().retryInterval()).thenReturn(Duration.ofMillis(10)); when(changeCheckerFactoryMock.create(any())).thenReturn(changeCheckerAbsentMock); - ScheduledExecutorService executor = Executors.newScheduledThreadPool(2); + FailsafeExecutor<Boolean> indexExecutor = new ForwardedIndexExecutorProvider(configMock).get(); handler = new ForwardedIndexChangeHandler( - indexerMock, configMock, executor, ctxMock, changeCheckerFactoryMock); + indexerMock, indexExecutor, ctxMock, changeCheckerFactoryMock); } @Test public void changeIsIndexedWhenUpToDate() throws Exception { setupChangeAccessRelatedMocks(CHANGE_EXISTS, CHANGE_UP_TO_DATE); - handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.empty()); + handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.empty()).get(10, SECONDS); verify(indexerMock, times(1)).index(any(Change.class)); } @Test public void changeIsStillIndexedEvenWhenOutdated() throws Exception { setupChangeAccessRelatedMocks(CHANGE_EXISTS, CHANGE_OUTDATED); - handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.of(new IndexEvent())); + handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.of(new IndexEvent())).get(10, SECONDS); verify(indexerMock, atLeast(1)).index(any(Change.class)); } @Test public void changeIsDeletedFromIndex() throws Exception { - handler.index(TEST_CHANGE_ID, Operation.DELETE, Optional.empty()); + handler.index(TEST_CHANGE_ID, Operation.DELETE, Optional.empty()).get(10, SECONDS); verify(indexerMock, times(1)).delete(id); } @Test public void changeToIndexDoesNotExist() throws Exception { setupChangeAccessRelatedMocks(CHANGE_DOES_NOT_EXIST, CHANGE_OUTDATED); - handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.empty()); + handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.empty()).get(10, SECONDS); verify(indexerMock, times(0)).delete(id); } @@ -123,7 +128,7 @@ .index(any(Change.class)); assertThat(Context.isForwardedEvent()).isFalse(); - handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.empty()); + handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.empty()).get(10, SECONDS); assertThat(Context.isForwardedEvent()).isFalse(); verify(indexerMock, times(1)).index(any(Change.class)); @@ -142,11 +147,12 @@ .index(any(Change.class)); assertThat(Context.isForwardedEvent()).isFalse(); - IOException thrown = + ExecutionException thrown = assertThrows( - IOException.class, - () -> handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.empty())); - assertThat(thrown).hasMessageThat().isEqualTo("someMessage"); + ExecutionException.class, + () -> + handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.empty()).get(10, SECONDS)); + assertThat(thrown.getCause()).hasMessageThat().isEqualTo("someMessage"); assertThat(Context.isForwardedEvent()).isFalse(); verify(indexerMock, times(1)).index(any(Change.class));
diff --git a/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexGroupHandlerTest.java b/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexGroupHandlerTest.java index 5fdb151..babe1d4 100644 --- a/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexGroupHandlerTest.java +++ b/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexGroupHandlerTest.java
@@ -16,6 +16,7 @@ import static com.google.common.truth.Truth.assertThat; import static com.google.gerrit.testing.GerritJUnit.assertThrows; +import static java.util.concurrent.TimeUnit.SECONDS; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.verify; @@ -46,7 +47,7 @@ @Test public void testSuccessfulIndexing() throws Exception { - handler.index(uuid, Operation.INDEX, Optional.empty()); + handler.index(uuid, Operation.INDEX, Optional.empty()).get(10, SECONDS); verify(indexerMock).index(uuid); } @@ -55,7 +56,7 @@ UnsupportedOperationException thrown = assertThrows( UnsupportedOperationException.class, - () -> handler.index(uuid, Operation.DELETE, Optional.empty())); + () -> handler.index(uuid, Operation.DELETE, Optional.empty()).get(10, SECONDS)); assertThat(thrown).hasMessageThat().contains("Delete from group index not supported"); } @@ -73,7 +74,7 @@ .index(uuid); assertThat(Context.isForwardedEvent()).isFalse(); - handler.index(uuid, Operation.INDEX, Optional.empty()); + handler.index(uuid, Operation.INDEX, Optional.empty()).get(10, SECONDS); assertThat(Context.isForwardedEvent()).isFalse(); verify(indexerMock).index(uuid); @@ -93,7 +94,8 @@ assertThat(Context.isForwardedEvent()).isFalse(); IOException thrown = assertThrows( - IOException.class, () -> handler.index(uuid, Operation.INDEX, Optional.empty())); + IOException.class, + () -> handler.index(uuid, Operation.INDEX, Optional.empty()).get(10, SECONDS)); assertThat(thrown).hasMessageThat().isEqualTo("someMessage"); assertThat(Context.isForwardedEvent()).isFalse();
diff --git a/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexProjectHandlerTest.java b/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexProjectHandlerTest.java index 0c896b3..f91a83e 100644 --- a/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexProjectHandlerTest.java +++ b/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexProjectHandlerTest.java
@@ -16,6 +16,7 @@ import static com.google.common.truth.Truth.assertThat; import static com.google.gerrit.testing.GerritJUnit.assertThrows; +import static java.util.concurrent.TimeUnit.SECONDS; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.verify; @@ -46,7 +47,7 @@ @Test public void testSuccessfulIndexing() throws Exception { - handler.index(nameKey, Operation.INDEX, Optional.empty()); + handler.index(nameKey, Operation.INDEX, Optional.empty()).get(10, SECONDS); verify(indexerMock).index(nameKey); } @@ -55,7 +56,7 @@ UnsupportedOperationException thrown = assertThrows( UnsupportedOperationException.class, - () -> handler.index(nameKey, Operation.DELETE, Optional.empty())); + () -> handler.index(nameKey, Operation.DELETE, Optional.empty()).get(10, SECONDS)); assertThat(thrown).hasMessageThat().contains("Delete from project index not supported"); } @@ -73,7 +74,7 @@ .index(nameKey); assertThat(Context.isForwardedEvent()).isFalse(); - handler.index(nameKey, Operation.INDEX, Optional.empty()); + handler.index(nameKey, Operation.INDEX, Optional.empty()).get(10, SECONDS); assertThat(Context.isForwardedEvent()).isFalse(); verify(indexerMock).index(nameKey); @@ -93,7 +94,8 @@ assertThat(Context.isForwardedEvent()).isFalse(); IOException thrown = assertThrows( - IOException.class, () -> handler.index(nameKey, Operation.INDEX, Optional.empty())); + IOException.class, + () -> handler.index(nameKey, Operation.INDEX, Optional.empty()).get(10, SECONDS)); assertThat(thrown).hasMessageThat().isEqualTo("someMessage"); assertThat(Context.isForwardedEvent()).isFalse();