Package org.acemq.amqp.api

Class RetryPolicy

java.lang.Object
org.acemq.amqp.api.RetryPolicy
public final class RetryPolicy extends Object
When, and how often, a failed message should be tried again, and where the waiting happens.

A policy is a schedule, not a mechanism. It says a fourth attempt should happen roughly eight seconds after the third; it does not say how the waiting happens. That separation matters, because a long wait must never be a Thread.sleep inside a handler: a sleeping handler holds an unacknowledged delivery and a prefetch slot, so a handful of slow retries can stop a consumer dead while the queue behind it grows — and a consumer that restarts halfway through a five-minute backoff loses the whole wait, because the broker redelivers the unacknowledged message at once and a five-minute policy delivers in none.

So the waiting is split at a threshold. A wait shorter than brokerWaitThreshold() is spent in the consumer, where the seconds a restart loses are only seconds and a held prefetch slot is cheap. A wait at or above it is spent in the broker, in a {queue}.retry.{delay} queue whose x-message-ttl is the wait and whose dead-letter target is the source queue. Splitting rather than picking one of the two takes the durability where it is worth its complexity and leaves the simplicity where it is not: a schedule that runs in a few seconds costs the broker no queues at all.

Jitter follows the same line, and has to. A rung queue's time-to-live is fixed when the queue is declared, so a jittered delay would name a queue that does not exist; and above the threshold the spread comes free anyway, because each message's time-to-live starts when it enters the rung, so a fleet that failed over ten seconds is released over ten seconds. Jitter therefore applies to consumer waits only.

Two independent limits apply, and either one ends the retries:

  • maxAttempts() — how many deliveries in total, including the first;
  • maxMessageAge() — how old the message may get, measured from its first publish rather than from the most recent failure, or zero for no limit at all.

