SONAR-9041 add job canceling worn out tasks from CE_QUEUE regularly

12 files changed, 452 insertions, 13 deletions

diff --git a/server/sonar-ce/src/main/java/org/sonar/ce/cleaning/CeCleaningExecutorService.java b/server/sonar-ce/src/main/java/org/sonar/ce/cleaning/CeCleaningExecutorService.java
new file mode 100644
index 00000000000..a5a1e5bc9a6
--- /dev/null
+++ b/server/sonar-ce/src/main/java/org/sonar/ce/cleaning/CeCleaningExecutorService.java
@@ -0,0 +1,25 @@
+/*
+ * 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.cleaning;
+
+import java.util.concurrent.ScheduledExecutorService;
+
+public interface CeCleaningExecutorService extends ScheduledExecutorService {
+}
diff --git a/server/sonar-ce/src/main/java/org/sonar/ce/cleaning/CeCleaningExecutorServiceImpl.java b/server/sonar-ce/src/main/java/org/sonar/ce/cleaning/CeCleaningExecutorServiceImpl.java
new file mode 100644
index 00000000000..26fa1626a2c
--- /dev/null
+++ b/server/sonar-ce/src/main/java/org/sonar/ce/cleaning/CeCleaningExecutorServiceImpl.java
@@ -0,0 +1,38 @@
+/*
+ * 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.cleaning;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import org.sonar.server.util.AbstractStoppableScheduledExecutorServiceImpl;
+
+public class CeCleaningExecutorServiceImpl
+  extends AbstractStoppableScheduledExecutorServiceImpl<ScheduledExecutorService>
+  implements CeCleaningExecutorService {
+
+  public CeCleaningExecutorServiceImpl() {
+    super(Executors.newSingleThreadScheduledExecutor(
+      new ThreadFactoryBuilder()
+        .setDaemon(false)
+        .setNameFormat("CE_cleaning-%d")
+        .build()));
+  }
+}
diff --git a/server/sonar-ce/src/main/java/org/sonar/ce/cleaning/CeCleaningModule.java b/server/sonar-ce/src/main/java/org/sonar/ce/cleaning/CeCleaningModule.java
new file mode 100644
index 00000000000..8a3a52d033c
--- /dev/null
+++ b/server/sonar-ce/src/main/java/org/sonar/ce/cleaning/CeCleaningModule.java
@@ -0,0 +1,31 @@
+/*
+ * 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.cleaning;
+
+import org.sonar.core.platform.Module;
+
+public class CeCleaningModule extends Module {
+  @Override
+  protected void configureModule() {
+    add(
+      CeCleaningExecutorServiceImpl.class,
+      CeCleaningSchedulerImpl.class);
+  }
+}
diff --git a/server/sonar-ce/src/main/java/org/sonar/ce/cleaning/CeCleaningScheduler.java b/server/sonar-ce/src/main/java/org/sonar/ce/cleaning/CeCleaningScheduler.java
new file mode 100644
index 00000000000..69889b60395
--- /dev/null
+++ b/server/sonar-ce/src/main/java/org/sonar/ce/cleaning/CeCleaningScheduler.java
@@ -0,0 +1,24 @@
+/*
+ * 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.cleaning;
+
+public interface CeCleaningScheduler {
+  void startScheduling();
+}
diff --git a/server/sonar-ce/src/main/java/org/sonar/ce/cleaning/CeCleaningSchedulerImpl.java b/server/sonar-ce/src/main/java/org/sonar/ce/cleaning/CeCleaningSchedulerImpl.java
new file mode 100644
index 00000000000..82675c693d6
--- /dev/null
+++ b/server/sonar-ce/src/main/java/org/sonar/ce/cleaning/CeCleaningSchedulerImpl.java
@@ -0,0 +1,58 @@
+/*
+ * 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.cleaning;
+
+import org.sonar.api.utils.log.Logger;
+import org.sonar.api.utils.log.Loggers;
+import org.sonar.ce.configuration.CeConfiguration;
+import org.sonar.ce.queue.InternalCeQueue;
+
+import static java.util.concurrent.TimeUnit.MINUTES;
+
+public class CeCleaningSchedulerImpl implements CeCleaningScheduler {
+  private static final Logger LOG = Loggers.get(CeCleaningSchedulerImpl.class);
+
+  private final CeCleaningExecutorService executorService;
+  private final CeConfiguration ceConfiguration;
+  private final InternalCeQueue internalCeQueue;
+
+  public CeCleaningSchedulerImpl(CeCleaningExecutorService executorService, CeConfiguration ceConfiguration, InternalCeQueue internalCeQueue) {
+    this.executorService = executorService;
+    this.internalCeQueue = internalCeQueue;
+    this.ceConfiguration = ceConfiguration;
+  }
+
+  @Override
+  public void startScheduling() {
+    executorService.scheduleWithFixedDelay(this::cancelWornOuts,
+      ceConfiguration.getCancelWornOutsInitialDelay(),
+      ceConfiguration.getCancelWornOutsDelay(),
+      MINUTES);
+  }
+
+  private void cancelWornOuts() {
+    try {
+      LOG.info("Deleting any worn out task");
+      internalCeQueue.cancelWornOuts();
+    } catch (Exception e) {
+      LOG.warn("Failed to cancel worn out tasks", e);
+    }
+  }
+}
diff --git a/server/sonar-ce/src/main/java/org/sonar/ce/cleaning/package-info.java b/server/sonar-ce/src/main/java/org/sonar/ce/cleaning/package-info.java
new file mode 100644
index 00000000000..a7b03390d65
--- /dev/null
+++ b/server/sonar-ce/src/main/java/org/sonar/ce/cleaning/package-info.java
@@ -0,0 +1,23 @@
+/*
+ * 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.
+ */
+@ParametersAreNonnullByDefault
+package org.sonar.ce.cleaning;
+
+import javax.annotation.ParametersAreNonnullByDefault;
diff --git a/server/sonar-ce/src/main/java/org/sonar/ce/container/ComputeEngineContainerImpl.java b/server/sonar-ce/src/main/java/org/sonar/ce/container/ComputeEngineContainerImpl.java
index c36413a9ac3..b9e8262df0b 100644
--- a/server/sonar-ce/src/main/java/org/sonar/ce/container/ComputeEngineContainerImpl.java
+++ b/server/sonar-ce/src/main/java/org/sonar/ce/container/ComputeEngineContainerImpl.java
@@ -43,6 +43,7 @@ import org.sonar.ce.CeConfigurationModule;
 import org.sonar.ce.CeHttpModule;
 import org.sonar.ce.CeQueueModule;
 import org.sonar.ce.CeTaskCommonsModule;
