SONAR-9525 add CeWorker#getOrdinal

8 files changed, 58 insertions, 19 deletions

diff --git a/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeProcessingSchedulerImpl.java b/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeProcessingSchedulerImpl.java
index c9873e75264..44f22f74d30 100644
--- a/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeProcessingSchedulerImpl.java
+++ b/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeProcessingSchedulerImpl.java
@@ -51,10 +51,10 @@ public class CeProcessingSchedulerImpl implements CeProcessingScheduler, Startab
     this.delayBetweenTasks = ceConfiguration.getQueuePollingDelay();
     this.timeUnit = MILLISECONDS;
 
-    int workerCount = ceConfiguration.getWorkerMaxCount();
-    this.chainingCallbacks = new ChainingCallback[workerCount];
-    for (int i = 0; i < workerCount; i++) {
-      CeWorker worker = ceCeWorkerFactory.create();
+    int threadWorkerCount = ceConfiguration.getWorkerMaxCount();
+    this.chainingCallbacks = new ChainingCallback[threadWorkerCount];
+    for (int i = 0; i < threadWorkerCount; i++) {
+      CeWorker worker = ceCeWorkerFactory.create(i);
       chainingCallbacks[i] = new ChainingCallback(worker);
     }
   }
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 7390b4a1d41..f0da47fc356 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
@@ -29,5 +29,13 @@ import org.sonar.ce.queue.CeTask;
  * {@code false} otherwise.
  */
 public interface CeWorker extends Callable<Boolean> {
+  /**
+   * Position of the current CeWorker among all the running workers.
+   */
+  int getOrdinal();
+
+  /**
+   * UUID of the current CeWorker.
+   */
   String getUUID();
 }
