SONAR-7104 save snapshotId in CeActivity if returned by CeTaskProcessor

5 files changed, 60 insertions, 13 deletions

diff --git a/server/sonar-server/src/main/java/org/sonar/server/computation/queue/CeQueue.java b/server/sonar-server/src/main/java/org/sonar/server/computation/queue/CeQueue.java
index 74adfcc6020..ce1ab600e8c 100644
--- a/server/sonar-server/src/main/java/org/sonar/server/computation/queue/CeQueue.java
+++ b/server/sonar-server/src/main/java/org/sonar/server/computation/queue/CeQueue.java
@@ -20,6 +20,7 @@
 package org.sonar.server.computation.queue;
 
 import com.google.common.base.Optional;
+import javax.annotation.Nullable;
 import org.sonar.db.ce.CeActivityDto;
 
 /**
@@ -85,11 +86,11 @@ public interface CeQueue {
 
   /**
    * Removes a task from the queue and registers it to past activities. This method
-   * is called by Compute Engine workers when task is processed.
+   * is called by Compute Engine workers when task is processed and can include an option {@link CeTaskResult} object.
    *
-   * <p>An unchecked exception is thrown if the task does not exist in the queue.</p>
+   * @throws IllegalStateException if the task does not exist in the queue
    */
-  void remove(CeTask task, CeActivityDto.Status status);
+  void remove(CeTask task, CeActivityDto.Status status, @Nullable CeTaskResult taskResult);
 
   void pauseSubmit();
 
diff --git a/server/sonar-server/src/main/java/org/sonar/server/computation/queue/CeQueueImpl.java b/server/sonar-server/src/main/java/org/sonar/server/computation/queue/CeQueueImpl.java
index 17b9cb214bc..eee80fa2bb2 100644
--- a/server/sonar-server/src/main/java/org/sonar/server/computation/queue/CeQueueImpl.java
+++ b/server/sonar-server/src/main/java/org/sonar/server/computation/queue/CeQueueImpl.java
@@ -21,6 +21,7 @@ package org.sonar.server.computation.queue;
 
 import com.google.common.base.Optional;
 import java.util.concurrent.atomic.AtomicBoolean;
+import javax.annotation.Nullable;
 import org.sonar.api.server.ServerSide;
 import org.sonar.api.utils.System2;
 import org.sonar.core.util.UuidFactory;
@@ -172,7 +173,7 @@ public class CeQueueImpl implements CeQueue {
   }
 
   @Override