+import org.sonar.ce.cleaning.CeCleaningModule;
 import org.sonar.ce.db.ReadOnlyPropertiesDao;
 import org.sonar.ce.log.CeProcessLogging;
 import org.sonar.ce.platform.ComputeEngineExtensionInstaller;
@@ -393,6 +394,9 @@ public class ComputeEngineContainerImpl implements ComputeEngineContainer {
 
       InternalPropertiesImpl.class,
       ProjectSettingsFactory.class,
+
+      // cleaning
+      CeCleaningModule.class
     };
   }
 
diff --git a/server/sonar-ce/src/main/java/org/sonar/ce/queue/CeQueueInitializer.java b/server/sonar-ce/src/main/java/org/sonar/ce/queue/CeQueueInitializer.java
index d794a4c1ebb..770c3f9b1ae 100644
--- a/server/sonar-ce/src/main/java/org/sonar/ce/queue/CeQueueInitializer.java
+++ b/server/sonar-ce/src/main/java/org/sonar/ce/queue/CeQueueInitializer.java
@@ -22,6 +22,7 @@ package org.sonar.ce.queue;
 import org.sonar.api.ce.ComputeEngineSide;
 import org.sonar.api.platform.Server;
 import org.sonar.api.platform.ServerStartHandler;
+import org.sonar.ce.cleaning.CeCleaningScheduler;
 import org.sonar.ce.taskprocessor.CeProcessingScheduler;
 
 /**
@@ -32,11 +33,13 @@ import org.sonar.ce.taskprocessor.CeProcessingScheduler;
 @ComputeEngineSide
 public class CeQueueInitializer implements ServerStartHandler {
 
-  private final CeProcessingScheduler scheduler;
+  private final CeProcessingScheduler processingScheduler;
+  private final CeCleaningScheduler cleaningScheduler;
   private boolean done = false;
 
-  public CeQueueInitializer(CeProcessingScheduler scheduler) {
-    this.scheduler = scheduler;
+  public CeQueueInitializer(CeProcessingScheduler processingScheduler, CeCleaningScheduler cleaningScheduler) {
+    this.processingScheduler = processingScheduler;
+    this.cleaningScheduler = cleaningScheduler;
   }
 
   @Override
@@ -48,6 +51,7 @@ public class CeQueueInitializer implements ServerStartHandler {
   }
 
   private void initCe() {
-    scheduler.startScheduling();
+    processingScheduler.startScheduling();
+    cleaningScheduler.startScheduling();
   }
 }
diff --git a/server/sonar-ce/src/test/java/org/sonar/ce/cleaning/CeCleaningSchedulerImplTest.java b/server/sonar-ce/src/test/java/org/sonar/ce/cleaning/CeCleaningSchedulerImplTest.java
new file mode 100644
index 00000000000..5449865fc7e
--- /dev/null
+++ b/server/sonar-ce/src/test/java/org/sonar/ce/cleaning/CeCleaningSchedulerImplTest.java
@@ -0,0 +1,208 @@
+/*
+ * 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.cleaning;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.junit.Test;
+import org.sonar.ce.configuration.CeConfiguration;
+import org.sonar.ce.queue.InternalCeQueue;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class CeCleaningSchedulerImplTest {
+  @Test
+  public void startScheduling_does_not_fail_if_cancelWornOuts_send_even_an_Exception() {
+    InternalCeQueue mockedInternalCeQueue = mock(InternalCeQueue.class);
+    CeCleaningSchedulerImpl underTest = new CeCleaningSchedulerImpl(new CeCleaningAdapter() {
+      @Override
+      public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
+        // synchronously execute command
+        command.run();
+        return null;
+      }
+    }, mockCeConfiguration(1, 10), mockedInternalCeQueue);
+    doThrow(new IllegalArgumentException("faking unchecked exception thrown by cancelWornOuts")).when(mockedInternalCeQueue).cancelWornOuts();
+
+    underTest.startScheduling();
+
+    verify(mockedInternalCeQueue).cancelWornOuts();
+  }
+
+  @Test
+  public void startScheduling_fails_if_cancelWornOuts_send_even_an_Error() {
+    InternalCeQueue mockedInternalCeQueue = mock(InternalCeQueue.class);
+    CeCleaningSchedulerImpl underTest = new CeCleaningSchedulerImpl(new CeCleaningAdapter() {
+      @Override
+      public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
+        // synchronously execute command
+        command.run();
+        return null;
+      }
+    }, mockCeConfiguration(1, 10), mockedInternalCeQueue);
+    Error expected = new Error("faking Error thrown by cancelWornOuts");
+    doThrow(expected).when(mockedInternalCeQueue).cancelWornOuts();
+
+    try {
+      underTest.startScheduling();
+      fail("the error should have been thrown");
+    } catch (Error e) {
+      assertThat(e).isSameAs(expected);
+    }
+  }
+
+  @Test
+  public void startScheduling_calls_cancelWornOuts_of_internalCeQueue_at_fixed_rate_with_value_from_CeConfiguration() {
+    InternalCeQueue mockedInternalCeQueue = mock(InternalCeQueue.class);
+    long initialDelay = 10L;
+    long delay = 20L;
+    CeConfiguration mockedCeConfiguration = mockCeConfiguration(initialDelay, delay);
+    CeCleaningAdapter executorService = new CeCleaningAdapter() {
+      @Override
+      public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initDelay, long period, TimeUnit unit) {
+        assertThat(initDelay).isEqualTo(initialDelay);
+        assertThat(period).isEqualTo(delay);
+        assertThat(unit).isEqualTo(TimeUnit.MINUTES);
+        // synchronously execute command
+        command.run();
+        return null;
+      }
+    };
+    CeCleaningSchedulerImpl underTest = new CeCleaningSchedulerImpl(executorService, mockedCeConfiguration, mockedInternalCeQueue);
+
+    underTest.startScheduling();
+
+    verify(mockedInternalCeQueue).cancelWornOuts();
+  }
+
+  private CeConfiguration mockCeConfiguration(long initialDelay, long delay) {
+    CeConfiguration mockedCeConfiguration = mock(CeConfiguration.class);
+    when(mockedCeConfiguration.getCancelWornOutsInitialDelay()).thenReturn(initialDelay);
+    when(mockedCeConfiguration.getCancelWornOutsDelay()).thenReturn(delay);
+    return mockedCeConfiguration;
+  }
+
+  /**
+   * Implementation of {@link CeCleaningExecutorService} which throws {@link UnsupportedOperationException} for every
+   * method.
+   */
+  private static class CeCleaningAdapter implements CeCleaningExecutorService {
+
+    @Override
+    public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
+      throw createUnsupportedOperationException();
+    }
+
+    @Override
+    public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
+      throw createUnsupportedOperationException();
+    }
+
+    @Override
+    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
+      throw createUnsupportedOperationException();
+    }
+
+    @Override
+    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
+      throw createUnsupportedOperationException();
+    }
+
+    @Override
+    public void shutdown() {
+      throw createUnsupportedOperationException();
+    }
+
+    @Override
+    public List<Runnable> shutdownNow() {
+      throw createUnsupportedOperationException();
+    }
+
+    @Override
+    public boolean isShutdown() {
+      throw createUnsupportedOperationException();
+    }
+
+    @Override
+    public boolean isTerminated() {
+      throw createUnsupportedOperationException();
+    }
+
+    @Override
+    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
+      throw createUnsupportedOperationException();
+    }
+
+    @Override
+    public <T> Future<T> submit(Callable<T> task) {
+      throw createUnsupportedOperationException();
+    }
+
+    @Override
+    public <T> Future<T> submit(Runnable task, T result) {
+      throw createUnsupportedOperationException();
+    }
+
+    @Override
+    public Future<?> submit(Runnable task) {
+      throw createUnsupportedOperationException();
+    }
+
+    @Override
+    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
+      throw createUnsupportedOperationException();
+    }
+
+    @Override
+    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
+      throw createUnsupportedOperationException();
+    }
+
+    @Override
+    public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
+      throw createUnsupportedOperationException();
+    }
+
+    @Override
+    public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
+      throw createUnsupportedOperationException();
+    }
+
+    @Override
+    public void execute(Runnable command) {
+      throw createUnsupportedOperationException();
+    }
+
+    private UnsupportedOperationException createUnsupportedOperationException() {
+      return new UnsupportedOperationException("Unexpected call");
+    }
+  }
+}
diff --git a/server/sonar-ce/src/test/java/org/sonar/ce/configuration/CeConfigurationRule.java b/server/sonar-ce/src/test/java/org/sonar/ce/configuration/CeConfigurationRule.java
index 6bdf086d3c7..1c69bb40c25 100644
--- a/server/sonar-ce/src/test/java/org/sonar/ce/configuration/CeConfigurationRule.java
+++ b/server/sonar-ce/src/test/java/org/sonar/ce/configuration/CeConfigurationRule.java
@@ -27,8 +27,10 @@ import static com.google.common.base.Preconditions.checkArgument;
  * Mutable implementation of {@link CeConfiguration} as {@link org.junit.Rule}.
  */
 public class CeConfigurationRule extends ExternalResource implements CeConfiguration {
-  private int workerCount = CeConfigurationImpl.DEFAULT_WORKER_COUNT;
-  private long queuePollingDelay = CeConfigurationImpl.DEFAULT_QUEUE_POLLING_DELAY;
+  private int workerCount = 1;
+  private long queuePollingDelay = 2 * 1000L;
+  private long cancelWornOutsInitialDelay = 1L;
+  private long cancelWornOutsDelay = 10L;
 
   @Override
   public int getWorkerCount() {
@@ -50,4 +52,24 @@ public class CeConfigurationRule extends ExternalResource implements CeConfigura
     checkArgument(queuePollingDelay > 0, "Queue polling delay must be >= 0");
     this.queuePollingDelay = queuePollingDelay;
   }
+
+  @Override
+  public long getCancelWornOutsInitialDelay() {
+    return cancelWornOutsInitialDelay;
+  }
+
+  public void setCancelWornOutsInitialDelay(long cancelWornOutsInitialDelay) {
+    checkArgument(cancelWornOutsInitialDelay > 0, "cancel worn-outs polling initial delay must be >= 1");
+    this.cancelWornOutsInitialDelay = cancelWornOutsInitialDelay;
+  }
+
+  @Override
+  public long getCancelWornOutsDelay() {
+    return cancelWornOutsDelay;
+  }
+
+  public void setCancelWornOutsDelay(long cancelWornOutsDelay) {
+    checkArgument(cancelWornOutsDelay > 0, "cancel worn-outs polling delay must be >= 1");
+    this.cancelWornOutsDelay = cancelWornOutsDelay;
+  }
 }
