SONAR-9525 CeWorker can be disabled by EnabledCeWorkerController

in which case they won't attempt to retrieve a task to process and return immediatly

6 files changed, 90 insertions, 4 deletions

diff --git a/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeTaskProcessorModule.java b/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeTaskProcessorModule.java
index 9163507abe9..7b07618fd7b 100644
--- a/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeTaskProcessorModule.java
+++ b/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeTaskProcessorModule.java
@@ -27,6 +27,7 @@ public class CeTaskProcessorModule extends Module {
     add(
       CeTaskProcessorRepositoryImpl.class,
       CeWorkerFactoryImpl.class,
+      EnabledCeWorkerControllerImpl.class,
       CeProcessingSchedulerExecutorServiceImpl.class,
       CeProcessingSchedulerImpl.class);
   }
diff --git a/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorker.java b/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorker.java
index 9a7ffff9776..9db5a69ef2c 100644
--- a/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorker.java
+++ b/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorker.java
@@ -30,6 +30,8 @@ import org.sonar.ce.queue.CeTask;
  */
 public interface CeWorker extends Callable<CeWorker.Result> {
   enum Result {
+    /** Worker is disabled */
+    DISABLED,
     /** Worker found no task to process */
     NO_TASK,
     /** Worker found a task and processed it (either successfully or not) */
diff --git a/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorkerFactoryImpl.java b/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorkerFactoryImpl.java
index 21228bbc4cc..ed8e11cd3c5 100644
--- a/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorkerFactoryImpl.java
+++ b/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorkerFactoryImpl.java
@@ -34,19 +34,22 @@ public class CeWorkerFactoryImpl implements CeWorkerFactory {
   private final InternalCeQueue queue;
   private final CeLogging ceLogging;
   private final CeTaskProcessorRepository taskProcessorRepository;
+  private final EnabledCeWorkerController enabledCeWorkerController;
 
-  public CeWorkerFactoryImpl(InternalCeQueue queue, CeLogging ceLogging, CeTaskProcessorRepository taskProcessorRepository, UuidFactory uuidFactory) {
+  public CeWorkerFactoryImpl(InternalCeQueue queue, CeLogging ceLogging, CeTaskProcessorRepository taskProcessorRepository,
+    UuidFactory uuidFactory, EnabledCeWorkerController enabledCeWorkerController) {
     this.queue = queue;
     this.ceLogging = ceLogging;
     this.taskProcessorRepository = taskProcessorRepository;
-    this.uuidFactory= uuidFactory;
+    this.uuidFactory = uuidFactory;
+    this.enabledCeWorkerController = enabledCeWorkerController;
   }
 
   @Override
   public CeWorker create(int ordinal) {
     String uuid = uuidFactory.create();
     ceWorkerUUIDs.add(uuid);
-    return new CeWorkerImpl(ordinal, uuid, queue, ceLogging, taskProcessorRepository);
+    return new CeWorkerImpl(ordinal, uuid, queue, ceLogging, taskProcessorRepository, enabledCeWorkerController);
   }
 
   @Override
diff --git a/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorkerImpl.java b/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorkerImpl.java
index ba5e447ac7f..cbcabd72a84 100644
--- a/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorkerImpl.java
+++ b/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorkerImpl.java
@@ -33,6 +33,7 @@ import org.sonar.db.ce.CeActivityDto;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static java.lang.String.format;
+import static org.sonar.ce.taskprocessor.CeWorker.Result.DISABLED;
 import static org.sonar.ce.taskprocessor.CeWorker.Result.NO_TASK;
 import static org.sonar.ce.taskprocessor.CeWorker.Result.TASK_PROCESSED;
 
@@ -45,14 +46,17 @@ public class CeWorkerImpl implements CeWorker {
   private final InternalCeQueue queue;
   private final CeLogging ceLogging;
   private final CeTaskProcessorRepository taskProcessorRepository;
+  private final EnabledCeWorkerController enabledCeWorkerController;
 
   public CeWorkerImpl(int ordinal, String uuid,
-    InternalCeQueue queue, CeLogging ceLogging, CeTaskProcessorRepository taskProcessorRepository) {
+    InternalCeQueue queue, CeLogging ceLogging, CeTaskProcessorRepository taskProcessorRepository,
+    EnabledCeWorkerController enabledCeWorkerController) {
     this.ordinal = checkOrdinal(ordinal);
     this.uuid = uuid;
     this.queue = queue;
     this.ceLogging = ceLogging;
     this.taskProcessorRepository = taskProcessorRepository;
+    this.enabledCeWorkerController = enabledCeWorkerController;
   }
 
   private static int checkOrdinal(int ordinal) {
@@ -66,6 +70,9 @@ public class CeWorkerImpl implements CeWorker {
   }
 
   private Result findAndProcessTask() {
+    if (!enabledCeWorkerController.isEnabled(this)) {
+      return DISABLED;
+    }
     Optional<CeTask> ceTask = tryAndFindTaskToExecute();
     if (!ceTask.isPresent()) {
       return NO_TASK;
diff --git a/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/EnabledCeWorkerController.java b/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/EnabledCeWorkerController.java
new file mode 100644
index 00000000000..03e2a37f3bf
--- /dev/null
+++ b/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/EnabledCeWorkerController.java
@@ -0,0 +1,31 @@
+/*
+ * SonarQube
+ * Copyright (C) 2009-2017 SonarSource SA
+ * mailto:info AT sonarsource DOT com
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ */
+package org.sonar.ce.taskprocessor;
+
+/**
+ * This class is responsible of knowing/deciding which {@link CeWorker} is enabled and should actually try and find a
+ * task to process.
+ */
+public interface EnabledCeWorkerController {
+  /**
+   * Returns {@code true} if the specified {@link CeWorker} is enabled
+   */
+  boolean isEnabled(CeWorker ceWorker);
+}
diff --git a/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/EnabledCeWorkerControllerImpl.java b/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/EnabledCeWorkerControllerImpl.java
new file mode 100644
index 00000000000..19ae6998732
--- /dev/null
+++ b/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/EnabledCeWorkerControllerImpl.java
@@ -0,0 +1,42 @@
+/*
+ * SonarQube
+ * Copyright (C) 2009-2017 SonarSource SA
+ * mailto:info AT sonarsource DOT com
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ */
+package org.sonar.ce.taskprocessor;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import org.sonar.ce.configuration.CeConfiguration;
+
+public class EnabledCeWorkerControllerImpl implements EnabledCeWorkerController {
+  private final AtomicInteger workerCount;
+
+  public EnabledCeWorkerControllerImpl(CeConfiguration ceConfiguration) {
+    this.workerCount = new AtomicInteger(ceConfiguration.getWorkerCount());
+  }
+
+  /**
+   * Returns {@code true} if {@link CeWorker#getOrdinal() worker ordinal} is strictly less than
+   * {@link CeConfiguration#getWorkerCount()}.
+   *
+   * This method does not fail if ordinal is invalid (ie. < 0).
+   */
+  @Override
+  public boolean isEnabled(CeWorker ceWorker) {
+    return ceWorker.getOrdinal() < workerCount.get();
+  }
+}