diff --git a/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorkerFactory.java b/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorkerFactory.java
index 65d9096da66..376123fef5a 100644
--- a/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorkerFactory.java
+++ b/server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorkerFactory.java
@@ -26,12 +26,13 @@ import java.util.Set;
  */
 public interface CeWorkerFactory {
   /**
-   * Create a new CeWorker object.
+   * Create a new CeWorker object with the specified ordinal.
    * 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();
+  CeWorker create(int ordinal);
+
   /**
    * @return  the UUIDs of each {@link CeWorker} object returned by {@link #create}.
    */
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 e684caa7fe7..21228bbc4cc 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
@@ -43,10 +43,10 @@ public class CeWorkerFactoryImpl implements CeWorkerFactory {
   }
 
   @Override
-  public CeWorker create() {
+  public CeWorker create(int ordinal) {
     String uuid = uuidFactory.create();
     ceWorkerUUIDs.add(uuid);
-    return new CeWorkerImpl(queue, ceLogging, taskProcessorRepository, uuid);
+    return new CeWorkerImpl(ordinal, uuid, queue, ceLogging, taskProcessorRepository);
   }
 
   @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 025f6366655..6052abb03d6 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
@@ -36,16 +36,19 @@ public class CeWorkerImpl implements CeWorker {
 
   private static final Logger LOG = Loggers.get(CeWorkerImpl.class);
 
+  private final int ordinal;
+  private final String uuid;
   private final InternalCeQueue queue;
   private final CeLogging ceLogging;
   private final CeTaskProcessorRepository taskProcessorRepository;
-  private final String uuid;
 
-  public CeWorkerImpl(InternalCeQueue queue, CeLogging ceLogging, CeTaskProcessorRepository taskProcessorRepository, String uuid) {
+  public CeWorkerImpl(int ordinal, String uuid,
+    InternalCeQueue queue, CeLogging ceLogging, CeTaskProcessorRepository taskProcessorRepository) {
+    this.ordinal = ordinal;
+    this.uuid = uuid;
     this.queue = queue;
     this.ceLogging = ceLogging;
     this.taskProcessorRepository = taskProcessorRepository;
-    this.uuid = uuid;
   }
 
   @Override
@@ -63,7 +66,6 @@ public class CeWorkerImpl implements CeWorker {
     return true;
   }
 
-
   private Optional<CeTask> tryAndFindTaskToExecute() {
     try {
       return queue.peek(uuid);
@@ -74,6 +76,11 @@ public class CeWorkerImpl implements CeWorker {
   }
 
   @Override
+  public int getOrdinal() {
+    return ordinal;
+  }
+
+  @Override
   public String getUUID() {
     return uuid;
   }
diff --git a/server/sonar-ce/src/test/java/org/sonar/ce/taskprocessor/CeProcessingSchedulerImplTest.java b/server/sonar-ce/src/test/java/org/sonar/ce/taskprocessor/CeProcessingSchedulerImplTest.java
index 4d9e8a037ea..dc0b2598cab 100644
--- a/server/sonar-ce/src/test/java/org/sonar/ce/taskprocessor/CeProcessingSchedulerImplTest.java
+++ b/server/sonar-ce/src/test/java/org/sonar/ce/taskprocessor/CeProcessingSchedulerImplTest.java
@@ -219,7 +219,9 @@ public class CeProcessingSchedulerImplTest {
       verify(processingExecutorService).schedule(workers[i], ceConfiguration.getQueuePollingDelay(), MILLISECONDS);
     }
     verify(listenableScheduledFuture, times(workerCount)).addListener(any(Runnable.class), eq(processingExecutorService));
-    verify(ceWorkerFactory, times(workerCount)).create();
+    for (int i = 0; i < workerCount; i++) {
+      verify(ceWorkerFactory).create(i);
+    }
   }
 
   private void startSchedulingAndRun() throws ExecutionException, InterruptedException {
@@ -237,7 +239,7 @@ public class CeProcessingSchedulerImplTest {
     }
 
     @Override
-    public CeWorker create() {
+    public CeWorker create(int ordinal) {
       // This will throw an NoSuchElementException if there are too many calls
       return ceWorkers.next();
     }
diff --git a/server/sonar-ce/src/test/java/org/sonar/ce/taskprocessor/CeWorkerFactoryImplTest.java b/server/sonar-ce/src/test/java/org/sonar/ce/taskprocessor/CeWorkerFactoryImplTest.java
index 1f3b4aaea09..fbf868392e6 100644
--- a/server/sonar-ce/src/test/java/org/sonar/ce/taskprocessor/CeWorkerFactoryImplTest.java
+++ b/server/sonar-ce/src/test/java/org/sonar/ce/taskprocessor/CeWorkerFactoryImplTest.java
@@ -21,7 +21,9 @@
 package org.sonar.ce.taskprocessor;
 
 import java.util.HashSet;
+import java.util.Random;
 import java.util.Set;
+import java.util.stream.IntStream;
 import org.junit.Test;
 import org.sonar.ce.log.CeLogging;
 import org.sonar.ce.queue.InternalCeQueue;
@@ -31,16 +33,33 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
 
 public class CeWorkerFactoryImplTest {
+  private int randomOrdinal = new Random().nextInt();
   private CeWorkerFactoryImpl underTest = new CeWorkerFactoryImpl(mock(InternalCeQueue.class), mock(CeLogging.class),
     mock(CeTaskProcessorRepository.class), UuidFactoryImpl.INSTANCE);
 
   @Test
+  public void create_return_CeWorker_object_with_specified_ordinal() {
+    CeWorker ceWorker = underTest.create(randomOrdinal);
+
+    assertThat(ceWorker.getOrdinal()).isEqualTo(randomOrdinal);
+  }
+
+  @Test
+  public void create_allows_multiple_calls_with_same_ordinal() {
+    IntStream.range(0, new Random().nextInt(50)).forEach(ignored -> {
+      CeWorker ceWorker = underTest.create(randomOrdinal);
+
+      assertThat(ceWorker.getOrdinal()).isEqualTo(randomOrdinal);
+    });
+  }
+
+  @Test
   public void each_call_must_return_a_new_ceworker_with_unique_uuid() {
     Set<CeWorker> ceWorkers = new HashSet<>();
     Set<String> ceWorkerUUIDs = new HashSet<>();
 
     for (int i = 0; i < 10; i++) {
-      CeWorker ceWorker = underTest.create();
+      CeWorker ceWorker = underTest.create(i);
       ceWorkers.add(ceWorker);
       ceWorkerUUIDs.add(ceWorker.getUUID());
     }
@@ -51,7 +70,7 @@ public class CeWorkerFactoryImplTest {
 
   @Test
   public void ceworker_created_by_factory_must_contain_uuid() {
-    CeWorker ceWorker = underTest.create();
+    CeWorker ceWorker = underTest.create(randomOrdinal);
     assertThat(ceWorker.getUUID()).isNotEmpty();
   }
 
@@ -65,7 +84,7 @@ public class CeWorkerFactoryImplTest {
     Set<String> ceWorkerUUIDs = new HashSet<>();
 
     for (int i = 0; i < 10; i++) {
-      ceWorkerUUIDs.add(underTest.create().getUUID());
+      ceWorkerUUIDs.add(underTest.create(i).getUUID());
     }
 
     assertThat(underTest.getWorkerUUIDs()).isEqualTo(ceWorkerUUIDs);
diff --git a/server/sonar-ce/src/test/java/org/sonar/ce/taskprocessor/CeWorkerImplTest.java b/server/sonar-ce/src/test/java/org/sonar/ce/taskprocessor/CeWorkerImplTest.java
index 155507bb171..041418c5cf8 100644
--- a/server/sonar-ce/src/test/java/org/sonar/ce/taskprocessor/CeWorkerImplTest.java
+++ b/server/sonar-ce/src/test/java/org/sonar/ce/taskprocessor/CeWorkerImplTest.java
@@ -21,6 +21,7 @@ package org.sonar.ce.taskprocessor;
 
 import java.util.List;
 import java.util.Optional;
+import java.util.Random;
 import java.util.UUID;
 import javax.annotation.Nullable;
 import org.junit.Rule;
@@ -57,13 +58,14 @@ public class CeWorkerImplTest {
   private ReportTaskProcessor taskProcessor = mock(ReportTaskProcessor.class);
   private CeLogging ceLogging = spy(CeLogging.class);
   private ArgumentCaptor<String> workerUuid = ArgumentCaptor.forClass(String.class);
-  private CeWorker underTest = new CeWorkerImpl(queue, ceLogging, taskProcessorRepository, UUID.randomUUID().toString());
+  private int randomOrdinal = new Random().nextInt();
+  private CeWorker underTest = new CeWorkerImpl(randomOrdinal, UUID.randomUUID().toString(), queue, ceLogging, taskProcessorRepository);
   private InOrder inOrder = Mockito.inOrder(ceLogging, taskProcessor, queue);
 
   @Test
   public void getUUID_must_return_the_uuid_of_constructor() {
     String uuid = UUID.randomUUID().toString();
-    CeWorker underTest = new CeWorkerImpl(queue, ceLogging, taskProcessorRepository, uuid);
+    CeWorker underTest = new CeWorkerImpl(randomOrdinal, uuid, queue, ceLogging, taskProcessorRepository);
     assertThat(underTest.getUUID()).isEqualTo(uuid);
   }