Architecture¶

This page provides a deep dive into the internal architecture of IgnisMQ — its components, threading model, data flows, type hierarchy, and operational mechanics.

1. System Overview¶

IgnisMQ is structured in distinct layers, each with a clear responsibility.

graph TB
    subgraph Application Layer
        A[IgnisMQBundle<br/><i>Dropwizard Bundle</i>]
        B[IgnisMQManager<br/><i>Queue Registry</i>]
    end

    subgraph Queue Layer
        C1[MagazineQueue #1]
        C2[MagazineQueue #2]
        CN[MagazineQueue #N]
        subgraph "Each MagazineQueue"
            MQ_MAIN[Main Magazine]
            MQ_SIDE[Sideline Magazine]
        end
    end

    subgraph Consumer Layer
        D1[MagazineConsumerTask x concurrency]
        D2[ShovelTask x shovel concurrency]
    end

    subgraph Coordination Layer
        E1[TaskInitializer]
        E2[LeaderElector]
        E3[Sweeper]
        E4[ZooKeeper / Curator]
    end

    subgraph Storage Layer
        F1[AerospikeQueueService]
        F2[AerospikeStoreClient]
        F3[(Aerospike)]
    end

    A -->|creates| B
    B -->|holds| C1
    B -->|holds| C2
    B -->|holds| CN
    C1 --> D1
    C1 --> D2
    E1 -->|creates| E2
    E1 -->|creates| E3
    E2 -->|LeaderSelector| E4
    E2 -->|activates the assigned instance| E3
    B --> F1
    E3 --> F1
    F1 --> F2
    F2 --> F3

2. Component Relationships¶

classDiagram
    class IgnisMQBundle {
        <<abstract>>
        -IgnisMQManager ignisMQManager
        +run(config, environment)
        +getIgnisMQManager() IgnisMQManager
        #context(config) IgnisMQContext
    }

    class IgnisMQContext {
        +BaseStorage storage
        +String clientId
        +String farmId
        +CuratorFramework curatorFramework
        +IgnisMQSettings settings
        +ConsoleConfiguration console
    }

    class IgnisMQManager {
        -Map~String, IQueue~ ignisMQMap
        +createQueue(CreateQueueRequest)
        +getQueue(name) IQueue
    }

    class IQueue~M~ {
        <<sealed interface>>
        +publish(M message) boolean
        +getUnconsumedCount() long
        +getMetaData() QueueMetaData
        +shovel(int concurrency)
    }

    class MagazineQueue~M~ {
        -Magazine~String~ magazine
        -Magazine~String~ sidelineMagazine
        -MessageHandler~M~ messageHandler
        -List~ScheduledFuture~ consumers
        -List~ScheduledFuture~ sidelineConsumers
        -HandlerExecutor handlerExecutor
        -ObjectMapper mapper
        +publish(M message) boolean
        +createConsumers(int count)
        +stopConsumers(int count)
    }

    class Magazine~T~ {
        -BaseMagazineStorage~T~ baseMagazineStorage
        -String magazineIdentifier
        +load(T data) boolean
        +fire() MagazineData~T~
        +delete(MagazineData~T~ magazineData)
        +deleteAll(Collection~MagazineData~T~~ magazineData)
    }

    class MagazineConsumerTask~M~ {
        -Magazine~String~ magazine
        -Magazine~String~ sidelineMagazine
        -MessageHandler~M~ messageHandler
        -HandlerExecutor handlerExecutor
        +run()
    }

    class ShovelTask {
        -Magazine~String~ magazine
        -Magazine~String~ sidelineMagazine
        +run()
    }

    class TaskInitializer {
        -LeaderElector leaderElector
        -ScheduledFuture sweeperTask
        +start()
        +stop()
    }

    class LeaderElector {
        -LeaderSelector leaderSelector
        -CuratorFramework curatorFramework
        +takeLeadership()
        -updateState(boolean force)
    }

    class Sweeper {
        <<implements LoadBalancer>>
        -AtomicBoolean active
        -QueueSweeper queueSweeper
        +activate()
        +deactivate()
        +run()
    }

    class QueueStatGuage {
        <<Supplier~List~>>
        +get() List
    }

    class QueueService {
        <<sealed interface>>
        +exists(String name) boolean
        +get(String name) Optional~QueueEntity~
        +store(String name, QueueEntity entity, int ttl)
        +getQueues(boolean active) Map~String, QueueEntity~
        +updateSweepProgress(...)
    }

    class AerospikeQueueService {
        -IAerospikeClient client
        -String namespace
        -String setName
    }

    class AerospikeStoreClient {
        -IAerospikeClient client
        -AerospikeConfiguration config
    }

    IgnisMQBundle ..> IgnisMQContext : context(config) returns
    IgnisMQBundle --> IgnisMQManager : creates from the context
    IgnisMQManager o-- IQueue : ConcurrentHashMap
    IQueue <|.. MagazineQueue : permits
    MagazineQueue *-- Magazine : main
    MagazineQueue *-- Magazine : sideline
    MagazineQueue *-- MagazineConsumerTask : consumers
    MagazineQueue *-- ShovelTask : sidelineConsumers
    MagazineConsumerTask --> Magazine : reads from main
    MagazineConsumerTask --> Magazine : writes to sideline
    ShovelTask --> Magazine : reads from sideline
    ShovelTask --> Magazine : writes to main
    TaskInitializer --> LeaderElector : creates
    TaskInitializer --> Sweeper : creates
    LeaderElector ..> Sweeper : activates/deactivates
    QueueService <|.. AerospikeQueueService : permits
    AerospikeQueueService --> AerospikeStoreClient
    IgnisMQManager --> QueueStatGuage : reads
    IgnisMQBundle --> QueueStatGuage : exposes as CachedGauge

3. Threading Model¶

The three pools¶

Sharing one pool coupled control with data: enough slow consumers and the watcher silently stops refreshing queues while the sweeper stops recovering orphans.

Schedule summary¶

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

What bounds a thread¶

The split bounds the blast radius of a badly behaved handler; it does not make the worker pool immune to one. Three mechanisms do that:

4. Data Flow¶

4.1 Publish Flow¶

sequenceDiagram
    participant App as Application
    participant IQ as IQueue.publish()
    participant OM as ObjectMapper
    participant Mag as Main Magazine
    participant MS as BaseMagazineStorage
    participant AS as Aerospike

    App->>IQ: publish(message)
    IQ->>OM: writeValueAsString(message)
    OM-->>IQ: JSON string
    IQ->>Mag: load(json)
    Mag->>MS: write slot, advance pointer
    MS->>AS: put(key, bins)
    AS-->>MS: ok
    MS-->>Mag: ok
    Mag-->>IQ: true
    IQ-->>App: boolean

4.2 Consume Flow¶

sequenceDiagram
    participant T as Worker pool
    participant CT as MagazineConsumerTask
    participant M as Main Magazine
    participant MS as BaseMagazineStorage
    participant AS as Aerospike
    participant H as MessageHandler

    T->>CT: run()
    CT->>M: fire()
    M->>MS: claim next pointer, read slot
    MS->>AS: operate(meta) + get(data)
    AS-->>MS: Record
    MS-->>M: Record
    M-->>CT: MagazineData
    CT->>CT: deserialize(JSON → M)
    CT->>H: handle(message) or handle(batch), on the handler pool, with a timeout

    alt Handler returns true
        H-->>CT: true
        CT->>M: deleteAll(batch)
        M->>MS: deleteAll(slots)
        MS->>AS: one batched delete, or one delete per key<br/>when the batch holds a single record
    else Returns false, throws, times out, or the pool is saturated
        H-->>CT: failure
        CT->>CT: sidelineMagazine.load(payload)
        alt Sideline accepted it
            CT->>M: delete(record), one at a time
        else Sideline refused it
            CT->>CT: leave the record for the sweeper
        end
    end

4.3 Shovel Flow¶

sequenceDiagram
    participant T as Worker pool
    participant ST as ShovelTask
    participant SL as Sideline Magazine
    participant M as Main Magazine
    participant AS as Aerospike

    T->>ST: run()
    ST->>SL: fire()
    SL->>AS: query(sidelineSet)
    AS-->>SL: Record
    SL-->>ST: Record
    ST->>M: load(record)
    M->>AS: put(mainSet, record)
    AS-->>M: ok
    M-->>ST: ok
    ST->>SL: delete(id)
    SL->>AS: delete(sidelineSet, id)

4.4 Sweep Flow¶

sequenceDiagram
    participant Sched as Control pool
    participant SW as Sweeper
    participant SP as Sweep pool (fixed, 64)
    participant QS as QueueService
    participant M as Main Magazine
    participant SL as Sideline Magazine

    Sched->>SW: run()
    SW->>QS: getQueues(active=true)
    QS-->>SW: Map<String, QueueEntity>

    loop For each queue (parallel, capped at 64)
        SW->>SP: submit(sweepTask)
        SP->>M: firePointerBefore(now - sweepDuration)
        M-->>SP: per-shard watermark, or absent
        loop Until each shard reaches its watermark
            SP->>M: peek(pointers, batched across shards)
            M-->>SP: records still present = abandoned
            SP->>SL: load(payload)
            SP->>M: delete(record) only if the load succeeded
        end
        SP->>QS: updateSweepProgress(pointers, swept)
    end

5. Sealed Type Hierarchy¶

IgnisMQ uses Java 17 sealed types to enforce a closed set of implementations.

classDiagram
    class IQueue~M~ {
        <<sealed interface>>
        +publish(M message) boolean
        +getUnconsumedCount() long
        +getMetaData() QueueMetaData
        +shovel(int concurrency)
    }
    class MagazineQueue~M~ {
        <<final>>
    }
    IQueue <|.. MagazineQueue : permits

    class QueueService {
        <<sealed interface>>
        +exists(String name) boolean
        +get(String name) Optional~QueueEntity~
        +store(String name, QueueEntity entity, int ttl)
        +updateSweepProgress(...)
    }
    class AerospikeQueueService {
        <<final>>
    }
    QueueService <|.. AerospikeQueueService : permits

    class BaseStorage {
        <<abstract sealed>>
        +accept(StorageVisitor~T~ visitor) T
    }
    class AerospikeStorage {
        <<final>>
        -AerospikeConfiguration configuration
        -String namespace
    }
    BaseStorage <|-- AerospikeStorage : permits

    class StorageClient~T~ {
        <<interface>>
        +getClient() T
        +stop()
    }
    class AerospikeStoreClient {
        <<final>>
    }
    StorageClient <|.. AerospikeStoreClient : implements

6. Visitor Pattern¶

IgnisMQ uses the visitor pattern to decouple storage construction from storage type specifics.

classDiagram
    class BaseStorage {
        <<abstract sealed>>
        +accept(StorageVisitor~T~ visitor) T
    }

    class AerospikeStorage {
        -AerospikeConfiguration configuration
        -String namespace
        +accept(StorageVisitor~T~ visitor) T
    }

    class StorageVisitor~T~ {
        <<interface>>
        +visit(AerospikeStorage storage) T
    }

    class MagazineStorageVisitor {
        <<final, top-level>>
        +visit(AerospikeStorage storage) BaseMagazineStorage~String~
    }

    class QueueServiceVisitor {
        <<anonymous, inside IgnisMQManager>>
        +visit(AerospikeStorage storage) QueueService
    }

    class StoreClientVisitor {
        <<anonymous, inside IgnisMQManager>>
        +visit(AerospikeStorage storage) StorageClient
    }

    BaseStorage <|-- AerospikeStorage : permits
    StorageVisitor <|.. MagazineStorageVisitor
    StorageVisitor <|.. QueueServiceVisitor
    StorageVisitor <|.. StoreClientVisitor
    BaseStorage --> StorageVisitor : accept()
    AerospikeStorage ..|> BaseStorage

How it works:

1. IgnisMQManager receives a BaseStorage instance (which is actually AerospikeStorage).
2. To build a QueueService, it calls storage.accept(...) with an anonymous StorageVisitor.
3. AerospikeStorage.accept() dispatches to visitor.visit(this).
4. The visitor has access to the concrete AerospikeStorage fields and constructs AerospikeQueueService.

The Magazine instances themselves are built through MagazineStorageVisitor, which returns a BaseMagazineStorage<String> — the storage a Magazine is constructed on, not a Magazine.

7. Metrics¶

Core records metrics through Micrometer. ignismq-dw-bundle bridges them into the application's Dropwizard MetricRegistry and supplies Dropwizard-specific gauges and lifecycle adapters. ignismq-core has no Dropwizard dependency at any scope, and an enforcer rule fails the build if one appears.

Registered metrics¶

Full definitions, tags and semantics: Metrics. In outline:

Every dimension is a tag rather than part of the name, so a queue is a tag value and not a new meter.

Dropwizard has no notion of tags, so the bundle's bridge flattens each Micrometer tag into the metric name. magazine.fire.outcomes tagged magazine=main, outcome=delivered is published as magazine.fire.outcomes.magazine.main.outcome.delivered.

The queue-stat gauges¶

flowchart LR
    A[Metrics scrape] --> B[Queue depth gauges]
    B --> C{Snapshot fresh?}
    C -->|Yes| D[Serve from the cached snapshot]
    C -->|No| E[One metadata read per queue]
    E --> F[published, consumed, unconsumed,<br/>sidelined, shovelled]
    F --> D

Depth is the one value that costs a storage read, so all five gauges for a queue share a single refresh rather than each triggering their own. Without that, a scrape would cost one metadata read per queue per meter.

What happened to commands.*¶

Earlier versions wove @MonitoredFunction timers into most public methods with AspectJ. That was removed along with the Dropwizard coupling: it pulled AspectJ and a build-time weaving step into ignismq-core, and the resulting timers measured method entry and exit rather than anything operationally meaningful.

The two that were worth keeping — publish and consume latency — are now ignismq.publish and ignismq.consume, with the queue as a tag. The rest of the commands.* namespace is gone, and nothing is aliased: see the upgrade notes.

8. ZooKeeper Node Structure¶

IgnisMQ uses ZooKeeper (via Apache Curator) for leader election and work distribution among instances.

/{clientId}-ignis-workers/
  └── {clientId}/
      ├── loadbalancer-leader          ← LeaderSelector (persistent node)
      │                                  Only one instance holds leadership.
      │                                  The leader assigns the sweep
      │                                  partition; the assigned instance
      │                                  runs the Sweeper, leader or not.
      │
      ├── loadbalancer/
      │   └── {balancerId}             ← Ephemeral nodes, named by the
      │                                  instance's balancerId UUID.
      │                                  Not sequential: the UUID is
      │                                  already the identity.
      │
      └── readers/
          └── {partition}              ← Communicator nodes.
                                         Each partition stores the
                                         assigned balancerId, mapping
                                         partitions to instances.

graph TD
    ROOT["/{clientId}-ignis-workers"] --> CLIENT["/{clientId}"]
    CLIENT --> LEADER["loadbalancer-leader<br/><i>LeaderSelector</i>"]
    CLIENT --> LB["loadbalancer/"]
    CLIENT --> READERS["readers/"]

    LB --> LB1["{balancerId-001}<br/><i>ephemeral</i>"]
    LB --> LB2["{balancerId-002}<br/><i>ephemeral</i>"]
    LB --> LBN["{balancerId-N}<br/><i>ephemeral</i>"]

    READERS --> R1["1<br/>assigned: balancerId-002"]

    style LEADER fill:#f9a825,stroke:#f57f17
    style LB1 fill:#81d4fa,stroke:#0277bd
    style LB2 fill:#81d4fa,stroke:#0277bd
    style LBN fill:#81d4fa,stroke:#0277bd

Leader election flow:

1. Each instance creates an ephemeral node under loadbalancer/.
2. Curator's LeaderSelector competes for loadbalancer-leader.
3. The winning instance calls takeLeadership(), which forces an immediate reassignment pass. Leadership is what makes an instance the assigner; it is not what activates a sweeper.
4. The leader reassigns partitions across live balancer nodes by round-robin, writing each assignment into that partition's node under readers/. It rewrites them only when membership changes, or on the forced pass taken at step 3.
5. Every instance polls its assignment every 30 seconds - these nodes are read, not watched - and the instance whose balancerId matches activates its Sweeper. That may or may not be the leader, and activation can lag a reassignment by up to 30 seconds.
6. When the leader dies, its ephemeral node disappears and Curator triggers a new election.