The age limit is what stops a message from circulating for days after an outage: five attempts with an hour between them is six hours of retrying that almost nobody intends. It is off unless giveUpAfter(Duration) turns it on, because a limit nobody asked for is a policy nobody asked for: maxAttempts() already bounds the retrying, and a library that also invented an age at which to stop would be making a decision on the caller's behalf and dead-lettering messages for it. Zero means never, in this library and in the Go, .NET, Python and Ruby ones, so the same policy abandons the same message in all five.

  • Field Details

    • DEFAULT_BROKER_WAIT_THRESHOLD

      public static final Duration DEFAULT_BROKER_WAIT_THRESHOLD
      Waits this long or longer are spent in the broker rather than in the consumer.

      Thirty seconds is roughly where the two costs cross. Below it, the seconds a restart loses are only seconds and holding one prefetch slot is cheaper than asking an operator's broker for another queue. Above it, a consumer that restarts mid-wait loses the wait entirely, which is a correctness bug rather than a throughput one.

      The number is part of the cross-language contract: the Go, .NET, Python and Ruby libraries default to the same thirty seconds, so the same policy needs the same rungs whichever of them declares the topology.

  • Method Details

    • none

      public static RetryPolicy none()
      A policy that never retries: one attempt, then the dead-letter queue.
      Returns:
      the no-retry policy

    • fixed

      public static RetryPolicy fixed(int maxAttempts, Duration delay)
      Retries a fixed number of times with the same delay between each, with no jitter.

      No jitter is the point of asking for a fixed policy: a caller who says "every thirty seconds" has said something exact, and a library that quietly returned twenty-six would be answering a question nobody asked. Add it back with withJitter(double) when a fleet failing in lockstep is the worry.

      Parameters:
      maxAttempts - total deliveries including the first; must be at least 1
      delay - wait between attempts
      Returns:
      a fixed-delay policy

    • exponential

      public static RetryPolicy exponential(int maxAttempts, Duration initialDelay, Duration maxDelay)
      Retries with a doubling delay, capped.

      For example exponential(5, ofSeconds(1), ofMinutes(5)) waits one second, then two, four and eight, before giving up after the fifth delivery.

      Doubling and twenty percent jitter are the cross-language default rather than a taste: the same policy has to produce the same four numbers in Go, .NET, Python and Ruby, because the same message can be retried by a consumer written in any of them and a message that waited one second under one library and five under another has no schedule at all.

      Parameters:
      maxAttempts - total deliveries including the first; must be at least 1
      initialDelay - wait before the second attempt
      maxDelay - ceiling for any single wait
      Returns:
      an exponential policy with a multiplier of two

    • exponential

      public static RetryPolicy exponential(int maxAttempts, Duration initialDelay, double multiplier, Duration maxDelay)
      Retries with an exponentially growing delay and an explicit multiplier.
      Parameters:
      maxAttempts - total deliveries including the first; must be at least 1
      initialDelay - wait before the second attempt
      multiplier - growth factor; must be at least 1
      maxDelay - ceiling for any single wait
      Returns:
      an exponential policy

    • giveUpAfter

      public RetryPolicy giveUpAfter(Duration maxMessageAge)
      Returns a copy that abandons messages older than the given age.

      This is the only way to get an age limit; no factory sets one. The limit is reached rather than passed, so a message whose age is exactly the limit is dead-lettered — the same boundary the other four libraries keep.

      Parameters:
      maxMessageAge - age measured from first publish, or zero for no limit
      Returns:
      a policy with the age limit applied

    • withJitter

      public RetryPolicy withJitter(double jitterFactor)
      Returns a copy with a different amount of randomness applied to each delay.

      Only to a delay waited in the consumer. A delay handed to a rung queue is left exactly as the schedule produced it, because the queue's time-to-live is fixed at declaration and a jittered delay would name a queue nobody declared.

      Parameters:
      jitterFactor - fraction of the delay to vary by, between 0 and 1
      Returns:
      a policy with that jitter

    • waitInBrokerFrom

      public RetryPolicy waitInBrokerFrom(Duration threshold)
      Returns a copy that moves the line between waiting here and waiting there.

      Zero is the way out: with no threshold nothing is long enough to reach the broker, so every wait is spent in the consumer and no rung queue is ever declared. That is the right setting for a service that may not create queues on the broker it consumes from, and the wrong one for a policy whose delays are measured in minutes.

      Parameters:
      threshold - waits this long or longer go to a rung queue, or zero for none of them
      Returns:
      a policy with that threshold

    • maxAttempts

      public int maxAttempts()
      Returns:
      total deliveries allowed, including the first

    • maxMessageAge

      public Duration maxMessageAge()
      Returns:
      how old a message may get before it is abandoned; zero means never

    • jitterFactor

      public double jitterFactor()
      Returns:
      the fraction of each delay that is randomised

    • brokerWaitThreshold

      public Duration brokerWaitThreshold()
      Returns:
      the wait at which the broker takes over from the consumer; zero means never

    • waitsInBroker

      public boolean waitsInBroker(Duration delay)
      Whether a wait of this length belongs on a rung queue rather than in the consumer.
      Parameters:
      delay - an unjittered delay, as schedule() reports them
      Returns:
      whether the broker rather than the consumer should hold it

    • brokerRungs

      public List<Duration> brokerRungs()
      The delays this policy needs a rung queue for, in the order the schedule reaches them.

      Exactly the entries of schedule() that are at or above the threshold, with repeats removed — a fixed policy that waits a minute three times needs one queue, not three. It is a finite list because the schedule is, which is what lets the rungs be declared with the rest of the topology instead of being conjured by a consumer at the moment it first fails.

      Returns:
      one delay per rung queue; empty when every wait is spent in the consumer

    • schedule

      public List<Duration> schedule()
      The delays this policy uses, without jitter.

      The engine reads this to work out which retry queues to create, which is why it is exposed rather than kept private: the topology is derived from the policy instead of being configured separately and drifting away from it.

      Returns:
      one delay per retry, in order; empty when the policy never retries

    • nextDelay

      public Optional<Duration> nextDelay(int attempt, Duration messageAge)
      How long to wait before the next attempt.

      The delay only. nextWait(int, Duration) is the whole answer, and is what a consumer needs: a delay that will be spent in the broker is reported here exactly as the schedule produced it, because that is the delay that names a rung queue, while a delay that will be spent in the consumer has already had jitter applied to it.

      Parameters:
      attempt - the attempt that just failed, starting at 1
      messageAge - how long ago the message was first published
      Returns:
      the delay before the next attempt, or empty when the message should be dead-lettered because the attempts or the age limit are exhausted

    • nextWait

      public Optional<RetryPolicy.Wait> nextWait(int attempt, Duration messageAge)
      How long to wait before the next attempt, and where the message spends it.
      Parameters:
      attempt - the attempt that just failed, starting at 1
      messageAge - how long ago the message was first published, or null when it is not known, which counts as young enough
      Returns:
      the wait, or empty when the message should be dead-lettered because the attempts or the age limit are exhausted

    • toString

      public String toString()
      Overrides:
      toString in class Object