aboutsummaryrefslogtreecommitdiffstats
path: root/server/sonar-ce
diff options
context:
space:
mode:
authorPierre <pierre.guillot@sonarsource.com>2020-06-11 14:53:08 +0200
committersonartech <sonartech@sonarsource.com>2020-06-26 20:04:57 +0000
commite85644a5ea11ff8cbd3f5593d917c68d5d3ee7d0 (patch)
tree5b9a1f0a08e7669bada357336fb6801d64d633c4 /server/sonar-ce
parent0a9b880143936ba89f9830d454f1dc3ef8d772b3 (diff)
downloadsonarqube-e85644a5ea11ff8cbd3f5593d917c68d5d3ee7d0.tar.gz
sonarqube-e85644a5ea11ff8cbd3f5593d917c68d5d3ee7d0.zip
SONAR-13400 exclude portfolio (but not applications) from worker peek during indexation
Diffstat (limited to 'server/sonar-ce')
-rw-r--r--server/sonar-ce/src/main/java/org/sonar/ce/queue/InternalCeQueue.java3
-rw-r--r--server/sonar-ce/src/main/java/org/sonar/ce/queue/InternalCeQueueImpl.java4
-rw-r--r--server/sonar-ce/src/main/java/org/sonar/ce/taskprocessor/CeWorkerImpl.java6
-rw-r--r--server/sonar-ce/src/test/java/org/sonar/ce/queue/InternalCeQueueImplTest.java40
-rw-r--r--server/sonar-ce/src/test/java/org/sonar/ce/taskprocessor/CeWorkerImplTest.java75
5 files changed, 72 insertions, 56 deletions
diff --git a/server/sonar-ce/src/main/java/org/sonar/ce/queue/InternalCeQueue.java b/server/sonar-ce/src/main/java/org/sonar/ce/queue/InternalCeQueue.java
index 30466805efc..be8f52d7b08 100644
--- a/server/sonar-ce/src/main/java/org/sonar/ce/queue/InternalCeQueue.java
+++ b/server/sonar-ce/src/main/java/org/sonar/ce/queue/InternalCeQueue.java
@@ -41,6 +41,7 @@ public interface InternalCeQueue extends CeQueue {
* Does not return anything if workers are paused or being paused (see {@link #getWorkersPauseStatus()}.
*
* @param excludeIndexationJob change the underlying request to exclude indexation tasks.
+ * @param excludeViewRefresh change the underlying request to exclude portfolios (but still include APP)
*
* <p>Only a single task can be peeked by project.</p>
*
@@ -49,7 +50,7 @@ public interface InternalCeQueue extends CeQueue {
* <p>Tasks which have been executed twice already but are still {@link org.sonar.db.ce.CeQueueDto.Status#PENDING}
* are ignored</p>
*/
- Optional<CeTask> peek(String workerUuid, boolean excludeIndexationJob);
+ Optional<CeTask> peek(String workerUuid, boolean excludeIndexationJob, boolean excludeViewRefresh);
/**
* Removes a task from the queue and registers it to past activities. This method
diff --git a/server/sonar-ce/src/main/java/org/sonar/ce/queue/InternalCeQueueImpl.java b/server/sonar-ce/src/main/java/org/sonar/ce/queue/InternalCeQueueImpl.java
index 711961b1f95..279b2f60021 100644
--- a/server/sonar-ce/src/main/java/org/sonar/ce/queue/InternalCeQueueImpl.java
+++ b/server/sonar-ce/src/main/java/org/sonar/ce/queue/InternalCeQueueImpl.java
@@ -73,7 +73,7 @@ public class InternalCeQueueImpl extends CeQueueImpl implements InternalCeQueue
}
@Override
- public Optional<CeTask> peek(String workerUuid, boolean excludeIndexationJob) {
+ public Optional<CeTask> peek(String workerUuid, boolean excludeIndexationJob, boolean excludeViewRefresh) {
requireNonNull(workerUuid, "workerUuid can't be null");
if (computeEngineStatus.getStatus() != ComputeEngineStatus.Status.STARTED || getWorkersPauseStatus() != WorkersPauseStatus.RESUMED) {
@@ -86,7 +86,7 @@ public class InternalCeQueueImpl extends CeQueueImpl implements InternalCeQueue
dbSession.commit();
LOG.debug("{} in progress tasks reset for worker uuid {}", i, workerUuid);
}
- Optional<CeQueueDto> opt = ceQueueDao.peek(dbSession, workerUuid, excludeIndexationJob);
+ Optional<CeQueueDto> opt = ceQueueDao.peek(dbSession, workerUuid, excludeIndexationJob, excludeViewRefresh);
if (!opt.isPresent()) {
return Optional.empty();
}
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 10c234639f5..c100f2e744a 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
@@ -171,7 +171,7 @@ public class CeWorkerImpl implements CeWorker {
if (indexationTaskLookupEnabled) {
return tryAndFindTaskToExecuteIncludingIndexation();
} else {
- return queue.peek(uuid, true);
+ return queue.peek(uuid, true, false);
}
} catch (Exception e) {
LOG.error("Failed to pop the queue of analysis reports", e);
@@ -181,12 +181,12 @@ public class CeWorkerImpl implements CeWorker {
private Optional<CeTask> tryAndFindTaskToExecuteIncludingIndexation() {
excludeIndexationJob = !excludeIndexationJob;
- Optional<CeTask> peek = queue.peek(uuid, excludeIndexationJob);
+ Optional<CeTask> peek = queue.peek(uuid, excludeIndexationJob, true);
if (peek.isPresent()) {
return peek;
}
if (excludeIndexationJob) {
- peek = queue.peek(uuid, false);
+ peek = queue.peek(uuid, false, true);
if (peek.isPresent()) {
return peek;
}
diff --git a/server/sonar-ce/src/test/java/org/sonar/ce/queue/InternalCeQueueImplTest.java b/server/sonar-ce/src/test/java/org/sonar/ce/queue/InternalCeQueueImplTest.java
index 922da41cdc3..dbfcd54a60b 100644
--- a/server/sonar-ce/src/test/java/org/sonar/ce/queue/InternalCeQueueImplTest.java
+++ b/server/sonar-ce/src/test/java/org/sonar/ce/queue/InternalCeQueueImplTest.java
@@ -159,18 +159,18 @@ public class InternalCeQueueImplTest {
expectedException.expect(NullPointerException.class);
expectedException.expectMessage("workerUuid can't be null");
- underTest.peek(null, false);
+ underTest.peek(null, false, false);
}
@Test
public void test_remove() {
CeTask task = submit(CeTaskTypes.REPORT, newProjectDto("PROJECT_1"));
- Optional<CeTask> peek = underTest.peek(WORKER_UUID_1, false);
+ Optional<CeTask> peek = underTest.peek(WORKER_UUID_1, false, false);
underTest.remove(peek.get(), CeActivityDto.Status.SUCCESS, null, null);
// queue is empty
assertThat(db.getDbClient().ceQueueDao().selectByUuid(db.getSession(), task.getUuid()).isPresent()).isFalse();
- assertThat(underTest.peek(WORKER_UUID_2, false).isPresent()).isFalse();
+ assertThat(underTest.peek(WORKER_UUID_2, false, false).isPresent()).isFalse();
// available in history
Optional<CeActivityDto> history = db.getDbClient().ceActivityDao().selectByUuid(db.getSession(), task.getUuid());
@@ -199,7 +199,7 @@ public class InternalCeQueueImplTest {
@Test
public void remove_does_not_set_analysisUuid_in_CeActivity_when_CeTaskResult_has_no_analysis_uuid() {
CeTask task = submit(CeTaskTypes.REPORT, newProjectDto("PROJECT_1"));
- Optional<CeTask> peek = underTest.peek(WORKER_UUID_1, false);
+ Optional<CeTask> peek = underTest.peek(WORKER_UUID_1, false, false);
underTest.remove(peek.get(), CeActivityDto.Status.SUCCESS, newTaskResult(null), null);
// available in history
@@ -212,7 +212,7 @@ public class InternalCeQueueImplTest {
public void remove_sets_analysisUuid_in_CeActivity_when_CeTaskResult_has_analysis_uuid() {
CeTask task = submit(CeTaskTypes.REPORT, newProjectDto("PROJECT_1"));
- Optional<CeTask> peek = underTest.peek(WORKER_UUID_2, false);
+ Optional<CeTask> peek = underTest.peek(WORKER_UUID_2, false, false);
underTest.remove(peek.get(), CeActivityDto.Status.SUCCESS, newTaskResult(AN_ANALYSIS_UUID), null);
// available in history
@@ -226,7 +226,7 @@ public class InternalCeQueueImplTest {
Throwable error = new NullPointerException("Fake NPE to test persistence to DB");
CeTask task = submit(CeTaskTypes.REPORT, newProjectDto("PROJECT_1"));
- Optional<CeTask> peek = underTest.peek(WORKER_UUID_1, false);
+ Optional<CeTask> peek = underTest.peek(WORKER_UUID_1, false, false);
underTest.remove(peek.get(), CeActivityDto.Status.FAILED, null, error);
Optional<CeActivityDto> activityDto = db.getDbClient().ceActivityDao().selectByUuid(session, task.getUuid());
@@ -242,7 +242,7 @@ public class InternalCeQueueImplTest {
Throwable error = new TypedExceptionImpl("aType", "aMessage");
CeTask task = submit(CeTaskTypes.REPORT, newProjectDto("PROJECT_1"));
- Optional<CeTask> peek = underTest.peek(WORKER_UUID_1, false);
+ Optional<CeTask> peek = underTest.peek(WORKER_UUID_1, false, false);
underTest.remove(peek.get(), CeActivityDto.Status.FAILED, null, error);
CeActivityDto activityDto = db.getDbClient().ceActivityDao().selectByUuid(session, task.getUuid()).get();
@@ -348,7 +348,7 @@ public class InternalCeQueueImplTest {
public void test_peek() {
CeTask task = submit(CeTaskTypes.REPORT, newProjectDto("PROJECT_1"));
- Optional<CeTask> peek = underTest.peek(WORKER_UUID_1, false);
+ Optional<CeTask> peek = underTest.peek(WORKER_UUID_1, false, false);
assertThat(peek.isPresent()).isTrue();
assertThat(peek.get().getUuid()).isEqualTo(task.getUuid());
assertThat(peek.get().getType()).isEqualTo(CeTaskTypes.REPORT);
@@ -356,7 +356,7 @@ public class InternalCeQueueImplTest {
assertThat(peek.get().getMainComponent()).contains(peek.get().getComponent().get());
// no more pending tasks
- peek = underTest.peek(WORKER_UUID_2, false);
+ peek = underTest.peek(WORKER_UUID_2, false, false);
assertThat(peek.isPresent()).isFalse();
}
@@ -366,7 +366,7 @@ public class InternalCeQueueImplTest {
ComponentDto branch = db.components().insertProjectBranch(project);
CeTask task = submit(CeTaskTypes.REPORT, branch);
- Optional<CeTask> peek = underTest.peek(WORKER_UUID_1, false);
+ Optional<CeTask> peek = underTest.peek(WORKER_UUID_1, false, false);
assertThat(peek.isPresent()).isTrue();
assertThat(peek.get().getUuid()).isEqualTo(task.getUuid());
assertThat(peek.get().getType()).isEqualTo(CeTaskTypes.REPORT);
@@ -374,7 +374,7 @@ public class InternalCeQueueImplTest {
assertThat(peek.get().getMainComponent()).contains(new CeTask.Component(project.uuid(), project.getDbKey(), project.name()));
// no more pending tasks
- peek = underTest.peek(WORKER_UUID_2, false);
+ peek = underTest.peek(WORKER_UUID_2, false, false);
assertThat(peek.isPresent()).isFalse();
}
@@ -383,11 +383,11 @@ public class InternalCeQueueImplTest {
CeTask task = submit(CeTaskTypes.REPORT, newProjectDto("PROJECT_1"));
underTest.pauseWorkers();
- Optional<CeTask> peek = underTest.peek(WORKER_UUID_1, false);
+ Optional<CeTask> peek = underTest.peek(WORKER_UUID_1, false, false);
assertThat(peek).isEmpty();
underTest.resumeWorkers();
- peek = underTest.peek(WORKER_UUID_1, false);
+ peek = underTest.peek(WORKER_UUID_1, false, false);
assertThat(peek).isPresent();
assertThat(peek.get().getUuid()).isEqualTo(task.getUuid());
}
@@ -401,7 +401,7 @@ public class InternalCeQueueImplTest {
makeInProgress(dto, "foo");
db.commit();
- assertThat(underTest.peek(WORKER_UUID_1, false)).isEmpty();
+ assertThat(underTest.peek(WORKER_UUID_1, false, false)).isEmpty();
}
@Test
@@ -409,7 +409,7 @@ public class InternalCeQueueImplTest {
submit(CeTaskTypes.REPORT, newProjectDto("PROJECT_1"));
when(computeEngineStatus.getStatus()).thenReturn(STOPPING);
- Optional<CeTask> peek = underTest.peek(WORKER_UUID_1, false);
+ Optional<CeTask> peek = underTest.peek(WORKER_UUID_1, false, false);
assertThat(peek.isPresent()).isFalse();
}
@@ -421,7 +421,7 @@ public class InternalCeQueueImplTest {
.setStatus(CeQueueDto.Status.PENDING));
db.commit();
- assertThat(underTest.peek(WORKER_UUID_1, false).get().getUuid()).isEqualTo("uuid");
+ assertThat(underTest.peek(WORKER_UUID_1, false, false).get().getUuid()).isEqualTo("uuid");
}
@Test
@@ -430,7 +430,7 @@ public class InternalCeQueueImplTest {
CeQueueDto u1 = insertPending("u1");// will be picked-because older than any of the reset ones
CeQueueDto u2 = insertInProgress("u2", WORKER_UUID_1);// will be reset
- assertThat(underTest.peek(WORKER_UUID_1, false).get().getUuid()).isEqualTo("u0");
+ assertThat(underTest.peek(WORKER_UUID_1, false, false).get().getUuid()).isEqualTo("u0");
verifyUnmodifiedTask(u1);
verifyResetTask(u2);
@@ -444,7 +444,7 @@ public class InternalCeQueueImplTest {
CeQueueDto u3 = insertInProgress("u3", WORKER_UUID_1);
CeQueueDto u4 = insertInProgress("u4", WORKER_UUID_2);
- assertThat(underTest.peek(WORKER_UUID_1, false).get().getUuid()).isEqualTo("u0");
+ assertThat(underTest.peek(WORKER_UUID_1, false, false).get().getUuid()).isEqualTo("u0");
verifyResetTask(u1);
verifyUnmodifiedTask(u2);
@@ -502,7 +502,7 @@ public class InternalCeQueueImplTest {
@Test
public void fail_to_cancel_if_in_progress() {
CeTask task = submit(CeTaskTypes.REPORT, newProjectDto("PROJECT_1"));
- underTest.peek(WORKER_UUID_2, false);
+ underTest.peek(WORKER_UUID_2, false, false);
CeQueueDto queueDto = db.getDbClient().ceQueueDao().selectByUuid(db.getSession(), task.getUuid()).get();
expectedException.expect(IllegalStateException.class);
@@ -516,7 +516,7 @@ public class InternalCeQueueImplTest {
CeTask inProgressTask = submit(CeTaskTypes.REPORT, newProjectDto("PROJECT_1"));
CeTask pendingTask1 = submit(CeTaskTypes.REPORT, newProjectDto("PROJECT_2"));
CeTask pendingTask2 = submit(CeTaskTypes.REPORT, newProjectDto("PROJECT_3"));
- underTest.peek(WORKER_UUID_2, false);
+ underTest.peek(WORKER_UUID_2, false, false);
int canceledCount = underTest.cancelAll();
assertThat(canceledCount).isEqualTo(2);
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 84943ca6009..ddb5027720d 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
@@ -19,6 +19,7 @@
*/
package org.sonar.ce.taskprocessor;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -145,7 +146,7 @@ public class CeWorkerImplTest {
@Test
public void no_pending_tasks_in_queue() throws Exception {
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.empty());
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.empty());
assertThat(underTest.call()).isEqualTo(NO_TASK);
@@ -154,7 +155,7 @@ public class CeWorkerImplTest {
@Test
public void no_pending_tasks_in_queue_without_listener() throws Exception {
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.empty());
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.empty());
assertThat(underTestNoListener.call()).isEqualTo(NO_TASK);
@@ -165,7 +166,7 @@ public class CeWorkerImplTest {
public void fail_when_no_CeTaskProcessor_is_found_in_repository() throws Exception {
CeTask task = createCeTask(null);
taskProcessorRepository.setNoProcessorForTask(CeTaskTypes.REPORT);
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(task));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(task));
assertThat(underTest.call()).isEqualTo(TASK_PROCESSED);
@@ -181,7 +182,7 @@ public class CeWorkerImplTest {
public void fail_when_no_CeTaskProcessor_is_found_in_repository_without_listener() throws Exception {
CeTask task = createCeTask(null);
taskProcessorRepository.setNoProcessorForTask(CeTaskTypes.REPORT);
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(task));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(task));
assertThat(underTestNoListener.call()).isEqualTo(TASK_PROCESSED);
@@ -194,7 +195,7 @@ public class CeWorkerImplTest {
public void peek_and_process_task() throws Exception {
CeTask task = createCeTask(null);
taskProcessorRepository.setProcessorForTask(task.getType(), taskProcessor);
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(task));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(task));
assertThat(underTest.call()).isEqualTo(TASK_PROCESSED);
@@ -211,7 +212,7 @@ public class CeWorkerImplTest {
public void peek_and_process_task_without_listeners() throws Exception {
CeTask task = createCeTask(null);
taskProcessorRepository.setProcessorForTask(task.getType(), taskProcessor);
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(task));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(task));
assertThat(underTestNoListener.call()).isEqualTo(TASK_PROCESSED);
@@ -224,7 +225,7 @@ public class CeWorkerImplTest {
@Test
public void fail_to_process_task() throws Exception {
CeTask task = createCeTask(null);
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(task));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(task));
taskProcessorRepository.setProcessorForTask(task.getType(), taskProcessor);
Throwable error = makeTaskProcessorFail(task);
@@ -242,7 +243,7 @@ public class CeWorkerImplTest {
@Test
public void fail_to_process_task_without_listeners() throws Exception {
CeTask task = createCeTask(null);
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(task));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(task));
taskProcessorRepository.setProcessorForTask(task.getType(), taskProcessor);
Throwable error = makeTaskProcessorFail(task);
@@ -256,7 +257,7 @@ public class CeWorkerImplTest {
@Test
public void log_task_characteristics() throws Exception {
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(createCeTask(null, "pullRequest", "123", "branch", "foo")));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(createCeTask(null, "pullRequest", "123", "branch", "foo")));
taskProcessorRepository.setProcessorForTask(CeTaskTypes.REPORT, taskProcessor);
underTest.call();
@@ -271,7 +272,7 @@ public class CeWorkerImplTest {
@Test
public void do_not_log_submitter_param_if_anonymous_and_success() throws Exception {
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(createCeTask(null)));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(createCeTask(null)));
taskProcessorRepository.setProcessorForTask(CeTaskTypes.REPORT, taskProcessor);
underTest.call();
@@ -287,7 +288,7 @@ public class CeWorkerImplTest {
@Test
public void do_not_log_submitter_param_if_anonymous_and_error() throws Exception {
CeTask ceTask = createCeTask(null);
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(ceTask));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(ceTask));
taskProcessorRepository.setProcessorForTask(ceTask.getType(), taskProcessor);
makeTaskProcessorFail(ceTask);
@@ -307,7 +308,7 @@ public class CeWorkerImplTest {
@Test
public void log_submitter_login_if_authenticated_and_success() throws Exception {
UserDto userDto = insertRandomUser();
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(createCeTask(toTaskSubmitter(userDto))));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(createCeTask(toTaskSubmitter(userDto))));
taskProcessorRepository.setProcessorForTask(CeTaskTypes.REPORT, taskProcessor);
underTest.call();
@@ -323,7 +324,7 @@ public class CeWorkerImplTest {
@Test
public void log_submitterUuid_if_user_matching_submitterUuid_can_not_be_found() throws Exception {
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(createCeTask(new CeTask.User("UUID_USER", null))));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(createCeTask(new CeTask.User("UUID_USER", null))));
taskProcessorRepository.setProcessorForTask(CeTaskTypes.REPORT, taskProcessor);
underTest.call();
@@ -341,7 +342,7 @@ public class CeWorkerImplTest {
public void display_submitterLogin_in_logs_when_set_in_case_of_error() throws Exception {
UserDto userDto = insertRandomUser();
CeTask ceTask = createCeTask(toTaskSubmitter(userDto));
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(ceTask));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(ceTask));
taskProcessorRepository.setProcessorForTask(ceTask.getType(), taskProcessor);
makeTaskProcessorFail(ceTask);
@@ -361,7 +362,7 @@ public class CeWorkerImplTest {
public void display_start_stop_at_debug_level_for_console_if_DEBUG_is_enabled_and_task_successful() throws Exception {
logTester.setLevel(LoggerLevel.DEBUG);
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(createCeTask(submitter)));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(createCeTask(submitter)));
taskProcessorRepository.setProcessorForTask(CeTaskTypes.REPORT, taskProcessor);
underTest.call();
@@ -380,7 +381,7 @@ public class CeWorkerImplTest {
logTester.setLevel(LoggerLevel.DEBUG);
CeTask ceTask = createCeTask(submitter);
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(ceTask));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(ceTask));
taskProcessorRepository.setProcessorForTask(CeTaskTypes.REPORT, taskProcessor);
makeTaskProcessorFail(ceTask);
@@ -400,7 +401,7 @@ public class CeWorkerImplTest {
@Test
public void call_sets_and_restores_thread_name_with_information_of_worker_when_there_is_no_task_to_process() throws Exception {
String threadName = randomAlphabetic(3);
- when(queue.peek(anyString(), anyBoolean())).thenAnswer(invocation -> {
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenAnswer(invocation -> {
assertThat(Thread.currentThread().getName())
.isEqualTo("Worker " + randomOrdinal + " (UUID=" + workerUuid + ") on " + threadName);
return Optional.empty();
@@ -414,7 +415,7 @@ public class CeWorkerImplTest {
@Test
public void call_sets_and_restores_thread_name_with_information_of_worker_when_a_task_is_processed() throws Exception {
String threadName = randomAlphabetic(3);
- when(queue.peek(anyString(), anyBoolean())).thenAnswer(invocation -> {
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenAnswer(invocation -> {
assertThat(Thread.currentThread().getName())
.isEqualTo("Worker " + randomOrdinal + " (UUID=" + workerUuid + ") on " + threadName);
return Optional.of(createCeTask(submitter));
@@ -430,7 +431,7 @@ public class CeWorkerImplTest {
public void call_sets_and_restores_thread_name_with_information_of_worker_when_an_error_occurs() throws Exception {
String threadName = randomAlphabetic(3);
CeTask ceTask = createCeTask(submitter);
- when(queue.peek(anyString(), anyBoolean())).thenAnswer(invocation -> {
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenAnswer(invocation -> {
assertThat(Thread.currentThread().getName())
.isEqualTo("Worker " + randomOrdinal + " (UUID=" + workerUuid + ") on " + threadName);
return Optional.of(ceTask);
@@ -458,7 +459,7 @@ public class CeWorkerImplTest {
@Test
public void log_error_when_task_fails_with_not_MessageException() throws Exception {
CeTask ceTask = createCeTask(submitter);
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(ceTask));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(ceTask));
taskProcessorRepository.setProcessorForTask(CeTaskTypes.REPORT, taskProcessor);
makeTaskProcessorFail(ceTask);
@@ -476,7 +477,7 @@ public class CeWorkerImplTest {
@Test
public void do_no_log_error_when_task_fails_with_MessageException() throws Exception {
CeTask ceTask = createCeTask(submitter);
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(ceTask));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(ceTask));
taskProcessorRepository.setProcessorForTask(CeTaskTypes.REPORT, taskProcessor);
makeTaskProcessorFail(ceTask, MessageException.of("simulate MessageException thrown by TaskProcessor#process"));
@@ -492,7 +493,7 @@ public class CeWorkerImplTest {
@Test
public void do_no_log_error_when_task_fails_with_BillingValidationsException() throws Exception {
CeTask ceTask = createCeTask(submitter);
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(ceTask));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(ceTask));
taskProcessorRepository.setProcessorForTask(CeTaskTypes.REPORT, taskProcessor);
makeTaskProcessorFail(ceTask, new BillingValidations.BillingValidationsException("simulate MessageException thrown by TaskProcessor#process"));
@@ -508,7 +509,7 @@ public class CeWorkerImplTest {
@Test
public void log_error_when_task_was_successful_but_ending_state_can_not_be_persisted_to_db() throws Exception {
CeTask ceTask = createCeTask(submitter);
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(ceTask));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(ceTask));
taskProcessorRepository.setProcessorForTask(CeTaskTypes.REPORT, taskProcessor);
doThrow(new RuntimeException("Simulate queue#remove failing")).when(queue).remove(ceTask, CeActivityDto.Status.SUCCESS, null, null);
@@ -520,7 +521,7 @@ public class CeWorkerImplTest {
@Test
public void log_error_when_task_failed_and_ending_state_can_not_be_persisted_to_db() throws Exception {
CeTask ceTask = createCeTask(submitter);
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(ceTask));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(ceTask));
taskProcessorRepository.setProcessorForTask(CeTaskTypes.REPORT, taskProcessor);
IllegalStateException ex = makeTaskProcessorFail(ceTask);
RuntimeException runtimeException = new RuntimeException("Simulate queue#remove failing");
@@ -549,7 +550,7 @@ public class CeWorkerImplTest {
@Test
public void log_error_as_suppressed_when_task_failed_with_MessageException_and_ending_state_can_not_be_persisted_to_db() throws Exception {
CeTask ceTask = createCeTask(submitter);
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(ceTask));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(ceTask));
taskProcessorRepository.setProcessorForTask(CeTaskTypes.REPORT, taskProcessor);
MessageException ex = makeTaskProcessorFail(ceTask, MessageException.of("simulate MessageException thrown by TaskProcessor#process"));
RuntimeException runtimeException = new RuntimeException("Simulate queue#remove failing");
@@ -580,7 +581,7 @@ public class CeWorkerImplTest {
CountDownLatch inCallLatch = new CountDownLatch(1);
CountDownLatch assertionsDoneLatch = new CountDownLatch(1);
// mock long running peek(String) call => Thread is executing call() but not running a task
- when(queue.peek(anyString(), anyBoolean())).thenAnswer((Answer<Optional<CeTask>>) invocation -> {
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenAnswer((Answer<Optional<CeTask>>) invocation -> {
inCallLatch.countDown();
try {
assertionsDoneLatch.await(10, TimeUnit.SECONDS);
@@ -615,7 +616,7 @@ public class CeWorkerImplTest {
String taskType = randomAlphabetic(12);
CeTask ceTask = mock(CeTask.class);
when(ceTask.getType()).thenReturn(taskType);
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(ceTask));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(ceTask));
taskProcessorRepository.setProcessorForTask(taskType, new SimpleCeTaskProcessor() {
@CheckForNull
@Override
@@ -654,11 +655,25 @@ public class CeWorkerImplTest {
}
@Test
+ public void do_not_exclude_portfolio_when_indexation_task_lookup_is_disabled() throws Exception {
+ // first call with empty queue to disable indexationTaskLookupEnabled
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.empty());
+ assertThat(underTest.call()).isEqualTo(NO_TASK);
+
+ ArgumentCaptor<Boolean> booleanCaptor = ArgumentCaptor.forClass(Boolean.class);
+ // following calls should not exclude portfolios
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(createCeTask(submitter)));
+ assertThat(underTest.call()).isEqualTo(TASK_PROCESSED);
+ verify(queue, times(3)).peek(anyString(), anyBoolean(), booleanCaptor.capture());
+ assertThat(booleanCaptor.getAllValues()).containsExactly(true, true, false);
+ }
+
+ @Test
public void getCurrentTask_returns_empty_when_a_thread_is_currently_calling_call_but_not_executing_a_task() throws InterruptedException {
CountDownLatch inCallLatch = new CountDownLatch(1);
CountDownLatch assertionsDoneLatch = new CountDownLatch(1);
// mock long running peek(String) call => Thread is executing call() but not running a task
- when(queue.peek(anyString(), anyBoolean())).thenAnswer((Answer<Optional<CeTask>>) invocation -> {
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenAnswer((Answer<Optional<CeTask>>) invocation -> {
inCallLatch.countDown();
try {
assertionsDoneLatch.await(10, TimeUnit.SECONDS);
@@ -689,7 +704,7 @@ public class CeWorkerImplTest {
String taskType = randomAlphabetic(12);
CeTask ceTask = mock(CeTask.class);
when(ceTask.getType()).thenReturn(taskType);
- when(queue.peek(anyString(), anyBoolean())).thenReturn(Optional.of(ceTask));
+ when(queue.peek(anyString(), anyBoolean(), anyBoolean())).thenReturn(Optional.of(ceTask));
taskProcessorRepository.setProcessorForTask(taskType, new SimpleCeTaskProcessor() {
@CheckForNull
@@ -746,7 +761,7 @@ public class CeWorkerImplTest {
}
private void verifyWorkerUuid() {
- verify(queue, atLeastOnce()).peek(workerUuidCaptor.capture(), anyBoolean());
+ verify(queue, atLeastOnce()).peek(workerUuidCaptor.capture(), anyBoolean(), anyBoolean());
assertThat(workerUuidCaptor.getValue()).isEqualTo(workerUuid);
}