Package org.acemq.amqp.patterns

Class JdbcIdempotencyStore

java.lang.Object
org.acemq.amqp.patterns.JdbcIdempotencyStore
All Implemented Interfaces:
IdempotencyStore
public final class JdbcIdempotencyStore extends Object implements IdempotencyStore
An idempotency store in a database table, shared by every consumer that points at it.

The one InMemoryIdempotencyStore cannot replace. An in-process store deduplicates within one JVM, which is enough when duplicates arrive seconds apart on the same instance and useless the moment there are three instances behind one queue: the redelivery lands on a different machine from the original, finds an empty map, and does the work again. Charging a card twice is not a caching problem.

The failure this has and the in-memory one does not

A shared store outlives the process using it, and that cuts both ways. If a consumer claims a message and then dies mid-handler, the claim stays in the table. With a naive implementation every redelivery of that message — forever — is discarded as a duplicate, and a crash that should have cost one retry has silently deleted a message instead. The in-memory store never had this problem only because a crash wiped it.

So a claim here is a lease, not a lock. It expires after claimTimeout, after which another consumer may take it over. That makes the timeout a real decision: too short and two consumers process the same message concurrently while the first is still working; too long and a crashed consumer stalls that message for the duration. It should comfortably exceed the slowest handler, and the default is deliberately generous.

Confirmations are kept for retention and then forgotten, because a table that remembers every identifier ever seen is a disk-space incident waiting to happen. A duplicate arriving after retention is handled again — retention is the window within which duplicates are actually expected to arrive, not a permanent record.

What it does not do

This deduplicates the delivery, not the work. If the handler writes to a different database from this table, a crash between the write committing and confirm(java.lang.String) landing leaves the work done and unrecorded, and the redelivery repeats it. Putting this table in the same database as the handler's writes, in the handler's transaction, is what closes that gap — and doing so is the caller's decision, since only the caller knows what the handler touches. 


 JdbcIdempotencyStore store = new JdbcIdempotencyStore(dataSource);
 store.createSchemaIfAbsent();          // development and tests only

 mq.consume("orders.new", OrderPlaced.class,
         ConsumerOptions.prefetch(20).idempotent(store),
         message -> payments.charge(message.payload()));

Schedule purgeExpired() — nothing here deletes rows on its own, because a store that deletes on the hot path pays for it on every single message.

  • Constructor Details

    • JdbcIdempotencyStore

      public JdbcIdempotencyStore(DataSource dataSource)
      Parameters:
      dataSource - where the table lives

    • JdbcIdempotencyStore

      public JdbcIdempotencyStore(DataSource dataSource, Duration retention, Duration claimTimeout, String table)
      Parameters:
      dataSource - where the table lives
      retention - how long a confirmed identifier is remembered; duplicates arriving later than this are handled again
      claimTimeout - how long one consumer may hold a message before another may take it over. Must exceed the slowest handler, or two consumers will process the same message at once
      table - table name; must be a plain SQL identifier

  • Method Details

    • createSchemaIfAbsent

      public void createSchemaIfAbsent()
      Creates the table if it is not already there.

      For development and tests. In production the table belongs in the migration tool that owns the rest of the schema.

      Throws:
      AceMqException - if the schema cannot be created

    • claim

      public boolean claim(String messageId)
      Description copied from interface: IdempotencyStore
      Takes ownership of a message identifier, if nobody else has it.
      Specified by:
      claim in interface IdempotencyStore
      Parameters:
      messageId - the identifier, normally Envelope.id()
      Returns:
      true when the caller now owns this identifier and should do the work; false when it is already confirmed or claimed elsewhere. Which of the two it is matters — only a confirmed one is a duplicate — so consumers call IdempotencyStore.tryClaim(java.lang.String)

    • confirm

      public void confirm(String messageId)
      Description copied from interface: IdempotencyStore
      Records that the work for a claimed identifier completed.

      After this, IdempotencyStore.claim(java.lang.String) returns false for the identifier until the store forgets it, which is what stops a redelivery from repeating the work.

      Specified by:
      confirm in interface IdempotencyStore
      Parameters:
      messageId - the identifier being confirmed

    • release

      public void release(String messageId)
      Description copied from interface: IdempotencyStore
      Gives up a claim without recording completion, so the message can be tried again.

      Called when a handler fails. Without it a failed attempt would poison the identifier and the retry would be discarded as a duplicate, which turns a transient failure into permanent message loss.

      Specified by:
      release in interface IdempotencyStore
      Parameters:
      messageId - the identifier being released

    • isConfirmed

      public boolean isConfirmed(String messageId)
      Specified by:
      isConfirmed in interface IdempotencyStore
      Parameters:
      messageId - the identifier to test
      Returns:
      whether the work for this identifier is already known to have completed

    • purgeExpired

      public int purgeExpired()
      Deletes rows nobody is bound by any more: confirmations past their retention, and claims whose lease has run out.

      Schedule this — hourly is ample. Nothing on the message path deletes anything, because a store that tidies up on the hot path makes every message pay for it.

      Returns:
      how many rows were removed

    • size

      public long size()
      Returns:
      how many identifiers the table currently holds, expired ones included

    • toString

      public String toString()
      Overrides:
      toString in class Object