diff options
author | Pierre <pierre.guillot@sonarsource.com> | 2020-06-11 14:53:08 +0200 |
---|---|---|
committer | sonartech <sonartech@sonarsource.com> | 2020-06-26 20:04:57 +0000 |
commit | e85644a5ea11ff8cbd3f5593d917c68d5d3ee7d0 (patch) | |
tree | 5b9a1f0a08e7669bada357336fb6801d64d633c4 /server/sonar-ce | |
parent | 0a9b880143936ba89f9830d454f1dc3ef8d772b3 (diff) | |
download | sonarqube-e85644a5ea11ff8cbd3f5593d917c68d5d3ee7d0.tar.gz sonarqube-e85644a5ea11ff8cbd3f5593d917c68d5d3ee7d0.zip |
SONAR-13400 exclude portfolio (but not applications) from worker peek during indexation
Diffstat (limited to 'server/sonar-ce')
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); } |