[th2-5267] Shutdown processor thread in MessagePersistor.close method (#125)

* Publish artifact to sonatype for reusing in other projects
* Split ErrorCollector to interface and implementation
* Updated th2 gradle plugin 0.2.4
* Added testDrainTaskParameters
---------
Co-authored-by: Oleg Smirnov <oleg.smirnov@exactprosystems.com>

Author: Nikita-Smirnov-Exactpro

.github/dependabot.yml

@@ -9,9 +9,6 @@ updates:
       - dependencies
     schedule:
       interval: daily
-    allow:
-      - dependency-name: '*'
-        dependency-type: production
   - package-ecosystem: github-actions
     directory: /
     reviewers:

.github/workflows/build-dev-release.yml

@@ -6,12 +6,16 @@ jobs:
   build:
     uses: th2-net/.github/.github/workflows/compound-java.yml@main
     with:
-      build-target: 'Docker'
+      build-target: 'Docker,Sonatype'
       devRelease: true
       createTag: true
       docker-username: ${{ github.actor }}
     secrets:
       docker-password: ${{ secrets.GITHUB_TOKEN }}
+      sonatypeUsername: ${{ secrets.SONATYPE_NEXUS_USERNAME }}
+      sonatypePassword: ${{ secrets.SONATYPE_NEXUS_PASSWORD }}
+      sonatypeSigningKey: ${{ secrets.SONATYPE_GPG_ARMORED_KEY }}
+      sonatypeSigningPassword: ${{ secrets.SONATYPE_SIGNING_PASSWORD }}
       nvd-api-key: ${{ secrets.NVD_APIKEY }}
       cisa-domain: ${{ secrets.CISA_DOMAIN }}
       cisa-user: ${{ secrets.CISA_USER }}

.github/workflows/build-release.yml

@@ -6,12 +6,16 @@ jobs:
   build:
     uses: th2-net/.github/.github/workflows/compound-java.yml@main
     with:
-      build-target: 'Docker'
+      build-target: 'Docker,Sonatype'
       devRelease: false
       createTag: true
       docker-username: ${{ github.actor }}
     secrets:
       docker-password: ${{ secrets.GITHUB_TOKEN }}
+      sonatypeUsername: ${{ secrets.SONATYPE_NEXUS_USERNAME }}
+      sonatypePassword: ${{ secrets.SONATYPE_NEXUS_PASSWORD }}
+      sonatypeSigningKey: ${{ secrets.SONATYPE_GPG_ARMORED_KEY }}
+      sonatypeSigningPassword: ${{ secrets.SONATYPE_SIGNING_PASSWORD }}
       nvd-api-key: ${{ secrets.NVD_APIKEY }}
       cisa-domain: ${{ secrets.CISA_DOMAIN }}
       cisa-user: ${{ secrets.CISA_USER }}

.github/workflows/build-sanpshot.yml

@@ -12,10 +12,14 @@ jobs:
   build-job:
     uses: th2-net/.github/.github/workflows/compound-java-dev.yml@main
     with:
-      build-target: 'Docker'
+      build-target: 'Docker,Sonatype'
       docker-username: ${{ github.actor }}
     secrets:
       docker-password: ${{ secrets.GITHUB_TOKEN }}
+      sonatypeUsername: ${{ secrets.SONATYPE_NEXUS_USERNAME }}
+      sonatypePassword: ${{ secrets.SONATYPE_NEXUS_PASSWORD }}
+      sonatypeSigningKey: ${{ secrets.SONATYPE_GPG_ARMORED_KEY }}
+      sonatypeSigningPassword: ${{ secrets.SONATYPE_SIGNING_PASSWORD }}
       nvd-api-key: ${{ secrets.NVD_APIKEY }}
       cisa-domain: ${{ secrets.CISA_DOMAIN }}
       cisa-user: ${{ secrets.CISA_USER }}

Dockerfile

@@ -1,4 +1,4 @@
-FROM gradle:8.11.1-jdk11 AS build
+FROM gradle:8.11.1-jdk21 AS build
 ARG release_version
 COPY ./ .
 RUN gradle --no-daemon clean build dockerPrepare -Prelease_version=${release_version}

README.md

@@ -1,4 +1,4 @@
-# Overview (5.9.0)
+# Overview (5.10.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.
 
@@ -45,7 +45,7 @@ book, session alias, direction and sequence number are a **compound unique ident
   "drain-interval" : 100,
   "prefetchRatioToDrain" : 0.8,
   "maxBatchSize" : 200000,
-  
+  "persisotr-termination-timeout" : 1000,
   "termination-timeout" : 1000
 }
 ```
@@ -59,6 +59,7 @@ book, session alias, direction and sequence number are a **compound unique ident
 + _prefetchRatioToDrain_ - Threshold ratio of fetched messages to RabbitMQ prefetch size to force aggregated batch draining.
 + _maxBatchSize_ - Maximum aggregated batch size in case of re-batching
 + _termination-timeout_ - Timeout in milliseconds to await for the inner drain scheduler to finish all the tasks. The default value is 5000.
++ _persisotr-termination-timeout_ - The timeout in milliseconds to await for the persisotr thread complete. The default value is 5000.
 
 If some of these parameters are not provided, mstore will use default(undocumented) value.
 If _maxTaskCount_ or _maxTaskDataSize_ limits are reached during processing, mstore will pause processing new messages
@@ -89,6 +90,7 @@ spec:
     prefetchRatioToDrain: 0.8
     maxBatchSize: 200000
     termination-timeout: 5000
+    persisotr-termination-timeout: 5000
   pins:
     mq:
       subscribers:
@@ -117,9 +119,13 @@ Please see more details about this feature via [link](https://github.com/th2-net
 
 # Release notes
 
-## next release
-* Updated:
-  * th2 gradle plugin `0.2.3` based on bom: `4.11.0`
+## 5.10.0
+* Publish mstore as moven artifact.
+* Prepared MessagePersistor class for using in other project.
+* Added `persisotr-termination-timeout` option
+* Updated
+    * th2 gradle plugin `0.2.4` based on bom: `4.11.0`
+    * task-utils: `0.1.3`
 
 ## 5.9.0
 * Updated

build.gradle

@@ -1,19 +1,13 @@
 plugins {
     id "application"
-    id "com.exactpro.th2.gradle.component" version "0.2.4"
-}
-
-ext {
-    cradleVersion = '5.4.4-dev'
+    id "maven-publish"
+    alias(libs.plugins.th2.publish)
+    alias(libs.plugins.th2.component)
 }
 
 group = 'com.exactpro.th2'
 version = release_version
 
-th2JavaRelease {
-    targetJavaVersion.set(JavaVersion.VERSION_11)
-}
-
 repositories {
     mavenCentral()
 
@@ -38,11 +32,11 @@ javadoc {
 }
 
 dependencies {
-    implementation 'com.exactpro.th2:common:5.14.0-dev'
-    implementation('com.exactpro.th2:common-utils:2.3.0-dev') {
+    implementation(libs.th2.common)
+    implementation(libs.th2.common.utils) {
         because("executor service utils is used")
     }
-    implementation 'com.exactpro.th2:task-utils:0.1.3'
+    implementation(libs.th2.task.utils)
 
     implementation 'com.google.protobuf:protobuf-java-util'
 
@@ -53,14 +47,14 @@ dependencies {
 
     implementation 'io.netty:netty-buffer'
 
-    implementation "com.exactpro.th2:cradle-core:$cradleVersion"
-    implementation "com.exactpro.th2:cradle-cassandra:$cradleVersion"
+    implementation(libs.cradle.core)
+    implementation(libs.cradle.cassandra)
 
     implementation 'org.apache.commons:commons-lang3'
 
     implementation "org.slf4j:slf4j-api"
-    testImplementation 'org.junit.jupiter:junit-jupiter:5.12.0'
-    testImplementation 'org.mockito:mockito-junit-jupiter:5.16.0'
+    testImplementation(libs.junit.jupiter)
+    testImplementation(libs.mockito.junit.jupiter)
     testRuntimeOnly('org.junit.platform:junit-platform-launcher') {
         because 'This artifact will not be added automatically since gradle 9.0. You can read more information here:' +
                 'https://docs.gradle.org/8.3/userguide/upgrading_version_8.html#test_framework_implementation_dependencies'

gradle.properties

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

gradle/libs.versions.toml

@@ -0,0 +1,18 @@
+[versions]
+th2-plugin = "0.2.4"
+cradle = "5.4.4-dev"
+
+[libraries]
+th2-common-utils = { group = "com.exactpro.th2", name = "common-utils", version = "2.3.0-dev" }
+th2-common = { group = "com.exactpro.th2", name = "common", version = "5.14.0-dev" }
+th2-task-utils = { group = "com.exactpro.th2", name = "task-utils", version = "0.1.3" }
+
+cradle-core = { group = "com.exactpro.th2", name = "cradle-core", version.ref = "cradle" }
+cradle-cassandra = { group = "com.exactpro.th2", name = "cradle-cassandra", version.ref = "cradle" }
+
+junit-jupiter = { group = "org.junit.jupiter", name = "junit-jupiter", version = "5.12.0" }
+mockito-junit-jupiter = { group = "org.mockito", name = "mockito-junit-jupiter", version = "5.16.0" }
+
+[plugins]
+th2-publish = { id = "com.exactpro.th2.gradle.publish", version.ref = "th2-plugin" }
+th2-component = { id = "com.exactpro.th2.gradle.component", version.ref = "th2-plugin" }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2022 Exactpro (Exactpro Systems Limited)
+ * Copyright 2020-2025 Exactpro (Exactpro Systems Limited)
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
@@ -24,6 +24,7 @@
 public class Configuration {
     private long  drainInterval;
     private long  terminationTimeout;
+    private long  persisotrTerminationTimeout;
     private int   maxTaskCount;
     private long  maxTaskDataSize;
     private int   maxRetryCount;
@@ -49,6 +50,11 @@ public long getTerminationTimeout() {
         return terminationTimeout;
     }
 
+    @JsonProperty("persisotr-termination-timeout")
+    public long getPersisotrTerminationTimeout() {
+        return persisotrTerminationTimeout;
+    }
+
     public Long getMaxTaskDataSize() {
         return maxTaskDataSize;
     }
@@ -101,6 +107,10 @@ public static final class Builder {
         @JsonPropertyDescription("The timeout in milliseconds to await for the inner drain scheduler to finish all the tasks")
         private long terminationTimeout;
 
+        @JsonProperty("persisotr-termination-timeout")
+        @JsonPropertyDescription("The timeout in milliseconds to await for the persisotr thread complete")
+        private long persisotrTerminationTimeout;
+
         @JsonProperty("maxTaskCount")
         private Integer maxTaskCount;
 
@@ -126,6 +136,7 @@ public static final class Builder {
         private Builder() {
             drainInterval = DEFAULT_DRAIN_INTERVAL;
             terminationTimeout = DEFAULT_WAIT_TIMEOUT;
+            persisotrTerminationTimeout = DEFAULT_WAIT_TIMEOUT;
             maxTaskDataSize = Runtime.getRuntime().totalMemory()  / 2;
             maxTaskCount = DEFAULT_MAX_TASK_COUNT;
             maxRetryCount = DEFAULT_MAX_TASK_RETRIES;
@@ -146,6 +157,11 @@ public Builder withTerminationTimeout(long terminationTimeout) {
             return this;
         }
 
+        public Builder withPersisotrTerminationTimeout(long terminationTimeout) {
+            this.persisotrTerminationTimeout = terminationTimeout;
+            return this;
+        }
+
         public Builder withMaxTaskCount(Integer maxTaskCount) {
             this.maxTaskCount = maxTaskCount;
             return this;
@@ -186,6 +202,7 @@ public Configuration build() {
             Configuration configuration = new Configuration();
             configuration.drainInterval = drainInterval;
             configuration.terminationTimeout = terminationTimeout;
+            configuration.persisotrTerminationTimeout = persisotrTerminationTimeout;
             configuration.maxRetryCount = this.maxRetryCount;
             configuration.maxTaskDataSize = this.maxTaskDataSize;
             configuration.retryDelayBase = this.retryDelayBase;