From ef5c09545c8ce2f1c48abe566c14c31aa7bbc966 Mon Sep 17 00:00:00 2001 From: Dmytro Shvaika Date: Tue, 1 Sep 2020 16:13:42 +0300 Subject: [PATCH 1/7] added ability to use exp-pause-between-retries on queue msgs reprocessing --- ...TbRuleEngineProcessingStrategyFactory.java | 36 ++++++++++++++++--- .../src/main/resources/thingsboard.yml | 6 ++++ ...leEngineQueueAckStrategyConfiguration.java | 2 ++ 3 files changed, 40 insertions(+), 4 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java index b6220f5f94..ecec278b10 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java @@ -27,6 +27,7 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; @Component @Slf4j @@ -58,6 +59,12 @@ public class TbRuleEngineProcessingStrategyFactory { private final double maxAllowedFailurePercentage; private final long pauseBetweenRetries; + private final boolean expPauseBetweenRetries; + + private long maxExpPauseBetweenRetries; + private int maxExpDegreeValue; + private AtomicInteger expDegreeStep; + private int initialTotalCount; private int retryCount; @@ -69,6 +76,12 @@ public class TbRuleEngineProcessingStrategyFactory { this.maxRetries = configuration.getRetries(); this.maxAllowedFailurePercentage = configuration.getFailurePercentage(); this.pauseBetweenRetries = configuration.getPauseBetweenRetries(); + this.expPauseBetweenRetries = configuration.isExpPauseBetweenRetries(); + if (this.expPauseBetweenRetries) { + this.expDegreeStep = new AtomicInteger(1); + this.maxExpPauseBetweenRetries = configuration.getMaxExpPauseBetweenRetries(); + this.maxExpDegreeValue = new Double(Math.log(maxExpPauseBetweenRetries) / Math.log(pauseBetweenRetries)).intValue(); + } } @Override @@ -103,10 +116,25 @@ public class TbRuleEngineProcessingStrategyFactory { toReprocess.forEach((id, msg) -> log.trace("Going to reprocess [{}]: {}", id, TbMsg.fromBytes(result.getQueueName(), msg.getValue().getTbMsg().toByteArray(), TbMsgCallback.EMPTY))); } if (pauseBetweenRetries > 0) { - try { - Thread.sleep(TimeUnit.SECONDS.toMillis(pauseBetweenRetries)); - } catch (InterruptedException e) { - throw new RuntimeException(e); + if (expPauseBetweenRetries) { + long pause; + if (maxExpDegreeValue > expDegreeStep.get()) { + pause = new Double(Math.pow(pauseBetweenRetries, expDegreeStep.getAndIncrement())).longValue(); + } else { + pause = maxExpPauseBetweenRetries; + } + try { + Thread.sleep(TimeUnit.SECONDS.toMillis( + pause)); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } else { + try { + Thread.sleep(TimeUnit.SECONDS.toMillis(pauseBetweenRetries)); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } } } return new TbRuleEngineProcessingDecision(false, toReprocess); diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index ac0b6f614c..2c0aa3b221 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -743,6 +743,8 @@ queue: retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited failure-percentage: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRY_PAUSE:3}"# Time in seconds to wait in consumer thread before retries; + exp-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_EXP_RETRY_PAUSE:false}"# Parameter to enable/disable exponential increase of pause between retries; + max-exp-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MAX_EXP_RETRY_PAUSE:86400}"# Max allowed time in seconds for pause between retries. - name: "${TB_QUEUE_RE_HP_QUEUE_NAME:HighPriority}" topic: "${TB_QUEUE_RE_HP_TOPIC:tb_rule_engine.hp}" poll-interval: "${TB_QUEUE_RE_HP_POLL_INTERVAL_MS:25}" @@ -758,6 +760,8 @@ queue: retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRIES:0}" # Number of retries, 0 is unlimited failure-percentage: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; + exp-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_EXP_RETRY_PAUSE:false}"# Parameter to enable/disable exponential increase of pause between retries; + max-exp-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MAX_EXP_RETRY_PAUSE:86400}"# Max allowed time in seconds for pause between retries. - name: "${TB_QUEUE_RE_SQ_QUEUE_NAME:SequentialByOriginator}" topic: "${TB_QUEUE_RE_SQ_TOPIC:tb_rule_engine.sq}" poll-interval: "${TB_QUEUE_RE_SQ_POLL_INTERVAL_MS:25}" @@ -773,6 +777,8 @@ queue: retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited failure-percentage: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; + exp-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_EXP_RETRY_PAUSE:false}"# Parameter to enable/disable exponential increase of pause between retries; + max-exp-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MAX_EXP_RETRY_PAUSE:86400}"# Max allowed time in seconds for pause between retries. transport: # For high priority notifications that require minimum latency and processing time notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}" diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbRuleEngineQueueAckStrategyConfiguration.java b/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbRuleEngineQueueAckStrategyConfiguration.java index 0d21c59c9c..2e61fe8b93 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbRuleEngineQueueAckStrategyConfiguration.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbRuleEngineQueueAckStrategyConfiguration.java @@ -24,5 +24,7 @@ public class TbRuleEngineQueueAckStrategyConfiguration { private int retries; private double failurePercentage; private long pauseBetweenRetries; + private boolean expPauseBetweenRetries; + private long maxExpPauseBetweenRetries; } From 258ed349798e63b3f88f8e0c9825d5326c0779da Mon Sep 17 00:00:00 2001 From: Dmytro Shvaika Date: Tue, 1 Sep 2020 17:06:50 +0300 Subject: [PATCH 2/7] change the defaults --- .../TbRuleEngineProcessingStrategyFactory.java | 4 ++-- application/src/main/resources/thingsboard.yml | 10 +++++----- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java index ecec278b10..a14addaadc 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java @@ -62,7 +62,7 @@ public class TbRuleEngineProcessingStrategyFactory { private final boolean expPauseBetweenRetries; private long maxExpPauseBetweenRetries; - private int maxExpDegreeValue; + private double maxExpDegreeValue; private AtomicInteger expDegreeStep; private int initialTotalCount; @@ -80,7 +80,7 @@ public class TbRuleEngineProcessingStrategyFactory { if (this.expPauseBetweenRetries) { this.expDegreeStep = new AtomicInteger(1); this.maxExpPauseBetweenRetries = configuration.getMaxExpPauseBetweenRetries(); - this.maxExpDegreeValue = new Double(Math.log(maxExpPauseBetweenRetries) / Math.log(pauseBetweenRetries)).intValue(); + this.maxExpDegreeValue = Math.log(maxExpPauseBetweenRetries) / Math.log(pauseBetweenRetries); } } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 2c0aa3b221..d60d8f03d2 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -744,7 +744,7 @@ queue: failure-percentage: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRY_PAUSE:3}"# Time in seconds to wait in consumer thread before retries; exp-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_EXP_RETRY_PAUSE:false}"# Parameter to enable/disable exponential increase of pause between retries; - max-exp-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MAX_EXP_RETRY_PAUSE:86400}"# Max allowed time in seconds for pause between retries. + max-exp-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MAX_EXP_RETRY_PAUSE:25}"# Max allowed time in seconds for pause between retries. - name: "${TB_QUEUE_RE_HP_QUEUE_NAME:HighPriority}" topic: "${TB_QUEUE_RE_HP_TOPIC:tb_rule_engine.hp}" poll-interval: "${TB_QUEUE_RE_HP_POLL_INTERVAL_MS:25}" @@ -760,8 +760,8 @@ queue: retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRIES:0}" # Number of retries, 0 is unlimited failure-percentage: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; - exp-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_EXP_RETRY_PAUSE:false}"# Parameter to enable/disable exponential increase of pause between retries; - max-exp-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MAX_EXP_RETRY_PAUSE:86400}"# Max allowed time in seconds for pause between retries. + exp-pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_EXP_RETRY_PAUSE:false}"# Parameter to enable/disable exponential increase of pause between retries; + max-exp-pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_MAX_EXP_RETRY_PAUSE:120}"# Max allowed time in seconds for pause between retries. - name: "${TB_QUEUE_RE_SQ_QUEUE_NAME:SequentialByOriginator}" topic: "${TB_QUEUE_RE_SQ_TOPIC:tb_rule_engine.sq}" poll-interval: "${TB_QUEUE_RE_SQ_POLL_INTERVAL_MS:25}" @@ -777,8 +777,8 @@ queue: retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited failure-percentage: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; - exp-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_EXP_RETRY_PAUSE:false}"# Parameter to enable/disable exponential increase of pause between retries; - max-exp-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MAX_EXP_RETRY_PAUSE:86400}"# Max allowed time in seconds for pause between retries. + exp-pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_EXP_RETRY_PAUSE:false}"# Parameter to enable/disable exponential increase of pause between retries; + max-exp-pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_MAX_EXP_RETRY_PAUSE:120}"# Max allowed time in seconds for pause between retries. transport: # For high priority notifications that require minimum latency and processing time notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}" From f15426d93fa5864a89323e18ecc7c7db436139e9 Mon Sep 17 00:00:00 2001 From: Dmytro Shvaika Date: Thu, 3 Sep 2020 17:04:52 +0300 Subject: [PATCH 3/7] changed logic to multiplication pause --- ...TbRuleEngineProcessingStrategyFactory.java | 43 ++++++------------- .../src/main/resources/thingsboard.yml | 12 +++--- ...leEngineQueueAckStrategyConfiguration.java | 4 +- 3 files changed, 21 insertions(+), 38 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java index a14addaadc..e63b0cef8e 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java @@ -27,7 +27,6 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; @Component @Slf4j @@ -57,13 +56,11 @@ public class TbRuleEngineProcessingStrategyFactory { private final boolean retryTimeout; private final int maxRetries; private final double maxAllowedFailurePercentage; - private final long pauseBetweenRetries; - private final boolean expPauseBetweenRetries; + private final boolean multiplyPauseBetweenRetries; + private final long maxPauseBetweenRetries; - private long maxExpPauseBetweenRetries; - private double maxExpDegreeValue; - private AtomicInteger expDegreeStep; + private long pauseBetweenRetries; private int initialTotalCount; private int retryCount; @@ -76,12 +73,8 @@ public class TbRuleEngineProcessingStrategyFactory { this.maxRetries = configuration.getRetries(); this.maxAllowedFailurePercentage = configuration.getFailurePercentage(); this.pauseBetweenRetries = configuration.getPauseBetweenRetries(); - this.expPauseBetweenRetries = configuration.isExpPauseBetweenRetries(); - if (this.expPauseBetweenRetries) { - this.expDegreeStep = new AtomicInteger(1); - this.maxExpPauseBetweenRetries = configuration.getMaxExpPauseBetweenRetries(); - this.maxExpDegreeValue = Math.log(maxExpPauseBetweenRetries) / Math.log(pauseBetweenRetries); - } + this.multiplyPauseBetweenRetries = configuration.isMultiplyPauseBetweenRetries(); + this.maxPauseBetweenRetries = configuration.getMaxPauseBetweenRetries(); } @Override @@ -116,24 +109,14 @@ public class TbRuleEngineProcessingStrategyFactory { toReprocess.forEach((id, msg) -> log.trace("Going to reprocess [{}]: {}", id, TbMsg.fromBytes(result.getQueueName(), msg.getValue().getTbMsg().toByteArray(), TbMsgCallback.EMPTY))); } if (pauseBetweenRetries > 0) { - if (expPauseBetweenRetries) { - long pause; - if (maxExpDegreeValue > expDegreeStep.get()) { - pause = new Double(Math.pow(pauseBetweenRetries, expDegreeStep.getAndIncrement())).longValue(); - } else { - pause = maxExpPauseBetweenRetries; - } - try { - Thread.sleep(TimeUnit.SECONDS.toMillis( - pause)); - } catch (InterruptedException e) { - throw new RuntimeException(e); - } - } else { - try { - Thread.sleep(TimeUnit.SECONDS.toMillis(pauseBetweenRetries)); - } catch (InterruptedException e) { - throw new RuntimeException(e); + try { + Thread.sleep(TimeUnit.SECONDS.toMillis(pauseBetweenRetries)); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + if (multiplyPauseBetweenRetries && maxPauseBetweenRetries > 0) { + if (pauseBetweenRetries != maxPauseBetweenRetries) { + pauseBetweenRetries = Math.min(maxPauseBetweenRetries, pauseBetweenRetries * pauseBetweenRetries); } } } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index d60d8f03d2..aeea3ec341 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -743,8 +743,8 @@ queue: retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited failure-percentage: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRY_PAUSE:3}"# Time in seconds to wait in consumer thread before retries; - exp-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_EXP_RETRY_PAUSE:false}"# Parameter to enable/disable exponential increase of pause between retries; - max-exp-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MAX_EXP_RETRY_PAUSE:25}"# Max allowed time in seconds for pause between retries. + multiply-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MUL_RETRY_PAUSE:false}"# Parameter to enable/disable multiplication of pause value between retries on each iteration; + max-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MAX_MUL_RETRY_PAUSE:25}"# Max allowed time in seconds for pause between retries. - name: "${TB_QUEUE_RE_HP_QUEUE_NAME:HighPriority}" topic: "${TB_QUEUE_RE_HP_TOPIC:tb_rule_engine.hp}" poll-interval: "${TB_QUEUE_RE_HP_POLL_INTERVAL_MS:25}" @@ -760,8 +760,8 @@ queue: retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRIES:0}" # Number of retries, 0 is unlimited failure-percentage: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; - exp-pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_EXP_RETRY_PAUSE:false}"# Parameter to enable/disable exponential increase of pause between retries; - max-exp-pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_MAX_EXP_RETRY_PAUSE:120}"# Max allowed time in seconds for pause between retries. + multiply-pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_MUL_RETRY_PAUSE:false}"# Parameter to enable/disable multiplication of pause value between retries on each iteration; + max-pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_MAX_MUL_RETRY_PAUSE:120}"# Max allowed time in seconds for pause between retries. - name: "${TB_QUEUE_RE_SQ_QUEUE_NAME:SequentialByOriginator}" topic: "${TB_QUEUE_RE_SQ_TOPIC:tb_rule_engine.sq}" poll-interval: "${TB_QUEUE_RE_SQ_POLL_INTERVAL_MS:25}" @@ -777,8 +777,8 @@ queue: retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited failure-percentage: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; - exp-pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_EXP_RETRY_PAUSE:false}"# Parameter to enable/disable exponential increase of pause between retries; - max-exp-pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_MAX_EXP_RETRY_PAUSE:120}"# Max allowed time in seconds for pause between retries. + multiply-pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_MUL_RETRY_PAUSE:false}"# Parameter to enable/disable multiplication of pause value between retries on each iteration; + max-pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_MAX_MUL_RETRY_PAUSE:120}"# Max allowed time in seconds for pause between retries. transport: # For high priority notifications that require minimum latency and processing time notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}" diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbRuleEngineQueueAckStrategyConfiguration.java b/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbRuleEngineQueueAckStrategyConfiguration.java index 2e61fe8b93..9427228047 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbRuleEngineQueueAckStrategyConfiguration.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbRuleEngineQueueAckStrategyConfiguration.java @@ -24,7 +24,7 @@ public class TbRuleEngineQueueAckStrategyConfiguration { private int retries; private double failurePercentage; private long pauseBetweenRetries; - private boolean expPauseBetweenRetries; - private long maxExpPauseBetweenRetries; + private boolean multiplyPauseBetweenRetries; + private long maxPauseBetweenRetries; } From 3ceb9d3101488f8ba40f172d86bd2577a5b657aa Mon Sep 17 00:00:00 2001 From: Dmytro Shvaika Date: Fri, 4 Sep 2020 15:47:11 +0300 Subject: [PATCH 4/7] clean up code --- .../processing/TbRuleEngineProcessingStrategyFactory.java | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java index e63b0cef8e..245d02bf5d 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java @@ -114,10 +114,8 @@ public class TbRuleEngineProcessingStrategyFactory { } catch (InterruptedException e) { throw new RuntimeException(e); } - if (multiplyPauseBetweenRetries && maxPauseBetweenRetries > 0) { - if (pauseBetweenRetries != maxPauseBetweenRetries) { - pauseBetweenRetries = Math.min(maxPauseBetweenRetries, pauseBetweenRetries * pauseBetweenRetries); - } + if (multiplyPauseBetweenRetries && maxPauseBetweenRetries > pauseBetweenRetries) { + pauseBetweenRetries = Math.min(maxPauseBetweenRetries, pauseBetweenRetries * 2); } } return new TbRuleEngineProcessingDecision(false, toReprocess); From 801593010dcc64f2dc97ebd1bfd944c989bf431b Mon Sep 17 00:00:00 2001 From: Dmytro Shvaika Date: Fri, 4 Sep 2020 16:13:29 +0300 Subject: [PATCH 5/7] removed unnecessary flag --- .../TbRuleEngineProcessingStrategyFactory.java | 5 +---- application/src/main/resources/thingsboard.yml | 9 +++------ .../TbRuleEngineQueueAckStrategyConfiguration.java | 1 - 3 files changed, 4 insertions(+), 11 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java index 245d02bf5d..b42615ad8f 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java @@ -56,8 +56,6 @@ public class TbRuleEngineProcessingStrategyFactory { private final boolean retryTimeout; private final int maxRetries; private final double maxAllowedFailurePercentage; - - private final boolean multiplyPauseBetweenRetries; private final long maxPauseBetweenRetries; private long pauseBetweenRetries; @@ -73,7 +71,6 @@ public class TbRuleEngineProcessingStrategyFactory { this.maxRetries = configuration.getRetries(); this.maxAllowedFailurePercentage = configuration.getFailurePercentage(); this.pauseBetweenRetries = configuration.getPauseBetweenRetries(); - this.multiplyPauseBetweenRetries = configuration.isMultiplyPauseBetweenRetries(); this.maxPauseBetweenRetries = configuration.getMaxPauseBetweenRetries(); } @@ -114,7 +111,7 @@ public class TbRuleEngineProcessingStrategyFactory { } catch (InterruptedException e) { throw new RuntimeException(e); } - if (multiplyPauseBetweenRetries && maxPauseBetweenRetries > pauseBetweenRetries) { + if (maxPauseBetweenRetries > pauseBetweenRetries) { pauseBetweenRetries = Math.min(maxPauseBetweenRetries, pauseBetweenRetries * 2); } } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index aeea3ec341..8d4046671c 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -743,8 +743,7 @@ queue: retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited failure-percentage: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRY_PAUSE:3}"# Time in seconds to wait in consumer thread before retries; - multiply-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MUL_RETRY_PAUSE:false}"# Parameter to enable/disable multiplication of pause value between retries on each iteration; - max-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MAX_MUL_RETRY_PAUSE:25}"# Max allowed time in seconds for pause between retries. + max-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MAX_MUL_RETRY_PAUSE:0}"# Max allowed time in seconds for pause between retries. - name: "${TB_QUEUE_RE_HP_QUEUE_NAME:HighPriority}" topic: "${TB_QUEUE_RE_HP_TOPIC:tb_rule_engine.hp}" poll-interval: "${TB_QUEUE_RE_HP_POLL_INTERVAL_MS:25}" @@ -760,8 +759,7 @@ queue: retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRIES:0}" # Number of retries, 0 is unlimited failure-percentage: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; - multiply-pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_MUL_RETRY_PAUSE:false}"# Parameter to enable/disable multiplication of pause value between retries on each iteration; - max-pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_MAX_MUL_RETRY_PAUSE:120}"# Max allowed time in seconds for pause between retries. + max-pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_MAX_MUL_RETRY_PAUSE:0}"# Max allowed time in seconds for pause between retries. - name: "${TB_QUEUE_RE_SQ_QUEUE_NAME:SequentialByOriginator}" topic: "${TB_QUEUE_RE_SQ_TOPIC:tb_rule_engine.sq}" poll-interval: "${TB_QUEUE_RE_SQ_POLL_INTERVAL_MS:25}" @@ -777,8 +775,7 @@ queue: retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited failure-percentage: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; - multiply-pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_MUL_RETRY_PAUSE:false}"# Parameter to enable/disable multiplication of pause value between retries on each iteration; - max-pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_MAX_MUL_RETRY_PAUSE:120}"# Max allowed time in seconds for pause between retries. + max-pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_MAX_MUL_RETRY_PAUSE:0}"# Max allowed time in seconds for pause between retries. transport: # For high priority notifications that require minimum latency and processing time notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}" diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbRuleEngineQueueAckStrategyConfiguration.java b/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbRuleEngineQueueAckStrategyConfiguration.java index 9427228047..978794662a 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbRuleEngineQueueAckStrategyConfiguration.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbRuleEngineQueueAckStrategyConfiguration.java @@ -24,7 +24,6 @@ public class TbRuleEngineQueueAckStrategyConfiguration { private int retries; private double failurePercentage; private long pauseBetweenRetries; - private boolean multiplyPauseBetweenRetries; private long maxPauseBetweenRetries; } From d3b7a29d8b6b6f6ddfe21f9afc3c6d82932b4519 Mon Sep 17 00:00:00 2001 From: Dmytro Shvaika Date: Fri, 4 Sep 2020 16:49:31 +0300 Subject: [PATCH 6/7] changed env variable name --- application/src/main/resources/thingsboard.yml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 8d4046671c..23e16bfb39 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -743,7 +743,7 @@ queue: retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited failure-percentage: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRY_PAUSE:3}"# Time in seconds to wait in consumer thread before retries; - max-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MAX_MUL_RETRY_PAUSE:0}"# Max allowed time in seconds for pause between retries. + max-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MAX_RETRY_PAUSE:0}"# Max allowed time in seconds for pause between retries. - name: "${TB_QUEUE_RE_HP_QUEUE_NAME:HighPriority}" topic: "${TB_QUEUE_RE_HP_TOPIC:tb_rule_engine.hp}" poll-interval: "${TB_QUEUE_RE_HP_POLL_INTERVAL_MS:25}" @@ -759,7 +759,7 @@ queue: retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRIES:0}" # Number of retries, 0 is unlimited failure-percentage: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; - max-pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_MAX_MUL_RETRY_PAUSE:0}"# Max allowed time in seconds for pause between retries. + max-pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_MAX_RETRY_PAUSE:0}"# Max allowed time in seconds for pause between retries. - name: "${TB_QUEUE_RE_SQ_QUEUE_NAME:SequentialByOriginator}" topic: "${TB_QUEUE_RE_SQ_TOPIC:tb_rule_engine.sq}" poll-interval: "${TB_QUEUE_RE_SQ_POLL_INTERVAL_MS:25}" @@ -775,7 +775,7 @@ queue: retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited failure-percentage: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; - max-pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_MAX_MUL_RETRY_PAUSE:0}"# Max allowed time in seconds for pause between retries. + max-pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_MAX_RETRY_PAUSE:0}"# Max allowed time in seconds for pause between retries. transport: # For high priority notifications that require minimum latency and processing time notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}" From 7e95f081c2bfea425460879aad0fff70e3c87a58 Mon Sep 17 00:00:00 2001 From: Dmytro Shvaika Date: Fri, 4 Sep 2020 17:42:07 +0300 Subject: [PATCH 7/7] changed defaults max-pause-between-retries params --- application/src/main/resources/thingsboard.yml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 23e16bfb39..3c666937ec 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -743,7 +743,7 @@ queue: retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited failure-percentage: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRY_PAUSE:3}"# Time in seconds to wait in consumer thread before retries; - max-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MAX_RETRY_PAUSE:0}"# Max allowed time in seconds for pause between retries. + max-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MAX_RETRY_PAUSE:3}"# Max allowed time in seconds for pause between retries. - name: "${TB_QUEUE_RE_HP_QUEUE_NAME:HighPriority}" topic: "${TB_QUEUE_RE_HP_TOPIC:tb_rule_engine.hp}" poll-interval: "${TB_QUEUE_RE_HP_POLL_INTERVAL_MS:25}" @@ -759,7 +759,7 @@ queue: retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRIES:0}" # Number of retries, 0 is unlimited failure-percentage: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; - max-pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_MAX_RETRY_PAUSE:0}"# Max allowed time in seconds for pause between retries. + max-pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_MAX_RETRY_PAUSE:5}"# Max allowed time in seconds for pause between retries. - name: "${TB_QUEUE_RE_SQ_QUEUE_NAME:SequentialByOriginator}" topic: "${TB_QUEUE_RE_SQ_TOPIC:tb_rule_engine.sq}" poll-interval: "${TB_QUEUE_RE_SQ_POLL_INTERVAL_MS:25}" @@ -775,7 +775,7 @@ queue: retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited failure-percentage: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; - max-pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_MAX_RETRY_PAUSE:0}"# Max allowed time in seconds for pause between retries. + max-pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_MAX_RETRY_PAUSE:5}"# Max allowed time in seconds for pause between retries. transport: # For high priority notifications that require minimum latency and processing time notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}"