[th2-2552] backpressure: added check for queue size limit

README.md

@@ -1,4 +1,4 @@
-# th2 common library (Java) (3.33.0)
+# th2 common library (Java) (3.34.0)
 
 ## Usage
 
@@ -80,6 +80,8 @@ The `CommonFactory` reads a RabbitMQ configuration from the rabbitMQ.json file.
 * maxConnectionRecoveryTimeout - this option defines a maximum interval in milliseconds between reconnect attempts, with its default value set to 60000. Common factory increases the reconnect interval values from minConnectionRecoveryTimeout to maxConnectionRecoveryTimeout.
 * prefetchCount - this option is the maximum number of messages that the server will deliver, with its value set to 0 if unlimited, the default value is set to 10.
 * messageRecursionLimit - an integer number denotes how deep nested protobuf message might be, default = 100
+* secondsToCheckVirtualQueueLimit - this option defines an interval in seconds between size check attempts, default = 10
+* batchesToCheckVirtualQueueLimit - this option defines the number of batches between size check attempts, default = 10000
 
 ```json
 {
@@ -95,7 +97,9 @@ The `CommonFactory` reads a RabbitMQ configuration from the rabbitMQ.json file.
   "minConnectionRecoveryTimeout": 10000,
   "maxConnectionRecoveryTimeout": 60000,
   "prefetchCount": 10,
-  "messageRecursionLimit": 100
+  "messageRecursionLimit": 100,
+  "secondsToCheckVirtualQueueLimit": 10,
+  "batchesToCheckVirtualQueueLimit": 10000
 }
 ```
 
@@ -117,6 +121,7 @@ The `CommonFactory` reads a message's router configuration from the `mq.json` fi
     * filters - pin's message's filters
         * metadata - a metadata filters
         * message - a message's fields filters
+  * virtualQueueLimit - MQ router calculates destination queues and compares their current size to this value. The router blocks the current thread to repeat the comparison if the size of any destination queues exceeds the virtual limit
 
 Filters format: 
 * fieldName - a field's name
@@ -154,7 +159,8 @@ Filters format:
             "operation": "WILDCARD"
           }
         ]
-      }
+      },
+      "virtualQueueLimit": 10000
     }
   }
 }
@@ -288,6 +294,12 @@ dependencies {
 
 ## Release notes
 
+### 3.34.0
+
++ Added backpressure support: lock sending if queue virtual size limit is exceeded
++ Added parameter `virtualQueueLimit` to `mq.json`
++ Added parameters `secondsToCheckVirtualQueueLimit` and `batchesToCheckVirtualQueueLimit` to `mq_router.json`
+
 ### 3.33.0
 
 + Added ability to read dictionaries by aliases and as group of all available aliases

build.gradle

@@ -171,6 +171,7 @@ dependencies {
     implementation "io.grpc:grpc-netty"
 
     implementation "com.rabbitmq:amqp-client"
+    implementation 'com.rabbitmq:http-client:4.0.0'
 
     implementation "org.jetbrains:annotations"
 

gradle.properties

@@ -1,5 +1,5 @@
 #
-# Copyright 2020-2021 Exactpro (Exactpro Systems Limited)
+# Copyright 2020-2022 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
@@ -13,7 +13,7 @@
 # limitations under the License.
 #
 
-release_version=3.33.0
+release_version=3.34.0
 
 description = 'th2 common library (Java)'
 

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

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

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2021 Exactpro (Exactpro Systems Limited)
+ * Copyright 2020-2022 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.
@@ -55,6 +55,7 @@ public abstract class AbstractRabbitSender<T> implements MessageSender<T> {
     private final AtomicReference<String> exchangeName = new AtomicReference<>();
     private final AtomicReference<ConnectionManager> connectionManager = new AtomicReference<>();
     private final String th2Type;
+    private long sentBeforeQueueSizeCheck;
 
     public AbstractRabbitSender(
             @NotNull ConnectionManager connectionManager,
@@ -81,15 +82,20 @@ public void send(T value) throws IOException {
         requireNonNull(value, "Value for send can not be null");
 
         try {
-            ConnectionManager connection = this.connectionManager.get();
+            ConnectionManager connectionManager = this.connectionManager.get();
             byte[] bytes = valueToBytes(value);
             MESSAGE_SIZE_PUBLISH_BYTES
                     .labels(th2Pin, th2Type, exchangeName.get(), routingKey.get())
                     .inc(bytes.length);
             MESSAGE_PUBLISH_TOTAL
                     .labels(th2Pin, th2Type, exchangeName.get(), routingKey.get())
                     .inc();
-            connection.basicPublish(exchangeName.get(), routingKey.get(), null, bytes);
+            sentBeforeQueueSizeCheck++;
+            if (sentBeforeQueueSizeCheck > connectionManager.getConnectionManagerConfiguration().getBatchesToCheckVirtualQueueLimit()) {
+                connectionManager.lockSendingIfSizeLimitExceeded(routingKey.get());
+                sentBeforeQueueSizeCheck = 0;
+            }
+            connectionManager.basicPublish(exchangeName.get(), routingKey.get(), null, bytes);
 
             if (LOGGER.isTraceEnabled()) {
                 LOGGER.trace("Message sent to exchangeName='{}', routing key='{}': '{}'",