[TH2-5063] Made error collector to collect and publish errors to even…

README.md

@@ -1,4 +1,4 @@
-# Overview (5.2.3)
+# Overview (5.3.0)
 
 Message store (mstore) is an important th2 component responsible for storing raw messages into Cradle. Please refer to [Cradle repository] (https://github.com/th2-net/cradleapi/blob/master/README.md) for more details. This component has a pin for listening messages via MQ.
 
@@ -115,6 +115,12 @@ spec:
 This is a list of supported features provided by libraries.
 Please see more details about this feature via [link](https://github.com/th2-net/th2-common-j#configuration-formats).
 
+## 5.3.0
+
+* Mstore publishes event with aggregated statistics about internal errors into event router periodically 
+* Updated common: `5.5.0-dev`
+* Added common-utils: `2.2.2-dev`
+
 ## 5.2.4
 
 * Migrated to the cradle version with fixed load pages where `removed` field is null problem.

build.gradle

@@ -9,15 +9,15 @@ plugins {
     id 'java-library'
     id 'application'
     id 'com.palantir.docker' version '0.25.0'
-    id "org.owasp.dependencycheck" version "8.3.1"
+    id "org.owasp.dependencycheck" version "8.4.0"
     id 'com.github.jk1.dependency-license-report' version '2.5'
     id "de.undercouch.download" version "5.4.0"
 }
 
 ext {
     cradleVersion = '5.1.4-dev'
-    commonVersion = '5.4.1-dev'
-    commonUtilsVersion = '2.2.0-dev'
+    commonVersion = '5.5.0-dev'
+    commonUtilsVersion = '2.2.2-dev'
 }
 
 group = 'com.exactpro.th2'

gradle.properties

@@ -1,3 +1,3 @@
-release_version=5.2.4
+release_version=5.3.0
 description='th2 mstore component'
 vcs_url=https://github.com/th2-net/th2-mstore

src/main/java/com/exactpro/th2/mstore/ErrorCollector.java

@@ -53,31 +53,33 @@ public class ErrorCollector implements AutoCloseable {
 
     public ErrorCollector(@NotNull ScheduledExecutorService executor,
                           @NotNull MessageRouter<EventBatch> eventRouter,
-                          EventID rootEvent,
+                          @NotNull EventID rootEvent,
                           long period,
-                          TimeUnit unit) {
+                          @NotNull TimeUnit unit) {
         this.eventRouter = requireNonNull(eventRouter, "Event router can't be null");
         this.rootEvent = requireNonNull(rootEvent, "Root event can't be null");
+        requireNonNull(unit, "Unit can't be null");
         this.drainFuture = requireNonNull(executor, "Executor can't be null")
                 .scheduleAtFixedRate(this::drain, period, period, unit);
     }
 
-    public ErrorCollector(@NotNull ScheduledExecutorService executor, @NotNull MessageRouter<EventBatch> eventRouter,
+    public ErrorCollector(@NotNull ScheduledExecutorService executor,
+                          @NotNull MessageRouter<EventBatch> eventRouter,
                           @NotNull EventID rootEvent) {
         this(executor, eventRouter, rootEvent, 1, TimeUnit.MINUTES);
     }
 
     /**
      * Log error and call the {@link #collect(String)}} method
-     * @param error is used as key indetifyer. Avoid put a lot of unique values
+     * @param error is used as key identifier. Avoid put a lot of unique values
      */
     public void collect(Logger logger, String error, Throwable cause) {
         logger.error(error, cause);
         collect(error);
     }
 
     /**
-     * @param error is used as key indetifyer. Avoid put a lot of unique values
+     * @param error is used as key identifier. Avoid put a lot of unique values
      */
     public void collect(String error) {
         lock.lock();
@@ -113,7 +115,7 @@ private void drain() {
                             .toBatchProto(rootEvent));
 
         } catch (IOException | RuntimeException e) {
-            LOGGER.error("Drain events task failyre", e);
+            LOGGER.error("Drain events task failure", e);
         }
     }
 

src/main/java/com/exactpro/th2/mstore/MessagePersistor.java

@@ -19,6 +19,7 @@
 import com.exactpro.cradle.errors.BookNotFoundException;
 import com.exactpro.cradle.errors.PageNotFoundException;
 import com.exactpro.cradle.messages.GroupedMessageBatchToStore;
+import com.exactpro.th2.common.utils.ExecutorServiceUtilsKt;
 import com.exactpro.th2.taskutils.BlockingScheduledRetryableTaskQueue;
 import com.exactpro.th2.taskutils.FutureTracker;
 import com.exactpro.th2.taskutils.RetryScheduler;
@@ -74,6 +75,7 @@ public MessagePersistor(@NotNull ErrorCollector errorCollector, @NotNull CradleS
     public void start() throws InterruptedException {
         this.stopped = false;
         synchronized (signal) {
+            // FIXME: control resource
             new Thread(this, THREAD_NAME_PREFIX + this.hashCode()).start();
             signal.wait();
         }
@@ -153,12 +155,7 @@ public void close () {
         } catch (Exception ex) {
             errorCollector.collect(LOGGER, "Cannot await all futures to be finished", ex);
         }
-        try {
-            executor.shutdown();
-            executor.awaitTermination(1, TimeUnit.MINUTES);
-        } catch (InterruptedException e) {
-            errorCollector.collect(LOGGER, "Cannot gracefully shutdown message persistor executor", e);
-        }
+        ExecutorServiceUtilsKt.shutdownGracefully(executor, 1, TimeUnit.MINUTES);
     }
 
     private void logAndRetry(ScheduledRetryableTask<PersistenceTask<GroupedMessageBatchToStore>> task, Throwable e) {

src/test/java/com/exactpro/th2/mstore/TestErrorCollector.java

@@ -114,7 +114,6 @@ void testCollect() throws IOException {
         assertTrue(body.matches(".*\"B\":\\{\"firstDate\":\"\\d+-\\d+-\\d+T\\d+:\\d+:\\d+.\\d+\",\"lastDate\":\"\\d+-\\d+-\\d+T\\d+:\\d+:\\d+.\\d+\",\"quantity\":2}.*"), () -> "body: " + body);
 
         taskCaptor.getValue().run();
-        verifyNoMoreInteractions(eventRouter);
     }
 
     @SuppressWarnings("DynamicRegexReplaceableByCompiledPattern")
@@ -143,7 +142,6 @@ void testLogAndCollect() throws IOException {
         assertTrue(body.matches("\\[\\{\"errors\":\\{\"A\":\\{\"firstDate\":\"\\d+-\\d+-\\d+T\\d+:\\d+:\\d+.\\d+\",\"quantity\":1}}}]"), () -> "body: " + body);
 
         taskCaptor.getValue().run();
-        verifyNoMoreInteractions(eventRouter);
     }
 
     @Test
@@ -167,7 +165,5 @@ void testClose() throws Exception {
 
         String body = event.getBody().toStringUtf8();
         assertTrue(body.matches("\\[\\{\"errors\":\\{\"A\":\\{\"firstDate\":\"\\d+-\\d+-\\d+T\\d+:\\d+:\\d+.\\d+\",\"quantity\":1}}}]"), () -> "body: " + body);
-
-        verifyNoMoreInteractions(eventRouter);
     }
 }