Core Concepts¶

This page covers the internal architecture and key mechanisms of IgnisMQ. Understanding these concepts is essential for operating IgnisMQ effectively in production.

Message Lifecycle¶

Every message in IgnisMQ follows a deterministic path from publication to deletion. The core abstraction is the Magazine — a sharded, persistent message store backed by your chosen datastore (Aerospike, etc.).

stateDiagram-v2
    [*] --> Published : magazine.load()
    Published --> Fired : magazine.fire()
    Fired --> Consumed : handler returns true
    Fired --> Sidelined : handler returns false / exception
    Consumed --> Deleted : magazine.deleteAll()
    Sidelined --> Deleted : magazine.delete() from main
    Sidelined --> Shoveled : ShovelTask
    Shoveled --> Published : reloaded to main magazine
    Fired --> Swept : stuck message detected
    Swept --> Sidelined : reloaded to sideline
    Deleted --> [*]

Step-by-step¶

1. Publish — queue.publish(message) serializes the message to JSON and appends it to a shard at the load pointer.
2. Fire — a consumer claims the next slot by advancing that shard's fire pointer and reading it. The message is now in flight, and no other consumer will see it.
3. Consume — your MessageHandler processes it, on the handler pool, under a timeout.
4. Delete, or sideline then delete — on success the whole accepted batch is retired in one magazine.deleteAll() call. On failure the record is copied to the sideline magazine first, one message at a time, and only deleted once the sideline has accepted it.

// Publishing. This is the only one of the four you call yourself.
IQueue<MyEvent> queue = manager.getQueue("orders");
queue.publish(new MyEvent("order-123", "PLACED"));

Sidelining¶

Sidelining is IgnisMQ's mechanism for isolating messages that could not be processed successfully. Rather than retrying in-place (which risks head-of-line blocking), failed messages are moved to a dedicated sideline magazine named {queueName}_SIDELINE.

flowchart TD
    Fire["magazine.fire()"] --> Deser["Deserialize JSON → M"]
    Deser --> Call["handler.handle(message)"]
    Call -->|returns true| Delete["magazine.deleteAll()<br/>the whole accepted batch, one call"]
    Call -->|returns false| Sideline["sidelineMagazine.load()"]
    Sideline --> DeleteOne["magazine.delete()<br/>per message"]
    Call -->|throws exception| Check{"Is exception<br/>ignorable?"}
    Check -->|Yes| DeleteOne
    Check -->|No| Sideline2["sidelineMagazine.load()"]
    Sideline2 --> DeleteOne

Decision logic in MagazineConsumerTask¶

public class OrderHandler implements MessageHandler<OrderEvent> {

    @Override
    public boolean handle(OrderEvent message) {
        // return true  → message consumed successfully
        // return false → message sidelined for later retry
        return orderService.process(message);
    }

    @Override
    public boolean handle(List<OrderEvent> messages) {
        // Called only when the queue has a BatchingConfig.
        return orderService.processAll(messages);
    }

    @Override
    public Set<Class<?>> getIgnorableExceptions() {
        // These exceptions cause the message to be deleted, not sidelined
        return Set.of(
            IllegalArgumentException.class,
            MalformedOrderException.class
        );
    }
}

Shoveling¶

The ShovelTask moves messages from the sideline magazine back into the main magazine for reprocessing. This is the retry mechanism in IgnisMQ.

sequenceDiagram
    participant ST as ShovelTask
    participant SM as Sideline Magazine
    participant MM as Main Magazine

    loop Every shovelInterval
        ST->>SM: fire()
        SM-->>ST: sidelined message
        ST->>MM: load(message)
        alt Load succeeds
            ST->>SM: delete(magazineData)
        else Load fails
            ST->>SM: reload(message) at the sideline tail
            ST->>SM: delete(magazineData)
            Note over ST,SM: Message stays in the sideline, but moves to the tail
        end
    end

Two modes of operation¶

Scheduled mode (default)

• Runs on a periodic timer controlled by ShovelConfig.timeIntervalInSecs.
• Default interval: 600 seconds (10 minutes).
• Keeps running indefinitely.

One-shot mode (autoDelete=true)

• Executes a single shovel pass and then simply does not repeat; nothing cancels it. Its slot against the shovel cap is released once it has finished.
• If an error occurs during the pass, it self-reschedules with a 10-second delay before retrying.
• Useful for on-demand sideline draining.

Configuration¶

ShovelConfig config = ShovelConfig.builder()
    .concurrency(4)            // default: 4 parallel shovel workers
    .timeIntervalInSecs(600)   // default: 600 (10 minutes)
    .build();

Sweeping¶

The Sweeper is a safety net that catches messages stuck in the "fired but never consumed" state — typically due to consumer crashes, network partitions, or long GC pauses.

flowchart TD
    Start["Sweeper task runs<br/>(every 15 min, control pool)"] --> Active{"Am I the<br/>assigned instance?"}
    Active -->|No| Skip["Skip sweep"]
    Active -->|Yes| Iterate["For each active queue + shard"]
    Iterate --> Scan["Per shard: watermark =<br/>firePointerBefore(now - sweepDuration)"]
    Scan --> Found{"Messages<br/>found?"}
    Found -->|No| Next["Next shard"]
    Found -->|Yes| Batch["peek() a window of slots,<br/>batched across shards"]
    Batch --> Reload["Still present = abandoned:<br/>load to sideline, then delete"]
    Reload --> Update["Advance the per-shard<br/>sweep pointer"]
    Update --> More{"More messages?"}
    More -->|Yes| Batch
    More -->|No| Next
    Next --> Done{"More shards?"}
    Done -->|Yes| Iterate
    Done -->|No| End["Sleep until next tick"]

