001 /**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements. See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License. You may obtain a copy of the License at
008 *
009 * http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017 package org.apache.camel.processor;
018
019 import java.io.Serializable;
020 import java.util.Random;
021
022 import org.apache.camel.Exchange;
023 import org.apache.camel.LoggingLevel;
024 import org.apache.camel.Predicate;
025 import org.apache.camel.util.ObjectHelper;
026 import org.slf4j.Logger;
027 import org.slf4j.LoggerFactory;
028
029 /**
030 * The policy used to decide how many times to redeliver and the time between
031 * the redeliveries before being sent to a <a
032 * href="http://camel.apache.org/dead-letter-channel.html">Dead Letter
033 * Channel</a>
034 * <p>
035 * The default values are:
036 * <ul>
037 * <li>maximumRedeliveries = 0</li>
038 * <li>redeliveryDelay = 1000L (the initial delay)</li>
039 * <li>maximumRedeliveryDelay = 60 * 1000L</li>
040 * <li>asyncDelayedRedelivery = false</li>
041 * <li>backOffMultiplier = 2</li>
042 * <li>useExponentialBackOff = false</li>
043 * <li>collisionAvoidanceFactor = 0.15d</li>
044 * <li>useCollisionAvoidance = false</li>
045 * <li>retriesExhaustedLogLevel = LoggingLevel.ERROR</li>
046 * <li>retryAttemptedLogLevel = LoggingLevel.DEBUG</li>
047 * <li>logRetryAttempted = true</li>
048 * <li>logRetryStackTrace = false</li>
049 * <li>logStackTrace = true</li>
050 * <li>logHandled = false</li>
051 * <li>logExhausted = true</li>
052 * <li>logExhaustedMessageHistory = true</li>
053 * </ul>
054 * <p/>
055 * Setting the maximumRedeliveries to a negative value such as -1 will then always redeliver (unlimited).
056 * Setting the maximumRedeliveries to 0 will disable redelivery.
057 * <p/>
058 * This policy can be configured either by one of the following two settings:
059 * <ul>
060 * <li>using conventional options, using all the options defined above</li>
061 * <li>using delay pattern to declare intervals for delays</li>
062 * </ul>
063 * <p/>
064 * <b>Note:</b> If using delay patterns then the following options is not used (delay, backOffMultiplier, useExponentialBackOff, useCollisionAvoidance)
065 * <p/>
066 * <b>Using delay pattern</b>:
067 * <br/>The delay pattern syntax is: <tt>limit:delay;limit 2:delay 2;limit 3:delay 3;...;limit N:delay N</tt>.
068 * <p/>
069 * How it works is best illustrate with an example with this pattern: <tt>delayPattern=5:1000;10:5000:20:20000</tt>
070 * <br/>The delays will be for attempt in range 0..4 = 0 millis, 5..9 = 1000 millis, 10..19 = 5000 millis, >= 20 = 20000 millis.
071 * <p/>
072 * If you want to set a starting delay, then use 0 as the first limit, eg: <tt>0:1000;5:5000</tt> will use 1 sec delay
073 * until attempt number 5 where it will use 5 seconds going forward.
074 *
075 * @version
076 */
077 public class RedeliveryPolicy implements Cloneable, Serializable {
078 protected static Random randomNumberGenerator;
079 private static final long serialVersionUID = -338222777701473252L;
080 private static final Logger LOG = LoggerFactory.getLogger(RedeliveryPolicy.class);
081
082 protected long redeliveryDelay = 1000L;
083 protected int maximumRedeliveries;
084 protected long maximumRedeliveryDelay = 60 * 1000L;
085 protected double backOffMultiplier = 2;
086 protected boolean useExponentialBackOff;
087 // +/-15% for a 30% spread -cgs
088 protected double collisionAvoidanceFactor = 0.15d;
089 protected boolean useCollisionAvoidance;
090 protected LoggingLevel retriesExhaustedLogLevel = LoggingLevel.ERROR;
091 protected LoggingLevel retryAttemptedLogLevel = LoggingLevel.DEBUG;
092 protected boolean logStackTrace = true;
093 protected boolean logRetryStackTrace;
094 protected boolean logHandled;
095 protected boolean logContinued;
096 protected boolean logExhausted = true;
097 protected boolean logExhaustedMessageHistory = true;
098 protected boolean logRetryAttempted = true;
099 protected String delayPattern;
100 protected boolean asyncDelayedRedelivery;
101 protected boolean allowRedeliveryWhileStopping = true;
102
103 public RedeliveryPolicy() {
104 }
105
106 @Override
107 public String toString() {
108 return "RedeliveryPolicy[maximumRedeliveries=" + maximumRedeliveries
109 + ", redeliveryDelay=" + redeliveryDelay
110 + ", maximumRedeliveryDelay=" + maximumRedeliveryDelay
111 + ", asyncDelayedRedelivery=" + asyncDelayedRedelivery
112 + ", allowRedeliveryWhileStopping=" + allowRedeliveryWhileStopping
113 + ", retriesExhaustedLogLevel=" + retriesExhaustedLogLevel
114 + ", retryAttemptedLogLevel=" + retryAttemptedLogLevel
115 + ", logRetryAttempted=" + logRetryAttempted
116 + ", logStackTrace=" + logStackTrace
117 + ", logRetryStackTrace=" + logRetryStackTrace
118 + ", logHandled=" + logHandled
119 + ", logContinued=" + logContinued
120 + ", logExhausted=" + logExhausted
121 + ", logExhaustedMessageHistory=" + logExhaustedMessageHistory
122 + ", useExponentialBackOff=" + useExponentialBackOff
123 + ", backOffMultiplier=" + backOffMultiplier
124 + ", useCollisionAvoidance=" + useCollisionAvoidance
125 + ", collisionAvoidanceFactor=" + collisionAvoidanceFactor
126 + ", delayPattern=" + delayPattern + "]";
127 }
128
129 public RedeliveryPolicy copy() {
130 try {
131 return (RedeliveryPolicy)clone();
132 } catch (CloneNotSupportedException e) {
133 throw new RuntimeException("Could not clone: " + e, e);
134 }
135 }
136
137 /**
138 * Returns true if the policy decides that the message exchange should be
139 * redelivered.
140 *
141 * @param exchange the current exchange
142 * @param redeliveryCounter the current retry counter
143 * @param retryWhile an optional predicate to determine if we should redeliver or not
144 * @return true to redeliver, false to stop
145 */
146 public boolean shouldRedeliver(Exchange exchange, int redeliveryCounter, Predicate retryWhile) {
147 // predicate is always used if provided
148 if (retryWhile != null) {
149 return retryWhile.matches(exchange);
150 }
151
152 if (getMaximumRedeliveries() < 0) {
153 // retry forever if negative value
154 return true;
155 }
156 // redeliver until we hit the max
157 return redeliveryCounter <= getMaximumRedeliveries();
158 }
159
160
161 /**
162 * Calculates the new redelivery delay based on the last one and then <b>sleeps</b> for the necessary amount of time.
163 * <p/>
164 * This implementation will block while sleeping.
165 *
166 * @param redeliveryDelay previous redelivery delay
167 * @param redeliveryCounter number of previous redelivery attempts
168 * @return the calculate delay
169 * @throws InterruptedException is thrown if the sleep is interrupted likely because of shutdown
170 */
171 public long sleep(long redeliveryDelay, int redeliveryCounter) throws InterruptedException {
172 redeliveryDelay = calculateRedeliveryDelay(redeliveryDelay, redeliveryCounter);
173
174 if (redeliveryDelay > 0) {
175 sleep(redeliveryDelay);
176 }
177 return redeliveryDelay;
178 }
179
180 /**
181 * Sleeps for the given delay
182 *
183 * @param redeliveryDelay the delay
184 * @throws InterruptedException is thrown if the sleep is interrupted likely because of shutdown
185 */
186 public void sleep(long redeliveryDelay) throws InterruptedException {
187 LOG.debug("Sleeping for: {} millis until attempting redelivery", redeliveryDelay);
188 Thread.sleep(redeliveryDelay);
189 }
190
191 /**
192 * Calculates the new redelivery delay based on the last one
193 *
194 * @param previousDelay previous redelivery delay
195 * @param redeliveryCounter number of previous redelivery attempts
196 * @return the calculate delay
197 */
198 public long calculateRedeliveryDelay(long previousDelay, int redeliveryCounter) {
199 if (ObjectHelper.isNotEmpty(delayPattern)) {
200 // calculate delay using the pattern
201 return calculateRedeliverDelayUsingPattern(delayPattern, redeliveryCounter);
202 }
203
204 // calculate the delay using the conventional parameters
205 long redeliveryDelayResult;
206 if (previousDelay == 0) {
207 redeliveryDelayResult = redeliveryDelay;
208 } else if (useExponentialBackOff && backOffMultiplier > 1) {
209 redeliveryDelayResult = Math.round(backOffMultiplier * previousDelay);
210 } else {
211 redeliveryDelayResult = previousDelay;
212 }
213
214 if (useCollisionAvoidance) {
215
216 /*
217 * First random determines +/-, second random determines how far to
218 * go in that direction. -cgs
219 */
220 Random random = getRandomNumberGenerator();
221 double variance = (random.nextBoolean() ? collisionAvoidanceFactor : -collisionAvoidanceFactor)
222 * random.nextDouble();
223 redeliveryDelayResult += redeliveryDelayResult * variance;
224 }
225
226 // ensure the calculated result is not bigger than the max delay (if configured)
227 if (maximumRedeliveryDelay > 0 && redeliveryDelayResult > maximumRedeliveryDelay) {
228 redeliveryDelayResult = maximumRedeliveryDelay;
229 }
230
231 return redeliveryDelayResult;
232 }
233
234 /**
235 * Calculates the delay using the delay pattern
236 */
237 protected static long calculateRedeliverDelayUsingPattern(String delayPattern, int redeliveryCounter) {
238 String[] groups = delayPattern.split(";");
239 // find the group where the redelivery counter matches
240 long answer = 0;
241 for (String group : groups) {
242 long delay = Long.valueOf(ObjectHelper.after(group, ":"));
243 int count = Integer.valueOf(ObjectHelper.before(group, ":"));
244 if (count > redeliveryCounter) {
245 break;
246 } else {
247 answer = delay;
248 }
249 }
250
251 return answer;
252 }
253
254 // Builder methods
255 // -------------------------------------------------------------------------
256
257 /**
258 * Sets the initial redelivery delay in milliseconds
259 *
260 * @deprecated will be removed in the near future. Instead use {@link #redeliveryDelay(long)} instead
261 */
262 @Deprecated
263 public RedeliveryPolicy redeliverDelay(long delay) {
264 return redeliveryDelay(delay);
265 }
266
267 /**
268 * Sets the initial redelivery delay in milliseconds
269 */
270 public RedeliveryPolicy redeliveryDelay(long delay) {
271 setRedeliveryDelay(delay);
272 return this;
273 }
274
275 /**
276 * Sets the maximum number of times a message exchange will be redelivered
277 */
278 public RedeliveryPolicy maximumRedeliveries(int maximumRedeliveries) {
279 setMaximumRedeliveries(maximumRedeliveries);
280 return this;
281 }
282
283 /**
284 * Enables collision avoidance which adds some randomization to the backoff
285 * timings to reduce contention probability
286 */
287 public RedeliveryPolicy useCollisionAvoidance() {
288 setUseCollisionAvoidance(true);
289 return this;
290 }
291
292 /**
293 * Enables exponential backoff using the {@link #getBackOffMultiplier()} to
294 * increase the time between retries
295 */
296 public RedeliveryPolicy useExponentialBackOff() {
297 setUseExponentialBackOff(true);
298 return this;
299 }
300
301 /**
302 * Enables exponential backoff and sets the multiplier used to increase the
303 * delay between redeliveries
304 */
305 public RedeliveryPolicy backOffMultiplier(double multiplier) {
306 useExponentialBackOff();
307 setBackOffMultiplier(multiplier);
308 return this;
309 }
310
311 /**
312 * Enables collision avoidance and sets the percentage used
313 */
314 public RedeliveryPolicy collisionAvoidancePercent(double collisionAvoidancePercent) {
315 useCollisionAvoidance();
316 setCollisionAvoidancePercent(collisionAvoidancePercent);
317 return this;
318 }
319
320 /**
321 * Sets the maximum redelivery delay if using exponential back off.
322 * Use -1 if you wish to have no maximum
323 */
324 public RedeliveryPolicy maximumRedeliveryDelay(long maximumRedeliveryDelay) {
325 setMaximumRedeliveryDelay(maximumRedeliveryDelay);
326 return this;
327 }
328
329 /**
330 * Sets the logging level to use for log messages when retries have been exhausted.
331 */
332 public RedeliveryPolicy retriesExhaustedLogLevel(LoggingLevel retriesExhaustedLogLevel) {
333 setRetriesExhaustedLogLevel(retriesExhaustedLogLevel);
334 return this;
335 }
336
337 /**
338 * Sets the logging level to use for log messages when retries are attempted.
339 */
340 public RedeliveryPolicy retryAttemptedLogLevel(LoggingLevel retryAttemptedLogLevel) {
341 setRetryAttemptedLogLevel(retryAttemptedLogLevel);
342 return this;
343 }
344
345 /**
346 * Sets whether to log retry attempts
347 */
348 public RedeliveryPolicy logRetryAttempted(boolean logRetryAttempted) {
349 setLogRetryAttempted(logRetryAttempted);
350 return this;
351 }
352
353 /**
354 * Sets whether to log stacktrace for failed messages.
355 */
356 public RedeliveryPolicy logStackTrace(boolean logStackTrace) {
357 setLogStackTrace(logStackTrace);
358 return this;
359 }
360
361 /**
362 * Sets whether to log stacktrace for failed redelivery attempts
363 */
364 public RedeliveryPolicy logRetryStackTrace(boolean logRetryStackTrace) {
365 setLogRetryStackTrace(logRetryStackTrace);
366 return this;
367 }
368
369 /**
370 * Sets whether to log errors even if its handled
371 */
372 public RedeliveryPolicy logHandled(boolean logHandled) {
373 setLogHandled(logHandled);
374 return this;
375 }
376
377 /**
378 * Sets whether to log exhausted errors
379 */
380 public RedeliveryPolicy logExhausted(boolean logExhausted) {
381 setLogExhausted(logExhausted);
382 return this;
383 }
384
385 /**
386 * Sets whether to log exhausted errors including message history
387 */
388 public RedeliveryPolicy logExhaustedMessageHistory(boolean logExhaustedMessageHistory) {
389 setLogExhaustedMessageHistory(logExhaustedMessageHistory);
390 return this;
391 }
392
393 /**
394 * Sets the delay pattern with delay intervals.
395 */
396 public RedeliveryPolicy delayPattern(String delayPattern) {
397 setDelayPattern(delayPattern);
398 return this;
399 }
400
401 /**
402 * Disables redelivery by setting maximum redeliveries to 0.
403 */
404 public RedeliveryPolicy disableRedelivery() {
405 setMaximumRedeliveries(0);
406 return this;
407 }
408
409 /**
410 * Allow asynchronous delayed redelivery.
411 *
412 * @see #setAsyncDelayedRedelivery(boolean)
413 */
414 public RedeliveryPolicy asyncDelayedRedelivery() {
415 setAsyncDelayedRedelivery(true);
416 return this;
417 }
418
419 /**
420 * Controls whether to allow redelivery while stopping/shutting down a route that uses error handling.
421 *
422 * @param redeliverWhileStopping <tt>true</tt> to allow redelivery, <tt>false</tt> to reject redeliveries
423 */
424 public RedeliveryPolicy allowRedeliveryWhileStopping(boolean redeliverWhileStopping) {
425 setAllowRedeliveryWhileStopping(redeliverWhileStopping);
426 return this;
427 }
428
429 // Properties
430 // -------------------------------------------------------------------------
431
432 /**
433 * @deprecated will be removed in the near future. Instead use {@link #getRedeliveryDelay()}
434 */
435 @Deprecated
436 public long getRedeliverDelay() {
437 return getRedeliveryDelay();
438 }
439
440 /**
441 * @deprecated will be removed in the near future. Instead use {@link #setRedeliveryDelay(long)}
442 */
443 @Deprecated
444 public void setRedeliverDelay(long redeliveryDelay) {
445 setRedeliveryDelay(redeliveryDelay);
446 }
447
448 public long getRedeliveryDelay() {
449 return redeliveryDelay;
450 }
451
452 /**
453 * Sets the initial redelivery delay in milliseconds
454 */
455 public void setRedeliveryDelay(long redeliverDelay) {
456 this.redeliveryDelay = redeliverDelay;
457 // if max enabled then also set max to this value in case max was too low
458 if (maximumRedeliveryDelay > 0 && redeliverDelay > maximumRedeliveryDelay) {
459 this.maximumRedeliveryDelay = redeliverDelay;
460 }
461 }
462
463 public double getBackOffMultiplier() {
464 return backOffMultiplier;
465 }
466
467 /**
468 * Sets the multiplier used to increase the delay between redeliveries if
469 * {@link #setUseExponentialBackOff(boolean)} is enabled
470 */
471 public void setBackOffMultiplier(double backOffMultiplier) {
472 this.backOffMultiplier = backOffMultiplier;
473 }
474
475 public long getCollisionAvoidancePercent() {
476 return Math.round(collisionAvoidanceFactor * 100);
477 }
478
479 /**
480 * Sets the percentage used for collision avoidance if enabled via
481 * {@link #setUseCollisionAvoidance(boolean)}
482 */
483 public void setCollisionAvoidancePercent(double collisionAvoidancePercent) {
484 this.collisionAvoidanceFactor = collisionAvoidancePercent * 0.01d;
485 }
486
487 public double getCollisionAvoidanceFactor() {
488 return collisionAvoidanceFactor;
489 }
490
491 /**
492 * Sets the factor used for collision avoidance if enabled via
493 * {@link #setUseCollisionAvoidance(boolean)}
494 */
495 public void setCollisionAvoidanceFactor(double collisionAvoidanceFactor) {
496 this.collisionAvoidanceFactor = collisionAvoidanceFactor;
497 }
498
499 public int getMaximumRedeliveries() {
500 return maximumRedeliveries;
501 }
502
503 /**
504 * Sets the maximum number of times a message exchange will be redelivered.
505 * Setting a negative value will retry forever.
506 */
507 public void setMaximumRedeliveries(int maximumRedeliveries) {
508 this.maximumRedeliveries = maximumRedeliveries;
509 }
510
511 public long getMaximumRedeliveryDelay() {
512 return maximumRedeliveryDelay;
513 }
514
515 /**
516 * Sets the maximum redelivery delay.
517 * Use -1 if you wish to have no maximum
518 */
519 public void setMaximumRedeliveryDelay(long maximumRedeliveryDelay) {
520 this.maximumRedeliveryDelay = maximumRedeliveryDelay;
521 }
522
523 public boolean isUseCollisionAvoidance() {
524 return useCollisionAvoidance;
525 }
526
527 /**
528 * Enables/disables collision avoidance which adds some randomization to the
529 * backoff timings to reduce contention probability
530 */
531 public void setUseCollisionAvoidance(boolean useCollisionAvoidance) {
532 this.useCollisionAvoidance = useCollisionAvoidance;
533 }
534
535 public boolean isUseExponentialBackOff() {
536 return useExponentialBackOff;
537 }
538
539 /**
540 * Enables/disables exponential backoff using the
541 * {@link #getBackOffMultiplier()} to increase the time between retries
542 */
543 public void setUseExponentialBackOff(boolean useExponentialBackOff) {
544 this.useExponentialBackOff = useExponentialBackOff;
545 }
546
547 protected static synchronized Random getRandomNumberGenerator() {
548 if (randomNumberGenerator == null) {
549 randomNumberGenerator = new Random();
550 }
551 return randomNumberGenerator;
552 }
553
554 /**
555 * Sets the logging level to use for log messages when retries have been exhausted.
556 */
557 public void setRetriesExhaustedLogLevel(LoggingLevel retriesExhaustedLogLevel) {
558 this.retriesExhaustedLogLevel = retriesExhaustedLogLevel;
559 }
560
561 public LoggingLevel getRetriesExhaustedLogLevel() {
562 return retriesExhaustedLogLevel;
563 }
564
565 /**
566 * Sets the logging level to use for log messages when retries are attempted.
567 */
568 public void setRetryAttemptedLogLevel(LoggingLevel retryAttemptedLogLevel) {
569 this.retryAttemptedLogLevel = retryAttemptedLogLevel;
570 }
571
572 public LoggingLevel getRetryAttemptedLogLevel() {
573 return retryAttemptedLogLevel;
574 }
575
576 public String getDelayPattern() {
577 return delayPattern;
578 }
579
580 /**
581 * Sets an optional delay pattern to use instead of fixed delay.
582 */
583 public void setDelayPattern(String delayPattern) {
584 this.delayPattern = delayPattern;
585 }
586
587 public boolean isLogStackTrace() {
588 return logStackTrace;
589 }
590
591 /**
592 * Sets whether stack traces should be logged or not
593 */
594 public void setLogStackTrace(boolean logStackTrace) {
595 this.logStackTrace = logStackTrace;
596 }
597
598 public boolean isLogRetryStackTrace() {
599 return logRetryStackTrace;
600 }
601
602 /**
603 * Sets whether stack traces should be logged or not
604 */
605 public void setLogRetryStackTrace(boolean logRetryStackTrace) {
606 this.logRetryStackTrace = logRetryStackTrace;
607 }
608
609 public boolean isLogHandled() {
610 return logHandled;
611 }
612
613 /**
614 * Sets whether errors should be logged even if its handled
615 */
616 public void setLogHandled(boolean logHandled) {
617 this.logHandled = logHandled;
618 }
619
620 public boolean isLogContinued() {
621 return logContinued;
622 }
623
624 /**
625 * Sets whether errors should be logged even if its continued
626 */
627 public void setLogContinued(boolean logContinued) {
628 this.logContinued = logContinued;
629 }
630
631 public boolean isLogRetryAttempted() {
632 return logRetryAttempted;
633 }
634
635 /**
636 * Sets whether retry attempts should be logged or not
637 */
638 public void setLogRetryAttempted(boolean logRetryAttempted) {
639 this.logRetryAttempted = logRetryAttempted;
640 }
641
642 public boolean isLogExhausted() {
643 return logExhausted;
644 }
645
646 /**
647 * Sets whether exhausted exceptions should be logged or not
648 */
649 public void setLogExhausted(boolean logExhausted) {
650 this.logExhausted = logExhausted;
651 }
652
653 public boolean isLogExhaustedMessageHistory() {
654 return logExhaustedMessageHistory;
655 }
656
657 /**
658 * Sets whether exhausted exceptions should be logged with message history included.
659 */
660 public void setLogExhaustedMessageHistory(boolean logExhaustedMessageHistory) {
661 this.logExhaustedMessageHistory = logExhaustedMessageHistory;
662 }
663
664 public boolean isAsyncDelayedRedelivery() {
665 return asyncDelayedRedelivery;
666 }
667
668 /**
669 * Sets whether asynchronous delayed redelivery is allowed.
670 * <p/>
671 * This is disabled by default.
672 * <p/>
673 * When enabled it allows Camel to schedule a future task for delayed
674 * redelivery which prevents current thread from blocking while waiting.
675 * <p/>
676 * Exchange which is transacted will however always use synchronous delayed redelivery
677 * because the transaction must execute in the same thread context.
678 *
679 * @param asyncDelayedRedelivery whether asynchronous delayed redelivery is allowed
680 */
681 public void setAsyncDelayedRedelivery(boolean asyncDelayedRedelivery) {
682 this.asyncDelayedRedelivery = asyncDelayedRedelivery;
683 }
684
685 public boolean isAllowRedeliveryWhileStopping() {
686 return allowRedeliveryWhileStopping;
687 }
688
689 /**
690 * Controls whether to allow redelivery while stopping/shutting down a route that uses error handling.
691 *
692 * @param allowRedeliveryWhileStopping <tt>true</tt> to allow redelivery, <tt>false</tt> to reject redeliveries
693 */
694 public void setAllowRedeliveryWhileStopping(boolean allowRedeliveryWhileStopping) {
695 this.allowRedeliveryWhileStopping = allowRedeliveryWhileStopping;
696 }
697
698 }