-  public void remove(CeTask task, CeActivityDto.Status status) {
+  public void remove(CeTask task, CeActivityDto.Status status, CeTaskResult taskResult) {
     DbSession dbSession = dbClient.openSession(false);
     try {
       Optional<CeQueueDto> queueDto = dbClient.ceQueueDao().selectByUuid(dbSession, task.getUuid());
@@ -182,6 +183,7 @@ public class CeQueueImpl implements CeQueue {
       CeActivityDto activityDto = new CeActivityDto(queueDto.get());
       activityDto.setStatus(status);
       updateQueueStatus(status, activityDto);
+      updateTaskResult(activityDto, taskResult);
       remove(dbSession, task, queueDto.get(), activityDto);
 
     } finally {
@@ -189,6 +191,15 @@ public class CeQueueImpl implements CeQueue {
     }
   }
 
+  private void updateTaskResult(CeActivityDto activityDto, @Nullable CeTaskResult taskResult) {
+    if (taskResult != null) {
+      Long snapshotId = taskResult.getSnapshotId();
+      if (snapshotId != null) {
+        activityDto.setSnapshotId(snapshotId);
+      }
+    }
+  }
+
   private void updateQueueStatus(CeActivityDto.Status status, CeActivityDto activityDto) {
     Long startedAt = activityDto.getStartedAt();
     if (startedAt == null) {
diff --git a/server/sonar-server/src/main/java/org/sonar/server/computation/taskprocessor/CeWorkerCallableImpl.java b/server/sonar-server/src/main/java/org/sonar/server/computation/taskprocessor/CeWorkerCallableImpl.java
index 091bb7cdc48..b48f24bd4e0 100644
--- a/server/sonar-server/src/main/java/org/sonar/server/computation/taskprocessor/CeWorkerCallableImpl.java
+++ b/server/sonar-server/src/main/java/org/sonar/server/computation/taskprocessor/CeWorkerCallableImpl.java
@@ -28,6 +28,7 @@ import org.sonar.db.ce.CeActivityDto;
 import org.sonar.server.computation.log.CeLogging;
 import org.sonar.server.computation.queue.CeQueue;
 import org.sonar.server.computation.queue.CeTask;
+import org.sonar.server.computation.queue.CeTaskResult;
 
 import static java.lang.String.format;
 
@@ -72,21 +73,21 @@ public class CeWorkerCallableImpl implements CeWorkerCallable {
     Profiler ceProfiler = startProfiler(task);
 
     CeActivityDto.Status status = CeActivityDto.Status.FAILED;
+    CeTaskResult process = null;
     try {
       // TODO delegate the message to the related task processor, according to task type
       Optional<CeTaskProcessor> taskProcessor = taskProcessorRepository.getForCeTask(task);
       if (taskProcessor.isPresent()) {
-        taskProcessor.get().process(task);
+        process = taskProcessor.get().process(task);
         status = CeActivityDto.Status.SUCCESS;
       } else {
         LOG.error("No CeTaskProcessor is defined for task of type {}. Plugin configuration may have changed", task.getType());
         status = CeActivityDto.Status.FAILED;
       }
-      queue.remove(task, status);
     } catch (Throwable e) {
       LOG.error(format("Failed to execute task %s", task.getUuid()), e);
-      queue.remove(task, status);
     } finally {
+      queue.remove(task, status, process);
       // logging twice: once in sonar.log and once in CE appender
       stopProfiler(ceProfiler, task, status);
       ceLogging.clearForTask();
diff --git a/server/sonar-server/src/test/java/org/sonar/server/computation/queue/CeQueueImplTest.java b/server/sonar-server/src/test/java/org/sonar/server/computation/queue/CeQueueImplTest.java
index e291eae7539..35516f40622 100644
--- a/server/sonar-server/src/test/java/org/sonar/server/computation/queue/CeQueueImplTest.java
+++ b/server/sonar-server/src/test/java/org/sonar/server/computation/queue/CeQueueImplTest.java
@@ -43,6 +43,7 @@ import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
 
 public class CeQueueImplTest {
 
@@ -98,7 +99,7 @@ public class CeQueueImplTest {
   public void test_remove() {
     CeTask task = submit(CeTaskTypes.REPORT, "PROJECT_1");
     Optional<CeTask> peek = underTest.peek();
-    underTest.remove(peek.get(), CeActivityDto.Status.SUCCESS);
+    underTest.remove(peek.get(), CeActivityDto.Status.SUCCESS, null);
 
     // queue is empty
     assertThat(dbTester.getDbClient().ceQueueDao().selectByUuid(dbTester.getSession(), task.getUuid()).isPresent()).isFalse();
@@ -109,18 +110,45 @@ public class CeQueueImplTest {
     assertThat(history.isPresent()).isTrue();
     assertThat(history.get().getStatus()).isEqualTo(CeActivityDto.Status.SUCCESS);
     assertThat(history.get().getIsLast()).isTrue();
+    assertThat(history.get().getSnapshotId()).isNull();
 
     verify(listener).onRemoved(task, CeActivityDto.Status.SUCCESS);
   }
 
   @Test
+  public void remove_does_not_set_snapshotId_in_CeActivity_when_CeTaskResult_has_no_snapshot_id() {
+    CeTask task = submit(CeTaskTypes.REPORT, "PROJECT_1");
+    Optional<CeTask> peek = underTest.peek();
+    underTest.remove(peek.get(), CeActivityDto.Status.SUCCESS, newTaskResult(null));
+
+    // available in history
+    Optional<CeActivityDto> history = dbTester.getDbClient().ceActivityDao().selectByUuid(dbTester.getSession(), task.getUuid());
+    assertThat(history.isPresent()).isTrue();
+    assertThat(history.get().getSnapshotId()).isNull();
+  }
+
+  @Test
+  public void remove_sets_snapshotId_in_CeActivity_when_CeTaskResult_has_no_snapshot_id() {
+    CeTask task = submit(CeTaskTypes.REPORT, "PROJECT_1");
+    long snapshotId = 663L;
+
+    Optional<CeTask> peek = underTest.peek();
+    underTest.remove(peek.get(), CeActivityDto.Status.SUCCESS, newTaskResult(snapshotId));
+
+    // available in history
+    Optional<CeActivityDto> history = dbTester.getDbClient().ceActivityDao().selectByUuid(dbTester.getSession(), task.getUuid());
+    assertThat(history.isPresent()).isTrue();
+    assertThat(history.get().getSnapshotId()).isEqualTo(snapshotId);
+  }
+
+  @Test
   public void fail_to_remove_if_not_in_queue() throws Exception {
     expectedException.expect(IllegalStateException.class);
     CeTask task = submit(CeTaskTypes.REPORT, "PROJECT_1");
-    underTest.remove(task, CeActivityDto.Status.SUCCESS);
+    underTest.remove(task, CeActivityDto.Status.SUCCESS, null);
 
     // fail
-    underTest.remove(task, CeActivityDto.Status.SUCCESS);
+    underTest.remove(task, CeActivityDto.Status.SUCCESS, null);
   }
 
   @Test
@@ -222,4 +250,10 @@ public class CeQueueImplTest {
     submission.setComponentUuid(componentUuid);
     return underTest.submit(submission.build());
   }
+
+  private CeTaskResult newTaskResult(Long snapshotId) {
+    CeTaskResult taskResult = mock(CeTaskResult.class);
+    when(taskResult.getSnapshotId()).thenReturn(snapshotId);
+    return taskResult;
+  }
 }
diff --git a/server/sonar-server/src/test/java/org/sonar/server/computation/taskprocessor/CeWorkerCallableImplTest.java b/server/sonar-server/src/test/java/org/sonar/server/computation/taskprocessor/CeWorkerCallableImplTest.java
index 331ceaeac14..ced03cb5827 100644
--- a/server/sonar-server/src/test/java/org/sonar/server/computation/taskprocessor/CeWorkerCallableImplTest.java
+++ b/server/sonar-server/src/test/java/org/sonar/server/computation/taskprocessor/CeWorkerCallableImplTest.java
@@ -67,7 +67,7 @@ public class CeWorkerCallableImplTest {
     assertThat(underTest.call()).isTrue();
 
     inOrder.verify(ceLogging).initForTask(task);
-    inOrder.verify(queue).remove(task, CeActivityDto.Status.FAILED);
+    inOrder.verify(queue).remove(task, CeActivityDto.Status.FAILED, null);
     inOrder.verify(ceLogging).clearForTask();
   }
 
@@ -81,7 +81,7 @@ public class CeWorkerCallableImplTest {
 
     inOrder.verify(ceLogging).initForTask(task);
     inOrder.verify(taskProcessor).process(task);
-    inOrder.verify(queue).remove(task, CeActivityDto.Status.SUCCESS);
+    inOrder.verify(queue).remove(task, CeActivityDto.Status.SUCCESS, null);
     inOrder.verify(ceLogging).clearForTask();
   }
 
@@ -96,7 +96,7 @@ public class CeWorkerCallableImplTest {
 
     inOrder.verify(ceLogging).initForTask(task);
     inOrder.verify(taskProcessor).process(task);
-    inOrder.verify(queue).remove(task, CeActivityDto.Status.FAILED);
+    inOrder.verify(queue).remove(task, CeActivityDto.Status.FAILED, null);
     inOrder.verify(ceLogging).clearForTask();
   }
 }