Use dedicated Executor Service for batch indexing Implement separate flow for batch and interactive indexing on active and passive node. Introduce new BatchChangeHandler with separate indexExecutor to split batch and interactive index thread pools. Splitting interactive and batch indexing will reduce latency of the interactive indexing events during massive batch indexing tasks. Without this change there is a huge possibility that an new interactive indexing event will reach the passive node after batch backlog queue is empty. Bug: Issue 12500 Change-Id: I4cefa1f5f3c30cf8743bda6f626381f0caa48d2c
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 new file mode 100644 index 0000000..a6cbb7f --- /dev/null +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexBatchChangeHandler.java
@@ -0,0 +1,40 @@ +// Copyright (C) 2021 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; + +import com.ericsson.gerrit.plugins.highavailability.Configuration; +import com.ericsson.gerrit.plugins.highavailability.index.ChangeCheckerImpl.Factory; +import com.ericsson.gerrit.plugins.highavailability.index.ChangeDb; +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; + +@Singleton +public class ForwardedIndexBatchChangeHandler extends ForwardedIndexChangeHandler { + + @Inject + ForwardedIndexBatchChangeHandler( + ChangeIndexer indexer, + ChangeDb changeDb, + Configuration config, + @ForwardedBatchIndexExecutor ScheduledExecutorService indexExecutor, + OneOffRequestContext oneOffCtx, + Factory changeCheckerFactory) { + super(indexer, changeDb, config, indexExecutor, oneOffCtx, changeCheckerFactory); + } +}
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/IndexBatchChangeRestApiServlet.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/IndexBatchChangeRestApiServlet.java index c8518eb..6613ee6 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/IndexBatchChangeRestApiServlet.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/IndexBatchChangeRestApiServlet.java
@@ -14,7 +14,7 @@ package com.ericsson.gerrit.plugins.highavailability.forwarder.rest; -import com.ericsson.gerrit.plugins.highavailability.forwarder.ForwardedIndexChangeHandler; +import com.ericsson.gerrit.plugins.highavailability.forwarder.ForwardedIndexBatchChangeHandler; import com.google.gerrit.extensions.restapi.Url; import com.google.inject.Inject; import com.google.inject.Singleton; @@ -24,7 +24,7 @@ private static final long serialVersionUID = -1L; @Inject - IndexBatchChangeRestApiServlet(ForwardedIndexChangeHandler handler) { + IndexBatchChangeRestApiServlet(ForwardedIndexBatchChangeHandler handler) { super(handler, IndexName.CHANGE, true); }
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/BatchIndexExecutor.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/BatchIndexExecutor.java new file mode 100644 index 0000000..06da9f0 --- /dev/null +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/BatchIndexExecutor.java
@@ -0,0 +1,24 @@ +// Copyright (C) 2021 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 static java.lang.annotation.RetentionPolicy.RUNTIME; + +import com.google.inject.BindingAnnotation; +import java.lang.annotation.Retention; + +@Retention(RUNTIME) +@BindingAnnotation +@interface BatchIndexExecutor {}
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/BatchIndexExecutorProvider.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/BatchIndexExecutorProvider.java new file mode 100644 index 0000000..ef8d13f --- /dev/null +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/BatchIndexExecutorProvider.java
@@ -0,0 +1,31 @@ +// Copyright (C) 2021 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.server.git.WorkQueue; +import com.google.inject.Inject; +import com.google.inject.Singleton; + +@Singleton +class BatchIndexExecutorProvider extends ExecutorProvider { + + @Inject + BatchIndexExecutorProvider(WorkQueue workQueue, Configuration config) { + // TODO: Consider using different threadPollSize. + super(workQueue, config.index().threadPoolSize(), "Forward-BatchIndex-Event"); + } +}
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexExecutor.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexExecutor.java new file mode 100644 index 0000000..b5b6aab --- /dev/null +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexExecutor.java
@@ -0,0 +1,24 @@ +// Copyright (C) 2021 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 static java.lang.annotation.RetentionPolicy.RUNTIME; + +import com.google.inject.BindingAnnotation; +import java.lang.annotation.Retention; + +@Retention(RUNTIME) +@BindingAnnotation +public @interface ForwardedBatchIndexExecutor {}
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..f94b575 --- /dev/null +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexExecutorProvider.java
@@ -0,0 +1,31 @@ +// Copyright (C) 2021 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.server.git.WorkQueue; +import com.google.inject.Inject; +import com.google.inject.Singleton; + +@Singleton +class ForwardedBatchIndexExecutorProvider extends ExecutorProvider { + + @Inject + ForwardedBatchIndexExecutorProvider(WorkQueue workQueue, Configuration config) { + // TODO: config new variable should be added to define separate thread pool size + super(workQueue, config.index().threadPoolSize(), "Forwarded-BatchIndex-Event"); + } +}
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 633a098..068ef0d 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
@@ -37,6 +37,7 @@ ProjectIndexedListener { private static final FluentLogger log = FluentLogger.forEnclosingClass(); private final Executor executor; + private final Executor batchExecutor; private final Forwarder forwarder; private final String pluginName; private final Set<IndexTask> queuedTasks = Collections.newSetFromMap(new ConcurrentHashMap<>()); @@ -46,12 +47,14 @@ @Inject IndexEventHandler( @IndexExecutor Executor executor, + @BatchIndexExecutor Executor batchExecutor, @PluginName String pluginName, Forwarder forwarder, ChangeCheckerImpl.Factory changeChecker, CurrentRequestContext currCtx) { this.forwarder = forwarder; this.executor = executor; + this.batchExecutor = batchExecutor; this.pluginName = pluginName; this.changeChecker = changeChecker; this.currCtx = currCtx; @@ -93,7 +96,11 @@ .ifPresent( task -> { if (queuedTasks.add(task)) { - executor.execute(task); + if (task instanceof BatchIndexChangeTask) { + batchExecutor.execute(task); + } else { + executor.execute(task); + } } }); } catch (Exception e) {
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 c17731f..4258dce 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
@@ -29,9 +29,15 @@ @Override protected void configure() { bind(Executor.class).annotatedWith(IndexExecutor.class).toProvider(IndexExecutorProvider.class); + bind(Executor.class) + .annotatedWith(BatchIndexExecutor.class) + .toProvider(BatchIndexExecutorProvider.class); bind(ScheduledExecutorService.class) .annotatedWith(ForwardedIndexExecutor.class) .toProvider(ForwardedIndexExecutorProvider.class); + bind(ScheduledExecutorService.class) + .annotatedWith(ForwardedBatchIndexExecutor.class) + .toProvider(ForwardedBatchIndexExecutorProvider.class); listener().to(IndexExecutorProvider.class); DynamicSet.bind(binder(), ChangeIndexedListener.class).to(IndexEventHandler.class); DynamicSet.bind(binder(), AccountIndexedListener.class).to(IndexEventHandler.class);
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 2d12ca8..4c20ae1 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
@@ -89,6 +89,7 @@ indexEventHandler = new IndexEventHandler( MoreExecutors.directExecutor(), + MoreExecutors.directExecutor(), PLUGIN_NAME, forwarder, changeCheckerFactoryMock, @@ -178,8 +179,10 @@ @Test public void duplicateChangeEventOfAQueuedEventShouldGetDiscarded() { ScheduledThreadPoolExecutor poolMock = mock(ScheduledThreadPoolExecutor.class); + ScheduledThreadPoolExecutor poolBatchMock = mock(ScheduledThreadPoolExecutor.class); indexEventHandler = - new IndexEventHandler(poolMock, PLUGIN_NAME, forwarder, changeCheckerFactoryMock, currCtx); + new IndexEventHandler( + poolMock, poolBatchMock, PLUGIN_NAME, forwarder, changeCheckerFactoryMock, currCtx); indexEventHandler.onChangeIndexed(PROJECT_NAME, changeId.get()); indexEventHandler.onChangeIndexed(PROJECT_NAME, changeId.get()); verify(poolMock, times(1)) @@ -189,8 +192,10 @@ @Test public void duplicateAccountEventOfAQueuedEventShouldGetDiscarded() { ScheduledThreadPoolExecutor poolMock = mock(ScheduledThreadPoolExecutor.class); + ScheduledThreadPoolExecutor poolBatchMock = mock(ScheduledThreadPoolExecutor.class); indexEventHandler = - new IndexEventHandler(poolMock, PLUGIN_NAME, forwarder, changeCheckerFactoryMock, currCtx); + new IndexEventHandler( + poolMock, poolBatchMock, PLUGIN_NAME, forwarder, changeCheckerFactoryMock, currCtx); indexEventHandler.onAccountIndexed(accountId.get()); indexEventHandler.onAccountIndexed(accountId.get()); verify(poolMock, times(1)).execute(indexEventHandler.new IndexAccountTask(ACCOUNT_ID)); @@ -199,8 +204,10 @@ @Test public void duplicateGroupEventOfAQueuedEventShouldGetDiscarded() { ScheduledThreadPoolExecutor poolMock = mock(ScheduledThreadPoolExecutor.class); + ScheduledThreadPoolExecutor poolBatchMock = mock(ScheduledThreadPoolExecutor.class); indexEventHandler = - new IndexEventHandler(poolMock, PLUGIN_NAME, forwarder, changeCheckerFactoryMock, currCtx); + new IndexEventHandler( + poolMock, poolBatchMock, PLUGIN_NAME, forwarder, changeCheckerFactoryMock, currCtx); indexEventHandler.onGroupIndexed(accountGroupUUID.get()); indexEventHandler.onGroupIndexed(accountGroupUUID.get()); verify(poolMock, times(1)).execute(indexEventHandler.new IndexGroupTask(UUID));