Key parameters¶

Sideline sweep logic¶

When sweeping the sideline magazine, the sweeper uses a more conservative threshold:

sweepThreshold = min(now - sweepDuration, now - 2 * shovelInterval)

This prevents the sweeper from re-sidelining messages that are currently being shoveled back to the main queue. The 2x multiplier provides a safety margin.

Leader Election¶

IgnisMQ uses Apache Curator's LeaderSelector backed by ZooKeeper to ensure that cluster-wide singleton tasks (like sweeping) run on exactly one instance.

sequenceDiagram
    participant I1 as Instance 1
    participant I2 as Instance 2
    participant ZK as ZooKeeper
    participant Comm as Communicator Nodes

    I1->>ZK: Register ephemeral node<br/>/{clientId}-ignis-workers/{clientId}/loadbalancer/{balancerId}
    I2->>ZK: Register ephemeral node<br/>/{clientId}-ignis-workers/{clientId}/loadbalancer/{balancerId}

    Note over ZK: LeaderSelector elects I1

    I1->>I1: takeLeadership()
    I1->>ZK: Read member list
    I1->>I1: Round-robin partitions across members

    I1->>Comm: Write assignments to<br/>the partition's ZK node
    Note over I1,Comm: Rewritten only when membership<br/>changes, not on a timer

    loop Every 30 seconds
        I1->>I1: Poll own assignment
    end

    I1-->>I1: Check communicator node
    Note over I1: balancerId matches → activate sweeper

    I2-->>I2: Check communicator node
    Note over I2: balancerId does NOT match → sweeper inactive

    Note over ZK: I1 disconnects

    ZK->>ZK: Ephemeral node removed
    ZK->>I2: Leadership notification
    I2->>I2: takeLeadership()
    I2->>Comm: Assign each partition to a member (round-robin)
    Comm-->>I2: Assignment watched by every instance
    Note over I2: The assigned instance activates its sweeper<br/>- it may or may not be the leader

ZooKeeper path structure¶

/{clientId}-ignis-workers/
  └── {clientId}/
      ├── loadbalancer-leader    ← LeaderSelector election node
      ├── loadbalancer/
      │   ├── {balancerId}       ← ephemeral member node, one per instance
      │   └── {balancerId}
      └── readers/
          └── 1                  ← the sweep partition's assignment.
                                   One node: the sweeper is the only
                                   load-balanced worker today.

How activation works¶

1. The leader reads all registered member nodes under loadbalancer/.
2. It round-robins partitions across members.
3. It writes each assignment (including a balancerId) to that partition's node under readers/. The nodes are keyed by partition, not by instance. There is exactly one partition today, readers/1, because the sweeper is the only load-balanced worker.
4. Each instance polls those nodes every 30 seconds — they are read, not watched. If the balancerId in the assignment matches the instance's own ID, it sets the active AtomicBoolean to true, enabling the sweeper. Activation can therefore lag a reassignment by up to 30 seconds.
5. Only one instance has active = true at any time.

Queue Refresh Cycle¶

IgnisMQ keeps its in-memory queue state synchronized with the database through a background watcher.

What refreshQueues() does¶

1. Discovery — Queries the datastore for all queue definitions. Any queue found in the DB but not in memory is created (consumers are started, shovel tasks are scheduled).
2. Expiration — Queues past their expiry time are deactivated.
3. Cleanup — Inactive queues are removed from in-memory data structures and their consumers/shovelers are cancelled.
4. Reconciliation — If a queue's concurrency or ShovelConfig has changed in the DB, the in-memory configuration is updated. Consumers are scaled up or down to match.

flowchart LR
    Timer["Queue watcher<br/>(control pool, every 5 min)"] --> Fetch["Fetch queues from DB"]
    Fetch --> New{"New queues?"}
    New -->|Yes| Create["Create consumers<br/>+ shovelers"]
    New -->|No| Expired{"Expired queues?"}
    Create --> Expired
    Expired -->|Yes| Deactivate["Deactivate + remove"]
    Expired -->|No| Changed{"Config changed?"}
    Deactivate --> Changed
    Changed -->|Yes| Reconcile["Scale consumers<br/>Update shovel config"]
    Changed -->|No| Done["Done"]
    Reconcile --> Done

Threading and Isolation Model¶

A manager owns three pools:

What bounds each thread¶

Three mechanisms, each closing a different hole:

• The consumer run budget (30 s). A consumer drains until the queue is empty or the budget expires, then hands its thread back and resumes on the next scheduled run. Without it, one backlogged queue would hold a worker thread indefinitely. A batch already in hand is always finished rather than abandoned, so the budget never costs a delivery.
• The handler timeout (handlerTimeoutInMins, default 10, max 30). The handler runs on the handler pool and the worker stops waiting when the timeout expires. It is clamped to at most half of sweepDuration, so a handler cannot still be running when the sweeper re-homes the record it is working on. On timeout the message is sidelined under reason=timeout.
• Handler-pool saturation. If no handler thread becomes available within a 1 s grace period, the batch is refused and sidelined under reason=saturated rather than being run inline. A process leaking handler threads therefore sheds work visibly instead of quietly losing the bound.

Scheduling parameters¶

Fixed delay, not fixed rate: the 1 second gap is measured after the task finishes, so a slow consumer never has runs queued up behind it.

Scaling consumers at runtime¶

manager.increaseConsumers("orders", 5);
manager.decreaseConsumers("orders", 3);

Scaling down goes through the scheduler rather than cancelling the future directly, so the pool is resized too — otherwise scaling a queue down and up repeatedly would ratchet the thread count up for demand that no longer exists.