SONAR-4591 Logg a message every 10 min when processing notifications

diff --git a/sonar-core/src/main/java/org/sonar/core/notification/DefaultNotificationManager.java b/sonar-core/src/main/java/org/sonar/core/notification/DefaultNotificationManager.java
index e57c182f566d19e31040f21f7edb659ccfead5a0..d615b9405dc360af4fcad7816d5807af9a2b0360 100644 (file)
--- a/sonar-core/src/main/java/org/sonar/core/notification/DefaultNotificationManager.java
+++ b/sonar-core/src/main/java/org/sonar/core/notification/DefaultNotificationManager.java
@@ -93,7 +93,10 @@ public class DefaultNotificationManager implements NotificationManager {
 
     // If batchSize is increased then we should return a list instead of a single element
     return notifications.get(0).toNotification();
+  }
 
+  public long count() {
+    return notificationQueueDao.count();
   }
 
   /**

diff --git a/sonar-core/src/main/java/org/sonar/core/notification/db/NotificationQueueDao.java b/sonar-core/src/main/java/org/sonar/core/notification/db/NotificationQueueDao.java
index 1152e48937b36af4627260136aba9b2810fe665c..39bde4eab794d060e9705298cc9431feb7aaf5cb 100644 (file)
--- a/sonar-core/src/main/java/org/sonar/core/notification/db/NotificationQueueDao.java
+++ b/sonar-core/src/main/java/org/sonar/core/notification/db/NotificationQueueDao.java
@@ -74,4 +74,13 @@ public class NotificationQueueDao implements BatchComponent, ServerComponent {
       MyBatis.closeQuietly(session);
     }
   }
+
+  public long count() {
+    SqlSession session = mybatis.openSession();
+    try {
+      return session.getMapper(NotificationQueueMapper.class).count();
+    } finally {
+      MyBatis.closeQuietly(session);
+    }
+  }
 }

diff --git a/sonar-core/src/main/java/org/sonar/core/notification/db/NotificationQueueDto.java b/sonar-core/src/main/java/org/sonar/core/notification/db/NotificationQueueDto.java
index a42edc1a86e293a093dacd7b68c5e9d73629f14a..22aa913a9e005153fe10ad90d3014d3ce73876f1 100644 (file)
--- a/sonar-core/src/main/java/org/sonar/core/notification/db/NotificationQueueDto.java
+++ b/sonar-core/src/main/java/org/sonar/core/notification/db/NotificationQueueDto.java
@@ -37,6 +37,7 @@ import java.io.ObjectOutputStream;
  */
 public class NotificationQueueDto {
 
+  private static final String UNABLE_TO_READ_NOTIFICATION = "Unable to read notification";
   private Long id;
   private byte[] data;
 
@@ -92,10 +93,10 @@ public class NotificationQueueDto {
       return (Notification) result;
 
     } catch (IOException e) {
-      throw new SonarException("Unable to read notification", e);
+      throw new SonarException(UNABLE_TO_READ_NOTIFICATION, e);
 
     } catch (ClassNotFoundException e) {
-      throw new SonarException("Unable to read notification", e);
+      throw new SonarException(UNABLE_TO_READ_NOTIFICATION, e);
 
     } finally {
       IOUtils.closeQuietly(byteArrayInputStream);

diff --git a/sonar-core/src/main/java/org/sonar/core/notification/db/NotificationQueueMapper.java b/sonar-core/src/main/java/org/sonar/core/notification/db/NotificationQueueMapper.java
index 120637c6e98e6f15ebdefe11ee5396cb53f998b3..323d64c22892fe4bc03be5361054411d97261bfd 100644 (file)
--- a/sonar-core/src/main/java/org/sonar/core/notification/db/NotificationQueueMapper.java
+++ b/sonar-core/src/main/java/org/sonar/core/notification/db/NotificationQueueMapper.java
@@ -33,4 +33,6 @@ public interface NotificationQueueMapper {
 
   List<NotificationQueueDto> findOldest(int count);
 
+  long count();
+
 }

diff --git a/sonar-core/src/main/resources/org/sonar/core/notification/db/NotificationQueueMapper.xml b/sonar-core/src/main/resources/org/sonar/core/notification/db/NotificationQueueMapper.xml
index 550ce3ec673bad1a1dea27474db2ea90f29ed324..4d95fc707a9e4f0c61c9546c299d9055ce366333 100644 (file)
--- a/sonar-core/src/main/resources/org/sonar/core/notification/db/NotificationQueueMapper.xml
+++ b/sonar-core/src/main/resources/org/sonar/core/notification/db/NotificationQueueMapper.xml
@@ -13,6 +13,10 @@
     delete from notifications where id=#{id}
   </delete>
 
+  <select id="count" resultType="long">
+    select count(*) from notifications
+  </select>
+
   <select id="findOldest" parameterType="int" resultType="NotificationQueue">
     select id, data
     from notifications

diff --git a/sonar-core/src/test/java/org/sonar/core/notification/db/NotificationQueueDaoTest.java b/sonar-core/src/test/java/org/sonar/core/notification/db/NotificationQueueDaoTest.java
index b9fe696504d6a7f7232c370ec2e800d430c434cd..d4b48a37d1fb80d4700764b8a0315a5ba3823969 100644 (file)
--- a/sonar-core/src/test/java/org/sonar/core/notification/db/NotificationQueueDaoTest.java
+++ b/sonar-core/src/test/java/org/sonar/core/notification/db/NotificationQueueDaoTest.java
@@ -49,6 +49,17 @@ public class NotificationQueueDaoTest extends AbstractDaoTestCase {
     assertThat(dao.findOldest(1).get(0).toNotification().getType()).isEqualTo("email");
   }
 
+  @Test
+  public void should_count_notification_queue() {
+    NotificationQueueDto notificationQueueDto = NotificationQueueDto.toNotificationQueueDto(new Notification("email"));
+
+    assertThat(dao.count()).isEqualTo(0);
+
+    dao.insert(Arrays.asList(notificationQueueDto));
+
+    assertThat(dao.count()).isEqualTo(1);
+  }
+
   @Test
   public void should_delete_notification() {
     setupData("should_delete_notification");

diff --git a/sonar-server/src/main/java/org/sonar/server/notifications/NotificationService.java b/sonar-server/src/main/java/org/sonar/server/notifications/NotificationService.java
index c2b6dbf4ab86e3a0707ecd04ce20219999e0c9f8..8a48c9000176a5536893f07bde97bc1fb03856d2 100644 (file)
--- a/sonar-server/src/main/java/org/sonar/server/notifications/NotificationService.java
+++ b/sonar-server/src/main/java/org/sonar/server/notifications/NotificationService.java
@@ -113,19 +113,40 @@ public class NotificationService implements ServerComponent {
   @VisibleForTesting
   synchronized void processQueue() {
     TIME_PROFILER.start("Processing notifications queue");
+    long start = now();
+    long lastLog = start;
+    long notifSentCount = 0;
 
     Notification notifToSend = manager.getFromQueue();
     while (notifToSend != null) {
       deliver(notifToSend);
+      notifSentCount++;
       if (stopping) {
         break;
       }
+      long now = now();
+      if (now - lastLog > 10 * 60 * 1000) {
+        long remainingNotifCount = manager.count();
+        lastLog = now;
+        long spentTimeInMinutes = (now - start) / (60 * 1000);
+        log(notifSentCount, remainingNotifCount, spentTimeInMinutes);
+      }
       notifToSend = manager.getFromQueue();
     }
 
     TIME_PROFILER.stop();
   }
 
+  @VisibleForTesting
+  void log(long notifSentCount, long remainingNotifCount, long spentTimeInMinutes) {
+    LOG.info("{} notifications sent during the past {} minutes and {} still waiting to be sent", new Object[] {notifSentCount, spentTimeInMinutes, remainingNotifCount});
+  }
+
+  @VisibleForTesting
+  long now() {
+    return System.currentTimeMillis();
+  }
+
   private void deliver(Notification notification) {
     LOG.debug("Delivering notification " + notification);
     final SetMultimap<String, NotificationChannel> recipients = HashMultimap.create();

diff --git a/sonar-server/src/test/java/org/sonar/server/notifications/NotificationServiceTest.java b/sonar-server/src/test/java/org/sonar/server/notifications/NotificationServiceTest.java
index 7c14b46a8f50dcd8f6c1c9dad68c1eaef5d9079e..dfbc3d6893510fe308b9fa88c092f09f305e1940 100644 (file)
--- a/sonar-server/src/test/java/org/sonar/server/notifications/NotificationServiceTest.java
+++ b/sonar-server/src/test/java/org/sonar/server/notifications/NotificationServiceTest.java
@@ -35,6 +35,7 @@ import static org.mockito.Matchers.same;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -203,6 +204,21 @@ public class NotificationServiceTest {
     assertThat(service.getDispatchers()).hasSize(0);
   }
 
+  @Test
+  public void shouldLogEvery10Minutes() throws InterruptedException {
+    setUpMocks(CREATOR_EVGENY, ASSIGNEE_SIMON);
+    // Emulate 2 notifications in DB
+    when(manager.getFromQueue()).thenReturn(notification).thenReturn(notification).thenReturn(null);
+    when(manager.count()).thenReturn(1L).thenReturn(0L);
+    service = spy(service);
+    // Emulate processing of each notification take 10 min to have a log each time
+    when(service.now()).thenReturn(0L).thenReturn(10 * 60 * 1000 + 1L).thenReturn(20 * 60 * 1000 + 2L);
+    service.start();
+    verify(service, timeout(100)).log(1, 1, 10);
+    verify(service, timeout(100)).log(2, 0, 20);
+    service.stop();
+  }
+
   private static Answer<Object> addUser(final String user, final NotificationChannel channel) {
     return addUser(user, new NotificationChannel[] {channel});
   }