Merge branch 'stable-3.14' * stable-3.14: Never throw when describing a queued ITS task Describe ITS change events without resolving the patch set Run ITS rule evaluation asynchronously on a work queue Run ITS actions asynchronously on a work queue Change-Id: If245eb382ba84529e2eeebc80a2076e199b61eee
diff --git a/src/main/java/com/googlesource/gerrit/plugins/its/base/Actions.java b/src/main/java/com/googlesource/gerrit/plugins/its/base/Actions.java new file mode 100644 index 0000000..81a0b27 --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/its/base/Actions.java
@@ -0,0 +1,25 @@ +// Copyright (C) 2026 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.googlesource.gerrit.plugins.its.base; + +import static java.lang.annotation.RetentionPolicy.RUNTIME; + +import com.google.inject.BindingAnnotation; +import java.lang.annotation.Retention; + +/** Qualifier for the pool that applies ITS actions to the issue tracker. */ +@Retention(RUNTIME) +@BindingAnnotation +public @interface Actions {}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/its/base/Evaluation.java b/src/main/java/com/googlesource/gerrit/plugins/its/base/Evaluation.java new file mode 100644 index 0000000..bbd15da --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/its/base/Evaluation.java
@@ -0,0 +1,25 @@ +// Copyright (C) 2026 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.googlesource.gerrit.plugins.its.base; + +import static java.lang.annotation.RetentionPolicy.RUNTIME; + +import com.google.inject.BindingAnnotation; +import java.lang.annotation.Retention; + +/** Qualifier for the pool that detects issue ids and evaluates rules. */ +@Retention(RUNTIME) +@BindingAnnotation +public @interface Evaluation {}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/its/base/ItsHookModule.java b/src/main/java/com/googlesource/gerrit/plugins/its/base/ItsHookModule.java index 8835389..fb26c21 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/its/base/ItsHookModule.java +++ b/src/main/java/com/googlesource/gerrit/plugins/its/base/ItsHookModule.java
@@ -17,15 +17,20 @@ import com.google.gerrit.extensions.annotations.Exports; import com.google.gerrit.extensions.annotations.PluginName; import com.google.gerrit.extensions.config.FactoryModule; +import com.google.gerrit.extensions.events.LifecycleListener; import com.google.gerrit.extensions.registration.DynamicMap; import com.google.gerrit.extensions.registration.DynamicSet; import com.google.gerrit.server.config.PluginConfigFactory; import com.google.gerrit.server.config.ProjectConfigEntry; import com.google.gerrit.server.config.SitePaths; import com.google.gerrit.server.events.EventListener; +import com.google.gerrit.server.git.WorkQueue; import com.google.gerrit.server.git.validators.CommitValidationListener; +import com.google.inject.AbstractModule; import com.google.inject.Inject; +import com.google.inject.Key; import com.google.inject.Provides; +import com.google.inject.Singleton; import com.googlesource.gerrit.plugins.its.base.its.ItsConfig; import com.googlesource.gerrit.plugins.its.base.its.ItsHookEnabledConfigEntry; import com.googlesource.gerrit.plugins.its.base.validation.ItsValidateComment; @@ -35,15 +40,18 @@ import com.googlesource.gerrit.plugins.its.base.workflow.AddPropertyToField; import com.googlesource.gerrit.plugins.its.base.workflow.AddSoyComment; import com.googlesource.gerrit.plugins.its.base.workflow.AddStandardComment; +import com.googlesource.gerrit.plugins.its.base.workflow.BoundedOrderedDispatcher; import com.googlesource.gerrit.plugins.its.base.workflow.Condition; import com.googlesource.gerrit.plugins.its.base.workflow.CreateVersionFromProperty; import com.googlesource.gerrit.plugins.its.base.workflow.CustomAction; import com.googlesource.gerrit.plugins.its.base.workflow.FireEventOnCommits; import com.googlesource.gerrit.plugins.its.base.workflow.ItsRulesProjectCacheImpl; +import com.googlesource.gerrit.plugins.its.base.workflow.LifecycleThreadPool; import com.googlesource.gerrit.plugins.its.base.workflow.LogEvent; import com.googlesource.gerrit.plugins.its.base.workflow.Rule; import com.googlesource.gerrit.plugins.its.base.workflow.commit_collector.SinceLastTagCommitCollector; import java.nio.file.Path; +import java.util.concurrent.Executor; public class ItsHookModule extends FactoryModule { @@ -53,12 +61,24 @@ /** Folder where rules configuration files are located */ private static final String ITS_FOLDER = "its"; + private static final String EVALUATION_SECTION = "evaluation"; + + private static final String ACTIONS_SECTION = "actions"; + + private static final String KEY_THREADS = "threads"; + + private static final int DEFAULT_THREADS = 0; + private final String pluginName; private final PluginConfigFactory pluginCfgFactory; + private final int evaluationThreads; + private final int actionsThreads; public ItsHookModule(@PluginName String pluginName, PluginConfigFactory pluginCfgFactory) { this.pluginName = pluginName; this.pluginCfgFactory = pluginCfgFactory; + this.evaluationThreads = getThreadsFrom(EVALUATION_SECTION); + this.actionsThreads = getThreadsFrom(ACTIONS_SECTION); } @Override @@ -69,6 +89,7 @@ bind(ItsConfig.class); DynamicSet.bind(binder(), CommitValidationListener.class).to(ItsValidateComment.class); DynamicSet.bind(binder(), EventListener.class).to(ActionController.class); + configureExecutors(); factory(ActionRequest.Factory.class); factory(Condition.Factory.class); factory(Rule.Factory.class); @@ -102,4 +123,74 @@ String pluginRulesFileName() { return String.format(CONFIG_FILE_NAME, "-" + pluginName); } + + private int getThreadsFrom(String section) { + return pluginCfgFactory + .getGlobalPluginConfig(pluginName) + .getInt(section, KEY_THREADS, DEFAULT_THREADS); + } + + private void configureExecutors() { + checkThreadPoolConfig(); + if (evaluationThreads > 0) { + install(new EvaluationExecutorModule()); + } else { + bind(Executor.class).annotatedWith(Evaluation.class).toInstance(Runnable::run); + } + if (actionsThreads > 0) { + install(new ActionsExecutorModule()); + } else { + bind(Executor.class).annotatedWith(Actions.class).toInstance(Runnable::run); + } + } + + private void checkThreadPoolConfig() { + if (evaluationThreads > 0 && actionsThreads <= 0) { + addError("evaluation.threads (%d) requires actions.threads > 0", evaluationThreads); + } + } + + private class EvaluationExecutorModule extends AbstractModule { + @Override + protected void configure() { + DynamicSet.bind(binder(), LifecycleListener.class) + .to(Key.get(LifecycleThreadPool.class, Evaluation.class)); + } + + @Provides + @Singleton + @Evaluation + LifecycleThreadPool evaluationThreadPool(WorkQueue workQueue) { + return new LifecycleThreadPool(workQueue, evaluationThreads, pluginName + "-evaluation"); + } + + @Provides + @Singleton + @Evaluation + Executor evaluationExecutor(@Evaluation LifecycleThreadPool pool) { + return new BoundedOrderedDispatcher(pool, evaluationThreads); + } + } + + private class ActionsExecutorModule extends AbstractModule { + @Override + protected void configure() { + DynamicSet.bind(binder(), LifecycleListener.class) + .to(Key.get(LifecycleThreadPool.class, Actions.class)); + } + + @Provides + @Singleton + @Actions + LifecycleThreadPool actionsThreadPool(WorkQueue workQueue) { + return new LifecycleThreadPool(workQueue, actionsThreads, pluginName + "-actions"); + } + + @Provides + @Singleton + @Actions + Executor actionsExecutor(@Actions LifecycleThreadPool pool) { + return new BoundedOrderedDispatcher(pool, actionsThreads); + } + } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/its/base/its/ItsConfig.java b/src/main/java/com/googlesource/gerrit/plugins/its/base/its/ItsConfig.java index 668ff1e..ecc2520 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/its/base/its/ItsConfig.java +++ b/src/main/java/com/googlesource/gerrit/plugins/its/base/its/ItsConfig.java
@@ -66,6 +66,10 @@ currentProjectName.set(projectName); } + public static void clearCurrentProjectName() { + currentProjectName.remove(); + } + @Inject public ItsConfig( @PluginName String pluginName,
diff --git a/src/main/java/com/googlesource/gerrit/plugins/its/base/workflow/ActionController.java b/src/main/java/com/googlesource/gerrit/plugins/its/base/workflow/ActionController.java index d2166a3..74bf8e6 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/its/base/workflow/ActionController.java +++ b/src/main/java/com/googlesource/gerrit/plugins/its/base/workflow/ActionController.java
@@ -15,15 +15,24 @@ package com.googlesource.gerrit.plugins.its.base.workflow; import com.google.common.flogger.FluentLogger; +import com.google.gerrit.entities.Change; +import com.google.gerrit.entities.Project; +import com.google.gerrit.server.events.ChangeEvent; import com.google.gerrit.server.events.Event; import com.google.gerrit.server.events.EventListener; import com.google.gerrit.server.events.RefEvent; import com.google.inject.Inject; +import com.googlesource.gerrit.plugins.its.base.Actions; +import com.googlesource.gerrit.plugins.its.base.Evaluation; import com.googlesource.gerrit.plugins.its.base.its.ItsConfig; import com.googlesource.gerrit.plugins.its.base.util.PropertyExtractor; +import java.util.ArrayList; import java.util.Collection; +import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; +import java.util.concurrent.Executor; /** * Controller that takes actions according to {@code ChangeEvents@}. @@ -39,64 +48,136 @@ private final RuleBase ruleBase; private final ActionExecutor actionExecutor; private final ItsConfig itsConfig; + private final Executor evaluationExecutor; + private final Executor actionsExecutor; @Inject public ActionController( PropertyExtractor propertyExtractor, RuleBase ruleBase, ActionExecutor actionExecutor, - ItsConfig itsConfig) { + ItsConfig itsConfig, + @Evaluation Executor evaluationExecutor, + @Actions Executor actionsExecutor) { this.propertyExtractor = propertyExtractor; this.ruleBase = ruleBase; this.actionExecutor = actionExecutor; this.itsConfig = itsConfig; + this.evaluationExecutor = evaluationExecutor; + this.actionsExecutor = actionsExecutor; } @Override public void onEvent(Event event) { if (event instanceof RefEvent) { RefEvent refEvent = (RefEvent) event; - ItsConfig.setCurrentProjectName(refEvent.getProjectNameKey()); if (itsConfig.isEnabled(refEvent)) { - handleEvent(refEvent); + evaluationExecutor.execute(new EventHandler(refEvent)); } } } - private void handleEvent(RefEvent refEvent) { - RefEventProperties refEventProperties = propertyExtractor.extractFrom(refEvent); + private class EventHandler implements BoundedOrderedDispatcher.OrderedTask { + private final RefEvent refEvent; + private final List<Runnable> actionRunnables = new ArrayList<>(); - handleIssuesEvent(refEventProperties.getIssuesProperties()); - handleProjectEvent(refEventProperties.getProjectProperties()); - } + EventHandler(RefEvent refEvent) { + this.refEvent = refEvent; + } - private void handleIssuesEvent(Set<Map<String, String>> issuesProperties) { - for (Map<String, String> issueProperties : issuesProperties) { - Collection<ActionRequest> actions = ruleBase.actionRequestsFor(issueProperties); - if (!actions.isEmpty()) { - actionExecutor.executeOnIssue(actions, issueProperties); + @Override + public void run() { + ItsConfig.setCurrentProjectName(refEvent.getProjectNameKey()); + try { + RefEventProperties refEventProperties = propertyExtractor.extractFrom(refEvent); + handleIssuesEvent(refEventProperties.getIssuesProperties()); + handleProjectEvent(refEventProperties.getProjectProperties()); + } finally { + ItsConfig.clearCurrentProjectName(); + } + if (!actionRunnables.isEmpty()) { + actionsExecutor.execute(new ActionRunner()); + } + } + + private void handleIssuesEvent(Set<Map<String, String>> issuesProperties) { + for (Map<String, String> issueProperties : issuesProperties) { + Collection<ActionRequest> actions = ruleBase.actionRequestsFor(issueProperties); + if (!actions.isEmpty()) { + actionRunnables.add(() -> actionExecutor.executeOnIssue(actions, issueProperties)); + } + } + } + + private void handleProjectEvent(Map<String, String> projectProperties) { + if (projectProperties.isEmpty()) { + return; + } + + Collection<ActionRequest> projectActions = ruleBase.actionRequestsFor(projectProperties); + if (projectActions.isEmpty()) { + return; + } + if (!projectProperties.containsKey("its-project")) { + String project = projectProperties.get("project"); + logger.atFinest().log( + "Could not process project event. No its-project associated with project %s. " + + "Did you forget to configure the ITS project association in project.config?", + project); + return; + } + + actionRunnables.add(() -> actionExecutor.executeOnProject(projectActions, projectProperties)); + } + + @Override + public Optional<ChangeKey> key() { + return ChangeKey.optionallyFrom(refEvent); + } + + @Override + public String toString() { + return "its-evaluation: " + refEventToString(); + } + + private String refEventToString() { + String target = refEvent.getBranchNameKey().toString(); + if (refEvent instanceof ChangeEvent changeEvent) { + target = target + " " + changeEvent.getChangeKey(); + } + return refEvent.getType() + " " + target; + } + + private class ActionRunner implements BoundedOrderedDispatcher.OrderedTask { + @Override + public void run() { + ItsConfig.setCurrentProjectName(refEvent.getProjectNameKey()); + try { + actionRunnables.forEach(Runnable::run); + } finally { + ItsConfig.clearCurrentProjectName(); + } + } + + @Override + public Optional<ChangeKey> key() { + return ChangeKey.optionallyFrom(refEvent); + } + + @Override + public String toString() { + return "its-actions: " + refEventToString(); } } } - private void handleProjectEvent(Map<String, String> projectProperties) { - if (projectProperties.isEmpty()) { - return; + public record ChangeKey(Project.NameKey projectName, String destRefName, Change.Key changeKey) { + public static Optional<ChangeKey> optionallyFrom(RefEvent event) { + if (!(event instanceof ChangeEvent changeEvent)) { + return Optional.empty(); + } + return Optional.of( + new ChangeKey(event.getProjectNameKey(), event.getRefName(), changeEvent.getChangeKey())); } - - Collection<ActionRequest> projectActions = ruleBase.actionRequestsFor(projectProperties); - if (projectActions.isEmpty()) { - return; - } - if (!projectProperties.containsKey("its-project")) { - String project = projectProperties.get("project"); - logger.atFinest().log( - "Could not process project event. No its-project associated with project %s. " - + "Did you forget to configure the ITS project association in project.config?", - project); - return; - } - - actionExecutor.executeOnProject(projectActions, projectProperties); } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/its/base/workflow/BoundedOrderedDispatcher.java b/src/main/java/com/googlesource/gerrit/plugins/its/base/workflow/BoundedOrderedDispatcher.java new file mode 100644 index 0000000..cf96e55 --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/its/base/workflow/BoundedOrderedDispatcher.java
@@ -0,0 +1,52 @@ +// Copyright (C) 2026 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.googlesource.gerrit.plugins.its.base.workflow; + +import java.util.Optional; +import java.util.concurrent.Executor; +import javax.annotation.Nonnull; + +/** + * Runs ITS event tasks on the {@link LifecycleThreadPool}, bounding the backlog and serializing + * tasks that share an ordering key. Tasks that implement {@link OrderedTask} and return the same + * key run one at a time in submission order and tasks that are not {@link OrderedTask} (or return + * an empty key) run in parallel. + */ +public class BoundedOrderedDispatcher implements Executor { + private final LifecycleThreadPool pool; + private final Throttle throttle; + private final RuntimeQueueMap<Object> runtimeQueueMap; + + public BoundedOrderedDispatcher(LifecycleThreadPool pool, int maxInFlight) { + this.pool = pool; + this.throttle = new Throttle(maxInFlight); + this.runtimeQueueMap = new RuntimeQueueMap<>(); + } + + @Override + public void execute(@Nonnull Runnable task) { + Optional<?> orderingKey = + task instanceof OrderedTask orderedTask ? orderedTask.key() : Optional.empty(); + Executor taskRunner = + orderingKey.isEmpty() + ? pool + : runnable -> pool.execute(runtimeQueueMap.wrap(orderingKey.get(), runnable)); + throttle.execute(taskRunner, task); + } + + public interface OrderedTask extends Runnable { + Optional<?> key(); + } +}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/its/base/workflow/LifecycleThreadPool.java b/src/main/java/com/googlesource/gerrit/plugins/its/base/workflow/LifecycleThreadPool.java new file mode 100644 index 0000000..b483a1d --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/its/base/workflow/LifecycleThreadPool.java
@@ -0,0 +1,43 @@ +// Copyright (C) 2026 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.googlesource.gerrit.plugins.its.base.workflow; + +import com.google.gerrit.extensions.events.LifecycleListener; +import com.google.gerrit.server.git.WorkQueue; +import java.util.concurrent.Executor; +import java.util.concurrent.ScheduledExecutorService; +import javax.annotation.Nonnull; + +/** Thread pool that runs event tasks and shuts the pool down with the plugin's lifecycle. */ +public class LifecycleThreadPool implements Executor, LifecycleListener { + private final ScheduledExecutorService executor; + + public LifecycleThreadPool(WorkQueue workQueue, int poolSize, String queueName) { + this.executor = workQueue.createQueue(poolSize, queueName); + } + + @Override + public void execute(@Nonnull Runnable task) { + executor.execute(task); + } + + @Override + public void start() {} + + @Override + public void stop() { + executor.shutdown(); + } +}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/its/base/workflow/RuntimeQueueMap.java b/src/main/java/com/googlesource/gerrit/plugins/its/base/workflow/RuntimeQueueMap.java new file mode 100644 index 0000000..52f06e6 --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/its/base/workflow/RuntimeQueueMap.java
@@ -0,0 +1,135 @@ +// Copyright (C) 2026 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.googlesource.gerrit.plugins.its.base.workflow; + +import com.google.common.flogger.FluentLogger; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ExecutionException; + +/** + * Serializes {@link Runnable}s that share a key while letting tasks with different keys run + * concurrently. + * + * <p>A caller {@link #wrap(Object, Runnable) wraps} a task together with its key and hands the + * returned {@link Runnable} to an executor. Tasks that share a key never run concurrently. A task + * that arrives while another with the same key is running is queued and executed after it, in + * arrival order. Tasks with different keys are independent and may run in parallel. + * + * <p>Ordering is achieved without blocking executor threads. The first task to arrive for an + * otherwise idle key runs on the executor thread that picked it up and then drains any tasks that + * queued up behind it. While that chain is draining, further submissions for the same key hand + * themselves off to the running chain and return immediately, so at most one thread is ever + * occupied per key. Once a key's chain empties, its entry is removed, so the map only contains keys + * with work in flight. + * + * @param <K> type of the key that groups tasks which must run sequentially + */ +public class RuntimeQueueMap<K> { + private static final FluentLogger logger = FluentLogger.forEnclosingClass(); + + private final ConcurrentMap<K, CompletableFuture<Task>> map = new ConcurrentHashMap<>(); + + /** + * Wraps a task so that, once the returned {@link Runnable} is executed, it runs sequentially with + * respect to every other task wrapped under the same key. + * + * @param k key whose tasks must run one at a time + * @param mine task to run + * @return a {@link Runnable} to hand to the backing executor + */ + public Runnable wrap(K k, Runnable mine) { + return new Task(k, mine); + } + + private class Task implements Runnable { + private final K key; + private final Runnable task; + + final CompletableFuture<Task> nextFuture = new CompletableFuture<>(); + private volatile Task activeTask; + + Task(K key, Runnable task) { + this.key = key; + this.task = task; + this.activeTask = this; + } + + @Override + public void run() { + if (offerIfNotFirst()) { + return; + } + + Task current = this; + while (true) { + try { + current.task.run(); + } catch (Throwable t) { + logger.atSevere().withCause(t).log("Error executing task for key: %s", key); + } + + if (map.remove(key, current.nextFuture)) { + break; + } + this.activeTask = current = getAlways(current.nextFuture); + } + } + + private boolean offerIfNotFirst() { + CompletableFuture<Task> prevFuture = map.put(key, nextFuture); + if (prevFuture == null) { + return false; + } + + prevFuture.complete(this); + return true; + } + + private Task getAlways(CompletableFuture<Task> future) { + while (true) { + try { + return future.get(); + } catch (InterruptedException e) { + // A successor task has already claimed this slot, so the future is guaranteed to + // complete. Keep waiting for the hand-off. Any interrupt observed here was aimed at the + // previous task, which has already finished running, so it does not apply to this + // internal wait and is cleared and ignored. + Thread.interrupted(); + } catch (ExecutionException e) { + // should never reach here + logger.atSevere().withCause(e).log("Pipeline future failed unexpectedly"); + throw new RuntimeException(e); + } + } + } + + @Override + public String toString() { + try { + Runnable current = activeTask.task; + try { + return current.toString(); + } catch (Exception e) { + logger.atWarning().withCause(e).log("Cannot describe task"); + return current.getClass().getName(); + } + } catch (Exception e) { + return "unknown task"; + } + } + } +}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/its/base/workflow/Throttle.java b/src/main/java/com/googlesource/gerrit/plugins/its/base/workflow/Throttle.java new file mode 100644 index 0000000..0286570 --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/its/base/workflow/Throttle.java
@@ -0,0 +1,84 @@ +// Copyright (C) 2026 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.googlesource.gerrit.plugins.its.base.workflow; + +import com.google.common.flogger.FluentLogger; +import java.util.concurrent.Executor; +import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Bounds the number of tasks in flight on an executor, so that a caller submitting tasks is made to + * wait once the backlog reaches the limit. + */ +public class Throttle { + private static final FluentLogger logger = FluentLogger.forEnclosingClass(); + + private final Semaphore inflight; + + Throttle(int permits) { + this.inflight = new Semaphore(permits); + } + + void execute(Executor executor, Runnable task) { + ThrottledTask throttledTask = new ThrottledTask(task); + try { + executor.execute(throttledTask); + } catch (RuntimeException e) { + logger.atWarning().withCause(e).log("Failed to queue ITS task %s", task); + return; + } + inflight.acquireUninterruptibly(); + throttledTask.releaseIfReady(); + } + + private final class ThrottledTask implements Runnable { + private final Runnable task; + private final AtomicBoolean readyForRelease = new AtomicBoolean(); + + private ThrottledTask(Runnable task) { + this.task = task; + } + + @Override + public void run() { + try { + task.run(); + } finally { + releaseIfReady(); + } + } + + private void releaseIfReady() { + if (readyForRelease.getAndSet(true)) { + inflight.release(); + } + } + + @Override + public String toString() { + try { + try { + return task.toString(); + } catch (Exception e) { + logger.atWarning().withCause(e).log("Cannot describe task"); + return task.getClass().getName(); + } + } catch (Exception e) { + return "unknown task"; + } + } + } +}
diff --git a/src/main/resources/Documentation/config-common.md b/src/main/resources/Documentation/config-common.md index c8d8339..8f0b5cd 100644 --- a/src/main/resources/Documentation/config-common.md +++ b/src/main/resources/Documentation/config-common.md
@@ -8,6 +8,7 @@ - [Associating a Gerrit project with its ITS project counterpart](#associating-a-gerrit-project-with-its-its-project-counterpart) - [Configuring rules of when to take which actions in the ITS](#configuring-rules-of-when-to-take-which-actions-in-the-its) - [Multiple Its](#multiple-its) +- [Asynchronous event processing](#asynchronous-event-processing) - [Further common configuration details](#further-common-configuration-details) @@ -179,6 +180,73 @@ jar --verbose --create --manifest=META-INF/MANIFEST.MF --file=../its-bugzilla-external.jar . ``` +## Asynchronous event processing + +By default, @PLUGIN@ handles ITS events synchronously on Gerrit's event-dispatch +thread. Handling an event has two phases: + +* *evaluation*: detecting issue ids and evaluating the configured rules to decide + which actions to take +* *actions*: applying those actions, which make blocking calls to the issue + tracker + +Both phases run on the event-dispatch thread by default and can delay the Gerrit +operation that produced the event. Each phase can instead be handed off to its +own dedicated thread pool by setting a pool size in the plugin's own +configuration file `etc/@PLUGIN@.config`: + +```ini +[evaluation] + threads = 50 +[actions] + threads = 10 +``` + +<a name="common-config-evaluationThreads">`evaluation.threads`</a> +: The number of threads @PLUGIN@ uses to detect issue ids and evaluate the + configured rules asynchronously. + + When set to `0`, evaluation runs synchronously on Gerrit's event-dispatch + thread, blocking it until issue detection and rule evaluation complete. + + The events of a single change are always evaluated one at a time, in the + order they occurred, even when run asynchronously on the pool, so that the + resulting actions are applied in order. + + When set to a positive value, evaluation runs asynchronously on a pool of up + to that many threads, so at most that many evaluation tasks run at once. Once + all threads are busy, the event-dispatch thread blocks until a running task + completes, so the amount of queued work stays bounded. + + Setting `evaluation.threads` to a positive value requires `actions.threads` + to also be positive, so that the blocking issue tracker calls run on the + actions pool rather than on the evaluation pool. It is recommended that the + evaluation pool be larger than the actions pool, since all events are + evaluated but only some result in actions. + + Default is `0` + +<a name="common-config-actionsThreads">`actions.threads`</a> +: The number of threads @PLUGIN@ uses to apply ITS actions asynchronously. + + Only the resulting actions, which update the issue tracker, are handed off to + this thread pool, and only for events that trigger at least one action. + + When set to `0`, asynchronous processing is disabled and @PLUGIN@ runs the + actions synchronously on the event-dispatch thread, blocking it until the + issue tracker calls complete. + + The actions of a single change are always serialized, even when applied + asynchronously on the pool, so that issue tracker state transitions work as + expected. + + When set to a positive value, actions are applied asynchronously on a pool + of up to that many threads, so at most that many action tasks run at once. + Once all threads are busy, the event-dispatch thread blocks until a running + task completes, so the amount of queued work stays bounded. + + Default is `0` + ## Further common configuration details [common-config-commentlink](#common-config-commentlink)
diff --git a/src/test/java/com/googlesource/gerrit/plugins/its/base/workflow/ActionControllerTest.java b/src/test/java/com/googlesource/gerrit/plugins/its/base/workflow/ActionControllerTest.java index 27545a2..728b2f8 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/its/base/workflow/ActionControllerTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/its/base/workflow/ActionControllerTest.java
@@ -27,6 +27,8 @@ import com.google.gerrit.server.events.RefEvent; import com.google.inject.Guice; import com.google.inject.Injector; +import com.googlesource.gerrit.plugins.its.base.Actions; +import com.googlesource.gerrit.plugins.its.base.Evaluation; import com.googlesource.gerrit.plugins.its.base.its.ItsConfig; import com.googlesource.gerrit.plugins.its.base.testutil.LoggingMockingTestCase; import com.googlesource.gerrit.plugins.its.base.util.PropertyExtractor; @@ -35,6 +37,7 @@ import java.util.HashSet; import java.util.Map; import java.util.Set; +import java.util.concurrent.Executor; public class ActionControllerTest extends LoggingMockingTestCase { private static Project.NameKey testProjectName = Project.nameKey("test-project"); @@ -183,6 +186,9 @@ itsConfig = mock(ItsConfig.class); bind(ItsConfig.class).toInstance(itsConfig); + + bind(Executor.class).annotatedWith(Evaluation.class).toInstance(Runnable::run); + bind(Executor.class).annotatedWith(Actions.class).toInstance(Runnable::run); } } }