diff --git a/server/sonar-ce/src/test/java/org/sonar/ce/container/ComputeEngineContainerImplTest.java b/server/sonar-ce/src/test/java/org/sonar/ce/container/ComputeEngineContainerImplTest.java
index 33fd4cdbdb0..283a88b4d0f 100644
--- a/server/sonar-ce/src/test/java/org/sonar/ce/container/ComputeEngineContainerImplTest.java
+++ b/server/sonar-ce/src/test/java/org/sonar/ce/container/ComputeEngineContainerImplTest.java
@@ -95,6 +95,7 @@ public class ComputeEngineContainerImplTest {
           + 3 // content of CeTaskCommonsModule
           + 4 // content of ProjectAnalysisTaskModule
           + 4 // content of CeTaskProcessorModule
+          + 3 // CeCleaningModule + its content
     );
     assertThat(picoContainer.getParent().getComponentAdapters()).hasSize(
       CONTAINER_ITSELF
diff --git a/server/sonar-ce/src/test/java/org/sonar/ce/queue/CeQueueInitializerTest.java b/server/sonar-ce/src/test/java/org/sonar/ce/queue/CeQueueInitializerTest.java
index 9595e966583..dbe7113b208 100644
--- a/server/sonar-ce/src/test/java/org/sonar/ce/queue/CeQueueInitializerTest.java
+++ b/server/sonar-ce/src/test/java/org/sonar/ce/queue/CeQueueInitializerTest.java
@@ -22,6 +22,7 @@ package org.sonar.ce.queue;
 import java.io.IOException;
 import org.junit.Test;
 import org.sonar.api.platform.Server;
+import org.sonar.ce.cleaning.CeCleaningScheduler;
 import org.sonar.ce.taskprocessor.CeProcessingScheduler;
 
 import static org.mockito.Mockito.mock;
@@ -32,26 +33,26 @@ import static org.mockito.Mockito.verifyZeroInteractions;
 public class CeQueueInitializerTest {
 
   private Server server = mock(Server.class);
-  private CeProcessingScheduler scheduler = mock(CeProcessingScheduler.class);
-  private CeQueueInitializer underTest = new CeQueueInitializer(scheduler);
+  private CeProcessingScheduler processingScheduler = mock(CeProcessingScheduler.class);
+  private CeCleaningScheduler cleaningScheduler = mock(CeCleaningScheduler.class);
+  private CeQueueInitializer underTest = new CeQueueInitializer(processingScheduler, cleaningScheduler);
 
   @Test
   public void clean_queue_then_start_scheduler_of_workers() throws IOException {
     underTest.onServerStart(server);
 
-    verify(scheduler).startScheduling();
+    verify(processingScheduler).startScheduling();
+    verify(cleaningScheduler).startScheduling();
   }
 
   @Test
   public void onServerStart_has_no_effect_if_called_twice_to_support_medium_test_doing_startup_tasks_multiple_times() {
-
     underTest.onServerStart(server);
-
-    reset(scheduler);
+    reset(processingScheduler, cleaningScheduler);
 
     underTest.onServerStart(server);
 
-    verifyZeroInteractions(scheduler);
+    verifyZeroInteractions(processingScheduler, cleaningScheduler);
 
   }
 }