Reintroduce Forwarded(Batch)IndexExecutor for indexing threads Introduce the well-known executors with associated queues for running the reindexing tasks associated with the forwarded tasks form the HA nodes. The executors and associated queues were removed in If5ec4a1ed but the Forwarded(Batch)IndexExecutor class names kept with a different implementation on top of the async-retry with failsafe. The well-known executors names, lost during the refactoring introduced with If5ec4a1ed, which has caused two issues: a) The indexing tasks were not visible anymore in the executors queues, which made problematic managing the queues and giving the associated visibility to the Gerrit admin. b) Other plugins, like multi-site, were unaware that the indexing was happening because of the HA index forwarding, generating a continuous loop of indexing events between HA and multi-site. Reintroduce the proper executors in their original class names, made available thanks to the renaming made in I18cdfcaf, and add the UsedAt() notation as sentinel for preventing the executors to be removed unilaterally again. Change-Id: I7d4735155fb3d84cb445dfd4612bfbe90716766a
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 new file mode 100644 index 0000000..a3a7e64 --- /dev/null +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexExecutorProvider.java
@@ -0,0 +1,36 @@ +// Copyright (C) 2024 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.index; + +import com.ericsson.gerrit.plugins.highavailability.Configuration; +import com.ericsson.gerrit.plugins.highavailability.ExecutorProvider; +import com.google.gerrit.common.UsedAt; +import com.google.gerrit.server.git.WorkQueue; +import com.google.inject.Inject; +import com.google.inject.Singleton; + +@Singleton +class ForwardedBatchIndexExecutorProvider extends ExecutorProvider { + + @UsedAt(UsedAt.Project.PLUGIN_MULTI_SITE) + public static final String FORWARDED_BATCH_INDEX_EVENT_THREAD_PREFIX = + "Forwarded-BatchIndex-Event"; + + @Inject + ForwardedBatchIndexExecutorProvider(WorkQueue workQueue, Configuration config) { + super( + workQueue, config.index().batchThreadPoolSize(), FORWARDED_BATCH_INDEX_EVENT_THREAD_PREFIX); + } +}
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexFailsafeExecutorProvider.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexFailsafeExecutorProvider.java index d9afd1a..f52b3fe 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexFailsafeExecutorProvider.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexFailsafeExecutorProvider.java
@@ -15,6 +15,7 @@ package com.ericsson.gerrit.plugins.highavailability.index; import com.ericsson.gerrit.plugins.highavailability.Configuration; +import com.ericsson.gerrit.plugins.highavailability.ExecutorProvider; import com.google.inject.Inject; import com.google.inject.Singleton; @@ -22,12 +23,8 @@ class ForwardedBatchIndexFailsafeExecutorProvider extends ForwardedIndexFailsafeExecutorProvider { @Inject - ForwardedBatchIndexFailsafeExecutorProvider(Configuration cfg) { - super(cfg); - } - - @Override - protected int threadPoolSize(Configuration cfg) { - return cfg.index().batchThreadPoolSize(); + ForwardedBatchIndexFailsafeExecutorProvider( + Configuration cfg, @ForwardedBatchIndexExecutor ExecutorProvider indexExecutorProvider) { + super(cfg, indexExecutorProvider); } }
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 new file mode 100644 index 0000000..da623df --- /dev/null +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedIndexExecutorProvider.java
@@ -0,0 +1,34 @@ +// Copyright (C) 2018 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.index; + +import com.ericsson.gerrit.plugins.highavailability.Configuration; +import com.ericsson.gerrit.plugins.highavailability.ExecutorProvider; +import com.google.gerrit.common.UsedAt; +import com.google.gerrit.server.git.WorkQueue; +import com.google.inject.Inject; +import com.google.inject.Singleton; + +@Singleton +public class ForwardedIndexExecutorProvider extends ExecutorProvider { + + @UsedAt(UsedAt.Project.PLUGIN_MULTI_SITE) + public static final String FORWARDED_INDEX_EVENT_THREAD_PREFIX = "Forwarded-Index-Event"; + + @Inject + ForwardedIndexExecutorProvider(WorkQueue workQueue, Configuration config) { + super(workQueue, config.index().threadPoolSize(), FORWARDED_INDEX_EVENT_THREAD_PREFIX); + } +}
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedIndexFailsafeExecutorProvider.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedIndexFailsafeExecutorProvider.java index 3260ae6..7bae94a 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedIndexFailsafeExecutorProvider.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedIndexFailsafeExecutorProvider.java
@@ -15,6 +15,7 @@ package com.ericsson.gerrit.plugins.highavailability.index; import com.ericsson.gerrit.plugins.highavailability.Configuration; +import com.ericsson.gerrit.plugins.highavailability.ExecutorProvider; import com.google.common.flogger.FluentLogger; import com.google.inject.Inject; import com.google.inject.Provider; @@ -23,16 +24,18 @@ import dev.failsafe.FailsafeExecutor; import dev.failsafe.RetryPolicy; import java.io.IOException; -import java.util.concurrent.Executors; @Singleton public class ForwardedIndexFailsafeExecutorProvider implements Provider<FailsafeExecutor<Boolean>> { protected static final FluentLogger log = FluentLogger.forEnclosingClass(); private final Configuration cfg; + private final ExecutorProvider indexExecutorProvider; @Inject - public ForwardedIndexFailsafeExecutorProvider(Configuration cfg) { + public ForwardedIndexFailsafeExecutorProvider( + Configuration cfg, @ForwardedIndexExecutor ExecutorProvider indexExecutorProvider) { this.cfg = cfg; + this.indexExecutorProvider = indexExecutorProvider; } @Override @@ -49,11 +52,6 @@ .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(); + return Failsafe.with(retryPolicy).with(indexExecutorProvider.get()); } }
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 cc55bbe..1db3721 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
@@ -14,6 +14,7 @@ package com.ericsson.gerrit.plugins.highavailability.index; +import com.ericsson.gerrit.plugins.highavailability.ExecutorProvider; import com.google.gerrit.extensions.events.AccountIndexedListener; import com.google.gerrit.extensions.events.ChangeIndexedListener; import com.google.gerrit.extensions.events.GroupIndexedListener; @@ -33,11 +34,19 @@ .annotatedWith(ForwardedIndexExecutor.class) .toProvider(ForwardedIndexFailsafeExecutorProvider.class) .in(Scopes.SINGLETON); + bind(ExecutorProvider.class) + .annotatedWith(ForwardedIndexExecutor.class) + .to(ForwardedIndexExecutorProvider.class) + .in(Scopes.SINGLETON); bind(new TypeLiteral<FailsafeExecutor<Boolean>>() {}) .annotatedWith(ForwardedBatchIndexExecutor.class) .toProvider(ForwardedBatchIndexFailsafeExecutorProvider.class) .in(Scopes.SINGLETON); + bind(ExecutorProvider.class) + .annotatedWith(ForwardedBatchIndexExecutor.class) + .to(ForwardedBatchIndexExecutorProvider.class) + .in(Scopes.SINGLETON); DynamicSet.bind(binder(), ChangeIndexedListener.class) .to(IndexEventHandler.class)
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 2a8864d..427985a 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
@@ -29,6 +29,7 @@ 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.ericsson.gerrit.plugins.highavailability.index.ForwardedIndexFailsafeExecutorProvider; import com.google.gerrit.entities.Change; import com.google.gerrit.server.index.change.ChangeIndexer; @@ -39,6 +40,7 @@ import java.time.Duration; import java.util.Optional; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -67,6 +69,7 @@ @Mock private ChangeCheckerImpl.Factory changeCheckerFactoryMock; @Mock private ChangeChecker changeCheckerAbsentMock; @Mock private ChangeChecker changeCheckerPresentMock; + @Mock private ForwardedIndexExecutorProvider indexExecutorProviderMock; private ForwardedIndexChangeHandler handler; private Change.Id id; @@ -77,8 +80,9 @@ when(configMock.index().maxTries()).thenReturn(3); when(configMock.index().retryInterval()).thenReturn(Duration.ofMillis(10)); when(changeCheckerFactoryMock.create(any())).thenReturn(changeCheckerAbsentMock); + when(indexExecutorProviderMock.get()).thenReturn(Executors.newScheduledThreadPool(2)); FailsafeExecutor<Boolean> indexExecutor = - new ForwardedIndexFailsafeExecutorProvider(configMock).get(); + new ForwardedIndexFailsafeExecutorProvider(configMock, indexExecutorProviderMock).get(); handler = new ForwardedIndexChangeHandler( indexerMock, indexExecutor, ctxMock, changeCheckerFactoryMock);