SONAR-8985 add unique identifier to CeWorker

server/sonar-ce-api/src/main/java/org/sonar/ce/configuration/CeConfiguration.java

@@ -27,8 +27,8 @@ public interface CeConfiguration {
   int getWorkerCount();
 
   /**
-   * The delay in milliseconds before calling another {@link org.sonar.server.computation.taskprocessor.CeWorkerCallable}
-   * when previous one had nothing to do.
+   * The delay in millisecond before a {@link CeWorker} shall try and find a task
+   * to process when it's previous execution had nothing to do.
    */
   long getQueuePollingDelay();
 

server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeProcessingSchedulerExecutorService.java

@@ -23,7 +23,7 @@
 import org.sonar.server.util.StoppableExecutorService;
 
 /**
- * The {@link java.util.concurrent.ExecutorService} responsible for running {@link CeWorkerCallableImpl}.
+ * The {@link java.util.concurrent.ExecutorService} responsible for running {@link CeWorkerImpl}.
  */
 public interface CeProcessingSchedulerExecutorService extends StoppableExecutorService, ListeningScheduledExecutorService {
 }

server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeProcessingSchedulerImpl.java

@@ -39,24 +39,23 @@ public class CeProcessingSchedulerImpl implements CeProcessingScheduler, Startab
   private static final Logger LOG = Loggers.get(CeProcessingSchedulerImpl.class);
 
   private final CeProcessingSchedulerExecutorService executorService;
-  private final CeWorkerCallable workerRunnable;
 
   private final long delayBetweenTasks;
   private final TimeUnit timeUnit;
   private final ChainingCallback[] chainingCallbacks;
 
   public CeProcessingSchedulerImpl(CeConfiguration ceConfiguration,
-    CeProcessingSchedulerExecutorService processingExecutorService, CeWorkerCallable workerRunnable) {
+    CeProcessingSchedulerExecutorService processingExecutorService, CeWorkerFactory ceCeWorkerFactory) {
     this.executorService = processingExecutorService;
-    this.workerRunnable = workerRunnable;
 
     this.delayBetweenTasks = ceConfiguration.getQueuePollingDelay();
     this.timeUnit = MILLISECONDS;
 
     int workerCount = ceConfiguration.getWorkerCount();
     this.chainingCallbacks = new ChainingCallback[workerCount];
     for (int i = 0; i < workerCount; i++) {
-      chainingCallbacks[i] = new ChainingCallback();
+      CeWorker worker = ceCeWorkerFactory.create();
+      chainingCallbacks[i] = new ChainingCallback(worker);
     }
   }
 
@@ -68,7 +67,7 @@ public void start() {
   @Override
   public void startScheduling() {
     for (ChainingCallback chainingCallback : chainingCallbacks) {
-      ListenableScheduledFuture<Boolean> future = executorService.schedule(workerRunnable, delayBetweenTasks, timeUnit);
+      ListenableScheduledFuture<Boolean> future = executorService.schedule(chainingCallback.worker, delayBetweenTasks, timeUnit);
       addCallback(future, chainingCallback, executorService);
     }
   }
@@ -82,9 +81,15 @@ public void stop() {
 
   private class ChainingCallback implements FutureCallback<Boolean> {
     private final AtomicBoolean keepRunning = new AtomicBoolean(true);
+    private final CeWorker worker;
+
     @CheckForNull
     private ListenableFuture<Boolean> workerFuture;
 
+    public ChainingCallback(CeWorker worker) {
+      this.worker = worker;
+    }
+
     @Override
     public void onSuccess(@Nullable Boolean result) {
       if (result != null && result) {
@@ -105,14 +110,14 @@ public void onFailure(Throwable t) {
 
     private void chainWithoutDelay() {
       if (keepRunning()) {
-        workerFuture = executorService.submit(workerRunnable);
+        workerFuture = executorService.submit(worker);
       }
       addCallback();
     }
 
     private void chainWithDelay() {
       if (keepRunning()) {
-        workerFuture = executorService.schedule(workerRunnable, delayBetweenTasks, timeUnit);
+        workerFuture = executorService.schedule(worker, delayBetweenTasks, timeUnit);
       }
       addCallback();
     }

server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeTaskProcessorModule.java

@@ -26,7 +26,7 @@ public class CeTaskProcessorModule extends Module {
   protected void configureModule() {
     add(
       CeTaskProcessorRepositoryImpl.class,
-      CeWorkerCallableImpl.class,
+      CeWorkerFactoryImpl.class,
       CeProcessingSchedulerExecutorServiceImpl.class,
       CeProcessingSchedulerImpl.class);
   }

server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorkerCallable.java → server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorker.java

@@ -28,5 +28,6 @@
  * {@link Callable#call()} returns a Boolean which is {@code true} when some a {@link CeTask} was processed,
  * {@code false} otherwise.
  */
-public interface CeWorkerCallable extends Callable<Boolean> {
+public interface CeWorker extends Callable<Boolean> {
+  String getUUID();
 }

server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorkerFactory.java

@@ -0,0 +1,39 @@
+/*
+ * 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.Set;
+
+/**
+ * A factory that will create the CeWorkerFactory with an UUID
+ */
+public interface CeWorkerFactory {
+  /**
+   * Create a new CeWorker object.
+   * Each {@link CeWorker} returned by this method will have a different UUID from the others and all of these UUIDS will be returned by {@link #getWorkerUUIDs()}.
+   *
+   * @return the CeWorker
+   */
+  CeWorker create();
+  /**
+   * @return  the UUIDs of each {@link CeWorker} object returned by {@link #create}.
+   */
+  Set<String> getWorkerUUIDs();
+}

server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorkerFactoryImpl.java

@@ -0,0 +1,56 @@
+/*
+ * 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.HashSet;
+import java.util.Set;
+import org.sonar.ce.log.CeLogging;
+import org.sonar.ce.queue.InternalCeQueue;
+import org.sonar.core.util.UuidFactory;
+
+import static com.google.common.collect.ImmutableSet.copyOf;
+
+public class CeWorkerFactoryImpl implements CeWorkerFactory {
+  private final UuidFactory uuidFactory;
+  private final Set<String> ceWorkerUUIDs = new HashSet<>();
+  private final InternalCeQueue queue;
+  private final CeLogging ceLogging;
+  private final CeTaskProcessorRepository taskProcessorRepository;
+
+  public CeWorkerFactoryImpl(InternalCeQueue queue, CeLogging ceLogging, CeTaskProcessorRepository taskProcessorRepository, UuidFactory uuidFactory) {
+    this.queue = queue;
+    this.ceLogging = ceLogging;
+    this.taskProcessorRepository = taskProcessorRepository;
+    this.uuidFactory= uuidFactory;
+  }
+
+  @Override
+  public CeWorker create() {
+    String uuid = uuidFactory.create();
+    ceWorkerUUIDs.add(uuid);
+    return new CeWorkerImpl(queue, ceLogging, taskProcessorRepository, uuid);
+  }
+
+  @Override
+  public Set<String> getWorkerUUIDs() {
+    return copyOf(ceWorkerUUIDs);
+  }
+}

server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorkerCallableImpl.java → server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorkerImpl.java

@@ -20,7 +20,6 @@
 package org.sonar.ce.taskprocessor;
 
 import java.util.Optional;
-import java.util.concurrent.atomic.AtomicLong;
 import javax.annotation.Nullable;
 import org.sonar.api.utils.log.Logger;
 import org.sonar.api.utils.log.Loggers;
@@ -33,18 +32,20 @@
 
 import static java.lang.String.format;
 
-public class CeWorkerCallableImpl implements CeWorkerCallable {
+public class CeWorkerImpl implements CeWorker {
 
-  private static final Logger LOG = Loggers.get(CeWorkerCallableImpl.class);
+  private static final Logger LOG = Loggers.get(CeWorkerImpl.class);
 
   private final InternalCeQueue queue;
   private final CeLogging ceLogging;
   private final CeTaskProcessorRepository taskProcessorRepository;
+  private final String uuid;
 
-  public CeWorkerCallableImpl(InternalCeQueue queue, CeLogging ceLogging, CeTaskProcessorRepository taskProcessorRepository) {
+  public CeWorkerImpl(InternalCeQueue queue, CeLogging ceLogging, CeTaskProcessorRepository taskProcessorRepository, String uuid) {
     this.queue = queue;
     this.ceLogging = ceLogging;
     this.taskProcessorRepository = taskProcessorRepository;
+    this.uuid = uuid;
   }
 
   @Override
@@ -62,16 +63,21 @@ public Boolean call() throws Exception {
     return true;
   }
 
-  private static final AtomicLong counter = new AtomicLong(0);
+
   private Optional<CeTask> tryAndFindTaskToExecute() {
     try {
-      return queue.peek("uuid" + counter.addAndGet(100));
+      return queue.peek(uuid);
     } catch (Exception e) {
       LOG.error("Failed to pop the queue of analysis reports", e);
     }
     return Optional.empty();
   }
 
+  @Override
+  public String getUUID() {
+    return uuid;
+  }
+
   private void executeTask(CeTask task) {
     ceLogging.initForTask(task);
     Profiler ceProfiler = startActivityProfiler(task);
@@ -135,5 +141,4 @@ private static void addContext(Profiler profiler, CeTask task) {
       profiler.addContext("submitter", submitterLogin);
     }
   }
-
 }