[TH2-2212] merged with dev 5

README.md

@@ -1,4 +1,4 @@
-# th2 common library (Java) (5.7.1)
+# th2 common library (Java) (5.8.0)
 
 ## Usage
 
@@ -110,6 +110,7 @@ The `CommonFactory` reads a RabbitMQ configuration from the rabbitMQ.json file.
   "maxRecoveryAttempts": 5,
   "minConnectionRecoveryTimeout": 10000,
   "maxConnectionRecoveryTimeout": 60000,
+  "retryTimeDeviationPercent": 10,
   "prefetchCount": 10,
   "messageRecursionLimit": 100
 }
@@ -501,6 +502,14 @@ dependencies {
 
 ## Release notes
 
+### 5.8.0-dev
+
++ Added retry in case of a RabbitMQ channel or connection error (when possible).
++ Added InterruptedException to basicConsume method signature.
++ Added additional logging for RabbitMQ errors.
++ Fixed connection recovery delay time.
++ Integration tests for these scenarios.
+
 ### 5.7.1-dev
 
 #### Updated:
@@ -965,4 +974,4 @@ dependencies {
 ### 3.0.1
 
 + metrics related to time measurement of an incoming message handling (Raw / Parsed / Event) migrated to
-  Prometheus [histogram](https://prometheus.io/docs/concepts/metric_types/#histogram)
+  Prometheus [histogram](https://prometheus.io/docs/concepts/metric_types/#histogram)

build.gradle

@@ -263,12 +263,11 @@ dependencies {
     testImplementation "org.junit.jupiter:junit-jupiter:${junitVersion}"
     testImplementation 'org.mockito.kotlin:mockito-kotlin:4.0.0'
     testImplementation 'org.jetbrains.kotlin:kotlin-test-junit5'
-    testImplementation "org.testcontainers:testcontainers:1.17.1"
-    testImplementation "org.testcontainers:rabbitmq:1.17.1"
+    testImplementation "org.testcontainers:testcontainers:1.17.4"
+    testImplementation "org.testcontainers:rabbitmq:1.17.4"
     testImplementation("org.junit-pioneer:junit-pioneer:2.1.0") {
         because("system property tests")
     }
-
     testFixturesImplementation group: 'org.jetbrains.kotlin', name: 'kotlin-test-junit5', version: kotlin_version
     testFixturesImplementation "org.junit.jupiter:junit-jupiter:${junitVersion}"
 }
@@ -355,4 +354,4 @@ licenseReport {
     ]
     excludeOwnGroup = false
     allowedLicensesFile = new URL("https://raw.githubusercontent.com/th2-net/.github/main/license-compliance/gradle-license-report/allowed-licenses.json")
-}
+}

gradle.properties

@@ -13,7 +13,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-release_version=5.7.1
+
+release_version=5.8.0
 description='th2 common library (Java)'
 vcs_url=https://github.com/th2-net/th2-common-j
-kapt.include.compile.classpath=false
+kapt.include.compile.classpath=false

src/main/java/com/exactpro/th2/common/schema/factory/AbstractCommonFactory.java

@@ -658,7 +658,7 @@ protected PrometheusConfiguration loadPrometheusConfiguration() {
     }
 
     protected ConnectionManager createRabbitMQConnectionManager() {
-        return new ConnectionManager(getRabbitMqConfiguration(), getConnectionManagerConfiguration(), livenessMonitor::disable);
+        return new ConnectionManager(getRabbitMqConfiguration(), getConnectionManagerConfiguration());
     }
 
     protected ConnectionManager getRabbitMqConnectionManager() {

src/main/java/com/exactpro/th2/common/schema/grpc/router/AbstractGrpcRouter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2022 Exactpro (Exactpro Systems Limited)
+ * Copyright 2020-2023 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
@@ -15,8 +15,8 @@
 
 package com.exactpro.th2.common.schema.grpc.router;
 
-import com.exactpro.th2.common.grpc.router.ServerGrpcInterceptor;
 import com.exactpro.th2.common.grpc.router.MethodDetails;
+import com.exactpro.th2.common.grpc.router.ServerGrpcInterceptor;
 import com.exactpro.th2.common.metrics.CommonMetrics;
 import com.exactpro.th2.common.schema.grpc.configuration.GrpcConfiguration;
 import com.exactpro.th2.common.schema.grpc.configuration.GrpcRouterConfiguration;

src/main/java/com/exactpro/th2/common/schema/message/impl/rabbitmq/AbstractRabbitSubscriber.java

@@ -153,7 +153,7 @@ private void subscribe() {
             consumerMonitor.updateAndGet(previous -> previous == EMPTY_INITIALIZER
                             ? Suppliers.memoize(this::basicConsume)
                             : previous)
-                    .get(); // initialize subscribtion
+                    .get(); // initialize subscription
         } catch (Exception e) {
             throw new IllegalStateException("Can not start listening", e);
         }
@@ -189,6 +189,10 @@ private SubscriberMonitor basicConsume() {
             return connectionManager.basicConsume(queue, this::handle, this::canceled);
         } catch (IOException e) {
             throw new IllegalStateException("Can not subscribe to queue = " + queue, e);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            LOGGER.error("Interrupted exception while consuming from queue '{}'", queue);
+            throw new IllegalStateException("Thread was interrupted while consuming", e);
         }
     }
 
@@ -232,4 +236,4 @@ private void canceled(String consumerTag) {
     private static <T> Supplier<T> emptySupplier() {
         return (Supplier<T>) EMPTY_INITIALIZER;
     }
-}
+}