简柏林 / thingskit · Commits

Commit 827a1e30928fe0496d104ad11a7e754e34ad1609

Authored by Igor Kulikov
Committed by GitHub
2 parents 32b7ed5d 7e95f081

Merge pull request #3406 from ShvaykaD/improvements/queue-reprocessing-strategy-3.2 

[3.2] Improvements/queue reprocessing strategy
Inline Side-by-side
Showing 3 changed files with 11 additions and 1 deletions
@@ -56,7 +56,9 @@ public class TbRuleEngineProcessingStrategyFactory { @@ -56,7 +56,9 @@ public class TbRuleEngineProcessingStrategyFactory {
56 private final boolean retryTimeout; 56 private final boolean retryTimeout;
57 private final int maxRetries; 57 private final int maxRetries;
58 private final double maxAllowedFailurePercentage; 58 private final double maxAllowedFailurePercentage;
59 - private final long pauseBetweenRetries; 59 + private final long maxPauseBetweenRetries;
  60 +
  61 + private long pauseBetweenRetries;
60 62
61 private int initialTotalCount; 63 private int initialTotalCount;
62 private int retryCount; 64 private int retryCount;
@@ -69,6 +71,7 @@ public class TbRuleEngineProcessingStrategyFactory { @@ -69,6 +71,7 @@ public class TbRuleEngineProcessingStrategyFactory {
69 this.maxRetries = configuration.getRetries(); 71 this.maxRetries = configuration.getRetries();
70 this.maxAllowedFailurePercentage = configuration.getFailurePercentage(); 72 this.maxAllowedFailurePercentage = configuration.getFailurePercentage();
71 this.pauseBetweenRetries = configuration.getPauseBetweenRetries(); 73 this.pauseBetweenRetries = configuration.getPauseBetweenRetries();
  74 + this.maxPauseBetweenRetries = configuration.getMaxPauseBetweenRetries();
72 } 75 }
73 76
74 @Override 77 @Override
@@ -108,6 +111,9 @@ public class TbRuleEngineProcessingStrategyFactory { @@ -108,6 +111,9 @@ public class TbRuleEngineProcessingStrategyFactory {
108 } catch (InterruptedException e) { 111 } catch (InterruptedException e) {
109 throw new RuntimeException(e); 112 throw new RuntimeException(e);
110 } 113 }
  114 + if (maxPauseBetweenRetries > pauseBetweenRetries) {
  115 + pauseBetweenRetries = Math.min(maxPauseBetweenRetries, pauseBetweenRetries * 2);
  116 + }
111 } 117 }
112 return new TbRuleEngineProcessingDecision(false, toReprocess); 118 return new TbRuleEngineProcessingDecision(false, toReprocess);
113 } 119 }
@@ -758,6 +758,7 @@ queue: @@ -758,6 +758,7 @@ queue:
758 retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited 758 retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited
759 failure-percentage: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; 759 failure-percentage: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages;
760 pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRY_PAUSE:3}"# Time in seconds to wait in consumer thread before retries; 760 pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRY_PAUSE:3}"# Time in seconds to wait in consumer thread before retries;
  761 + max-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MAX_RETRY_PAUSE:3}"# Max allowed time in seconds for pause between retries.
761 - name: "${TB_QUEUE_RE_HP_QUEUE_NAME:HighPriority}" 762 - name: "${TB_QUEUE_RE_HP_QUEUE_NAME:HighPriority}"
762 topic: "${TB_QUEUE_RE_HP_TOPIC:tb_rule_engine.hp}" 763 topic: "${TB_QUEUE_RE_HP_TOPIC:tb_rule_engine.hp}"
763 poll-interval: "${TB_QUEUE_RE_HP_POLL_INTERVAL_MS:25}" 764 poll-interval: "${TB_QUEUE_RE_HP_POLL_INTERVAL_MS:25}"
@@ -773,6 +774,7 @@ queue: @@ -773,6 +774,7 @@ queue:
773 retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRIES:0}" # Number of retries, 0 is unlimited 774 retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRIES:0}" # Number of retries, 0 is unlimited
774 failure-percentage: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; 775 failure-percentage: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages;
775 pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; 776 pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries;
  777 + max-pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_MAX_RETRY_PAUSE:5}"# Max allowed time in seconds for pause between retries.
776 - name: "${TB_QUEUE_RE_SQ_QUEUE_NAME:SequentialByOriginator}" 778 - name: "${TB_QUEUE_RE_SQ_QUEUE_NAME:SequentialByOriginator}"
777 topic: "${TB_QUEUE_RE_SQ_TOPIC:tb_rule_engine.sq}" 779 topic: "${TB_QUEUE_RE_SQ_TOPIC:tb_rule_engine.sq}"
778 poll-interval: "${TB_QUEUE_RE_SQ_POLL_INTERVAL_MS:25}" 780 poll-interval: "${TB_QUEUE_RE_SQ_POLL_INTERVAL_MS:25}"
@@ -788,6 +790,7 @@ queue: @@ -788,6 +790,7 @@ queue:
788 retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited 790 retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited
789 failure-percentage: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; 791 failure-percentage: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages;
790 pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; 792 pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries;
  793 + max-pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_MAX_RETRY_PAUSE:5}"# Max allowed time in seconds for pause between retries.
791 transport: 794 transport:
792 # For high priority notifications that require minimum latency and processing time 795 # For high priority notifications that require minimum latency and processing time
793 notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}" 796 notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}"
@@ -24,5 +24,6 @@ public class TbRuleEngineQueueAckStrategyConfiguration { @@ -24,5 +24,6 @@ public class TbRuleEngineQueueAckStrategyConfiguration {
24 private int retries; 24 private int retries;
25 private double failurePercentage; 25 private double failurePercentage;
26 private long pauseBetweenRetries; 26 private long pauseBetweenRetries;
  27 + private long maxPauseBetweenRetries;
27 28
28 } 29 }