Package org.acemq.amqp.core

Class Replay

java.lang.Object
org.acemq.amqp.core.Replay
public final class Replay extends Object
Moves messages out of a dead-letter queue or parking lot and back to the queue they failed in.

The half of dead-lettering that is usually missing. Capturing a failed message is easy and most libraries stop there, which leaves an operator with a queue full of evidence and no way to act on it — and a dead-letter queue nobody can drain is a slower way of losing data, not an alternative to it. The messages are already safe; this is how they get another chance once the bug is fixed or the downstream service is back. 


 Replay replay = mq.replay("orders.new");

 replay.pending();              // 412 waiting in orders.new.dlq — look before touching
 replay.replay(50);             // move the first 50 back
 replay.replayAll();            // or all of them

 replay.parked().replayAll();   // the ones that could not even be decoded

What it does to a message

The body is untouched. The attempt counter is reset, so a replayed message gets the full retry ladder again rather than being dead-lettered immediately by the count that put it here: a message dead-lettered on the last attempt of a five-attempt policy would otherwise be dead-lettered again before any handler saw it, and the operator who had just fixed the bug would have moved two thousand messages from one queue to the same queue. keepingAttempts() turns that off for the cases that want exactly what was there. Three headers are added: where it came from, when it was replayed, and how many times it has been replayed before — the last one because a message on its fifth trip through a dead-letter queue is telling you something that a fresh-looking message would not.

Messages go back to the queue they failed in, not through the exchange that originally routed them. Republishing through the exchange would deliver to every queue bound to it, so replaying one failure in one consumer would hand duplicate work to every other consumer that had already succeeded.

What it does not do

Replay is at-least-once, like everything else here. Each message is published to the source queue and only then acknowledged in the dead-letter queue; if the process dies between the two, that message is replayed again on the next run. The alternative — acknowledging first — loses it, which is the wrong way round for a tool whose entire job is not losing things.

It does not fix anything. Replaying into a consumer that still has the bug simply refills the dead-letter queue, more slowly than the first time and with the replay count climbing.

  • Method Details

    • parked

      public Replay parked()
      Switches to the parking lot: messages that could not be decoded at all.

      Separate from the dead-letter queue because the two need different fixes. A dead-lettered message failed in a handler and may well succeed on a retry; a parked one could not be turned into an object, so it will fail identically every time until the code that reads it changes. Replaying the parking lot before deploying that change is pure churn.

      Returns:
      a replay reading the parking lot instead

    • keepingAttempts

      public Replay keepingAttempts()
      Returns a replay that leaves each message's attempt counter exactly as it was.

      Off by default, and the default is the one almost everybody wants: a replayed message goes back on attempt one, because a message dead-lettered on its last attempt would otherwise be dead-lettered again before any handler saw it.

      Turn it on to put back exactly what was there — for an audit, or for a queue read by something that counts attempts itself and would be misled by a counter this tool rewrote.

      Returns:
      a replay that preserves the attempt counter

    • pending

      public long pending()
      Counts what is waiting, without moving anything.

      Worth calling first. Replaying forty thousand messages into a consumer that is already behind is a decision, and this is the number that decision needs.

      Returns:
      how many messages are in the queue being replayed

    • from

      public String from()
      Returns:
      the queue messages are being taken from

    • to

      public String to()
      Returns:
      the queue messages are being returned to

    • replayAll

      public int replayAll()
      Moves every waiting message back, stopping when the queue is empty.

      "Empty" means empty at the moment it is asked, so a queue still being filled will not keep this running forever: what arrives after the drain reaches the end is left for the next run.

      Returns:
      how many messages were moved

    • replay

      public int replay(int max)
      Moves up to a given number of messages back.
      Parameters:
      max - most messages to move; the rest are left where they are
      Returns:
      how many were actually moved, which is fewer than max when the queue ran out

    • replay

      public int replay(int max, Predicate<InboundDelivery> filter)
      Moves messages matching a filter, leaving the rest in place.

      For the common case of one bad deployment among many failures: replay what failed with a particular error, or what came from a particular origin, and leave the genuinely broken messages to be looked at by a person.

      A message the filter rejects is requeued, and the drain stops there. That is deliberate and is the honest behaviour available: AMQP has no way to look past a message without taking it, so skipping one means holding it while reading the next, and a drain that holds every rejected message would run the queue's worth of unsettled deliveries into memory. Stopping means a filter that rejects the first message moves nothing, which is visible, rather than quietly reordering the queue, which is not.

      Parameters:
      max - most messages to move
      filter - decides whether a message should be moved
      Returns:
      how many were moved

    • toString

      public String toString()
      Overrides:
      toString in class Object