index: fix forwarded event flag and in-flight guard for async indexing Context.setForwardedEvent(true/false) was called in the base class try/finally block on the calling thread. For the change handler, which submits work to an executor, this meant the flag was set on the wrong thread and cleared as soon as doIndex() returned the CompletableFuture, before the actual indexing work started on the executor thread. To workaround that, the ForwardedIndexChangeHandler invoked Context.setForwardedEvent(true) again from the indexing task, which was an ugly workaround. Similarly, inFlightIndexing.remove(id) was called in the finally block on the calling thread, releasing the guard at future submission time rather than at future completion time. Stop assuming that the indexing task is executed synchronously from the calling thread. Move inFlightIndexing.remove(id) to a whenComplete() callback on the returned future, so the guard is held for the full duration of the async work. Remove setForwardedEvent/unsetForwardedEvent from the base class entirely and replace it with a withForwardedEventFlag() helper that concrete handlers call when wrapping the actual indexing work, ensuring the flag is set and cleared on the thread that does the work. Change-Id: I76ef61c760595d677db0ebf0f3ae1b057c51da9e
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 bee5f2e..57d6a4b 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
@@ -37,11 +37,16 @@ } @Override - protected CompletableFuture<Boolean> doIndex(Account.Id id, Optional<IndexEvent> indexEvent) { + protected CompletableFuture<Boolean> doIndex(Account.Id id, Optional<IndexEvent> indexEvent) + throws Exception { try { - indexer.index(id); - log.atFine().log("Account %s successfully indexed", id); - } catch (RuntimeException e) { + withForwardedEventFlag( + () -> { + indexer.index(id); + log.atFine().log("Account %s successfully indexed", id); + return null; + }); + } catch (Exception e) { log.atFine().log("Account %s failed to be indexed", id); throw e; }
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 3bf5872..ca9e0f7 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
@@ -65,13 +65,13 @@ protected CompletableFuture<Boolean> doIndex(String id, Optional<IndexEvent> indexEvent) throws IOException { return indexExecutor.getAsync( - () -> { - try (ManualRequestContext ctx = oneOffCtx.open()) { - Context.setForwardedEvent(true); - boolean result = indexOnce(id, indexEvent); - return result; - } - }); + () -> + withForwardedEventFlag( + () -> { + try (ManualRequestContext ctx = oneOffCtx.open()) { + return indexOnce(id, indexEvent); + } + })); } private boolean indexOnce(String id, Optional<IndexEvent> indexEvent) throws Exception {
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 01f8680..8b3c451 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
@@ -38,11 +38,15 @@ @Override protected CompletableFuture<Boolean> doIndex( - AccountGroup.UUID uuid, Optional<IndexEvent> indexEvent) { + AccountGroup.UUID uuid, Optional<IndexEvent> indexEvent) throws Exception { try { - indexer.index(uuid); - log.atFine().log("Group %s successfully indexed", uuid); - } catch (RuntimeException e) { + withForwardedEventFlag( + () -> { + indexer.index(uuid); + log.atFine().log("Group %s successfully indexed", uuid); + return null; + }); + } catch (Exception e) { log.atFine().log("Group %s could not be indexed", uuid); throw e; }
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 b16cb6e..0607f81 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
@@ -38,9 +38,13 @@ @Override protected CompletableFuture<Boolean> doIndex( - Project.NameKey projectName, Optional<IndexEvent> indexEvent) { - indexer.index(projectName); - log.atFine().log("Project %s successfully indexed", projectName); + Project.NameKey projectName, Optional<IndexEvent> indexEvent) throws Exception { + withForwardedEventFlag( + () -> { + indexer.index(projectName); + log.atFine().log("Project %s successfully indexed", projectName); + return null; + }); return CompletableFuture.completedFuture(true); }
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 6ce925a..11d9ecc 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
@@ -15,10 +15,10 @@ package com.ericsson.gerrit.plugins.highavailability.forwarder; import com.google.common.flogger.FluentLogger; -import java.io.IOException; import java.util.Collections; import java.util.Optional; import java.util.Set; +import java.util.concurrent.Callable; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; @@ -43,10 +43,10 @@ } protected abstract CompletableFuture<Boolean> doIndex(T id, Optional<IndexEvent> indexEvent) - throws IOException; + throws Exception; protected abstract CompletableFuture<Boolean> doDelete(T id, Optional<IndexEvent> indexEvent) - throws IOException; + throws Exception; /** * Index an item in the local node, indexing will not be forwarded to the other node. @@ -54,29 +54,41 @@ * @param id The id to index. * @param operation The operation to do; index or delete * @param indexEvent The index event details. - * @throws IOException If an error occur while indexing. + * @throws Exception If an error occur while indexing. */ public CompletableFuture<Boolean> index( - T id, Operation operation, Optional<IndexEvent> indexEvent) throws IOException { + T id, Operation operation, Optional<IndexEvent> indexEvent) throws Exception { log.atFine().log("%s %s %s", operation, id, indexEvent); if (inFlightIndexing.add(id)) { try { - Context.setForwardedEvent(true); + CompletableFuture<Boolean> future; switch (operation) { case INDEX: - return doIndex(id, indexEvent); + future = doIndex(id, indexEvent); + break; case DELETE: - return doDelete(id, indexEvent); + future = doDelete(id, indexEvent); + break; default: log.atSevere().log("unexpected operation: %s", operation); - return CompletableFuture.completedFuture(false); + future = CompletableFuture.completedFuture(false); } - } finally { - Context.unsetForwardedEvent(); + return future.whenComplete((_, _) -> inFlightIndexing.remove(id)); + } catch (Exception e) { inFlightIndexing.remove(id); + throw e; } } throw new InFlightIndexedException( String.format("Indexing for %s %s %s already in flight", operation, id, indexEvent)); } + + protected static <V> V withForwardedEventFlag(Callable<V> action) throws Exception { + Context.setForwardedEvent(true); + try { + return action.call(); + } finally { + Context.unsetForwardedEvent(); + } + } }
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/AbstractIndexRestApiServlet.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/AbstractIndexRestApiServlet.java index 851aa49..84ff1c5 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/AbstractIndexRestApiServlet.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/AbstractIndexRestApiServlet.java
@@ -22,7 +22,6 @@ import com.ericsson.gerrit.plugins.highavailability.forwarder.ProcessorMetricsRegistry; import com.google.gerrit.common.Nullable; import com.google.gerrit.extensions.restapi.NotImplementedException; -import java.io.IOException; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; @@ -31,7 +30,7 @@ @FunctionalInterface interface IndexingOperation { - void execute(String body) throws IOException; + void execute(String body) throws Exception; } public enum IndexName { @@ -64,7 +63,7 @@ op.execute(body); rsp.setStatus(SC_NO_CONTENT); return true; - } catch (IOException e) { + } catch (Exception e) { sendError(rsp, SC_CONFLICT, e.getMessage()); log.atSevere().withCause(e).log("Unable to update %s index", indexName); return false;
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 e549e50..78e32fc 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
@@ -61,9 +61,7 @@ } @Test - public void shouldSetAndUnsetForwardedContext() throws Exception { - // this doAnswer is to allow to assert that context is set to forwarded - // while cache eviction is called. + public void forwardedEventFlagIsSetOnExecutorThreadDuringIndex() throws Exception { doAnswer( (Answer<Void>) invocation -> { @@ -81,7 +79,7 @@ } @Test - public void shouldSetAndUnsetForwardedContextEvenIfExceptionIsThrown() throws Exception { + public void forwardedEventFlagIsUnsetAfterIndexingException() throws Exception { doAnswer( (Answer<Void>) invocation -> { @@ -101,4 +99,26 @@ verify(indexerMock).index(id); } + + @Test + public void inFlightGuardPreventsAndThenAllowsReindex() throws Exception { + doAnswer( + (Answer<Void>) + invocation -> { + // While this first indexing is executing, a second request for the same id + // must be rejected as in-flight. + assertThrows( + InFlightIndexedException.class, + () -> handler.index(id, Operation.INDEX, Optional.empty())); + return null; + }) + .when(indexerMock) + .index(id); + + handler.index(id, Operation.INDEX, Optional.empty()).get(10, SECONDS); + + // Guard is released after completion: a subsequent request must succeed. + handler.index(id, Operation.INDEX, Optional.empty()).get(10, SECONDS); + verify(indexerMock, org.mockito.Mockito.times(2)).index(id); + } }
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 2ad8fed..64464af 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
@@ -41,6 +41,8 @@ import java.io.IOException; import java.time.Duration; import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import org.junit.Before; @@ -149,10 +151,8 @@ } @Test - public void shouldSetAndUnsetForwardedContext() throws Exception { + public void forwardedEventFlagIsSetOnExecutorThreadDuringIndex() throws Exception { setupChangeAccessRelatedMocks(CHANGE_EXISTS, CHANGE_UP_TO_DATE); - // this doAnswer is to allow to assert that context is set to forwarded - // while cache eviction is called. doAnswer( (Answer<Void>) invocation -> { @@ -170,7 +170,7 @@ } @Test - public void shouldSetAndUnsetForwardedContextEvenIfExceptionIsThrown() throws Exception { + public void forwardedEventFlagIsUnsetAfterIndexingException() throws Exception { setupChangeAccessRelatedMocks(CHANGE_EXISTS, CHANGE_UP_TO_DATE); doAnswer( (Answer<Void>) @@ -193,6 +193,59 @@ verify(indexerMock, times(1)).reindexIfStale(any(Project.NameKey.class), any(Change.Id.class)); } + @Test + public void forwardedEventFlagIsNotSetOnCallingThreadDuringAsyncIndex() throws Exception { + setupChangeAccessRelatedMocks(CHANGE_EXISTS, CHANGE_UP_TO_DATE); + CountDownLatch taskRunning = new CountDownLatch(1); + CountDownLatch taskCanFinish = new CountDownLatch(1); + doAnswer( + (Answer<Void>) + invocation -> { + taskRunning.countDown(); + taskCanFinish.await(); + return null; + }) + .when(indexerMock) + .reindexIfStale(any(Project.NameKey.class), any(Change.Id.class)); + + CompletableFuture<Boolean> future = + handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.empty()); + taskRunning.await(10, SECONDS); + assertThat(Context.isForwardedEvent()).isFalse(); + taskCanFinish.countDown(); + future.get(10, SECONDS); + } + + @Test + public void inFlightGuardIsReleasedAfterAsyncIndexCompletes() throws Exception { + setupChangeAccessRelatedMocks(CHANGE_EXISTS, CHANGE_UP_TO_DATE); + CountDownLatch taskRunning = new CountDownLatch(1); + CountDownLatch taskCanFinish = new CountDownLatch(1); + doAnswer( + (Answer<Void>) + invocation -> { + taskRunning.countDown(); + taskCanFinish.await(); + return null; + }) + .when(indexerMock) + .reindexIfStale(any(Project.NameKey.class), any(Change.Id.class)); + + CompletableFuture<Boolean> firstFuture = + handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.empty()); + taskRunning.await(10, SECONDS); + + assertThrows( + InFlightIndexedException.class, + () -> handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.empty())); + + taskCanFinish.countDown(); + firstFuture.get(10, SECONDS); + + // Guard is released: a new request for the same id must now succeed + handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.empty()).get(10, SECONDS); + } + private void setupChangeAccessRelatedMocks(boolean changeExists, boolean changeIsUpToDate) throws IOException { if (changeExists) {
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 babe1d4..8eedffd 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
@@ -61,9 +61,7 @@ } @Test - public void shouldSetAndUnsetForwardedContext() throws Exception { - // this doAnswer is to allow to assert that context is set to forwarded - // while cache eviction is called. + public void forwardedEventFlagIsSetOnExecutorThreadDuringIndex() throws Exception { doAnswer( (Answer<Void>) invocation -> { @@ -81,7 +79,7 @@ } @Test - public void shouldSetAndUnsetForwardedContextEvenIfExceptionIsThrown() throws Exception { + public void forwardedEventFlagIsUnsetAfterIndexingException() throws Exception { doAnswer( (Answer<Void>) invocation -> {
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 f91a83e..d977acb 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
@@ -61,9 +61,7 @@ } @Test - public void shouldSetAndUnsetForwardedContext() throws Exception { - // this doAnswer is to allow to assert that context is set to forwarded - // while cache eviction is called. + public void forwardedEventFlagIsSetOnExecutorThreadDuringIndex() throws Exception { doAnswer( (Answer<Void>) invocation -> { @@ -81,7 +79,7 @@ } @Test - public void shouldSetAndUnsetForwardedContextEvenIfExceptionIsThrown() throws Exception { + public void forwardedEventFlagIsUnsetAfterIndexingException() throws Exception { doAnswer( (Answer<Void>) invocation -> {
diff --git a/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/jgroups/MessageProcessorTest.java b/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/jgroups/MessageProcessorTest.java index 39493a1..58b06bc 100644 --- a/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/jgroups/MessageProcessorTest.java +++ b/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/jgroups/MessageProcessorTest.java
@@ -113,7 +113,7 @@ } @Test - public void indexAccount() throws IOException { + public void indexAccount() throws Exception { int ACCOUNT_ID = 100; IndexAccount cmd = new IndexAccount(ACCOUNT_ID, Instant.now()); @@ -124,7 +124,7 @@ } @Test - public void indexChange() throws IOException { + public void indexChange() throws Exception { String PROJECT = "foo"; int CHANGE_ID = 100; @@ -136,7 +136,7 @@ } @Test - public void indexChangeBatchMode() throws IOException { + public void indexChangeBatchMode() throws Exception { String PROJECT = "foo"; int CHANGE_ID = 100; @@ -148,7 +148,7 @@ } @Test - public void deleteChangeFromIndex() throws IOException { + public void deleteChangeFromIndex() throws Exception { String PROJECT = "foo"; int CHANGE_ID = 100;