Aerospike Backend¶

Comprehensive documentation of the Aerospike storage backend for IgnisMQ.

Overview¶

IgnisMQ uses Aerospike as its primary storage backend, leveraging the Magazine library for queue primitives (load/fire operations) and AerospikeQueueService for queue metadata management.

The storage layer is split into two concerns:

• AerospikeStorage — a thin configuration holder that carries the AerospikeConfiguration and namespace, implementing the BaseStorage visitor pattern.
• AerospikeQueueService — the actual CRUD and sweep logic for queue metadata, fire timestamps, and sweep operations.

Configuration¶

AerospikeStorage storage = new AerospikeStorage(
    AerospikeConfiguration.builder()
        .hosts(List.of(AerospikeHost.builder()
            .host("localhost")
            .port(3000)
            .build()))
        .retries(3)
        .sleepBetweenRetries(100)
        .socketTimeout(3000)
        .totalTimeout(5000)
        .maxConnectionsPerNode(100)
        .threadPoolSize(4)   // default: availableProcessors * 4
        .build(),
    "my-namespace"
);

Queue Metadata Set: {farmId}_{clientId}_ignis_queues¶

Each record represents one queue. The Aerospike key is the queue name.

Magazine Data Sets (managed by Magazine library)¶

Secondary Indexes¶

A secondary index is automatically created on the active bin (IndexType.STRING) during AerospikeQueueService construction:

// Index name: {farmId}_{clientId}_ignis_queues_active
createIndex(setName + "_" + ACTIVE_BIN, ACTIVE_BIN, IndexType.STRING);

This enables efficient queries for active or inactive queues via Filter.equal(ACTIVE_BIN, "true").

Client Configuration¶

The Aerospike client is configured in AerospikeStoreClient with the following policies:

Retry Strategy¶

All AerospikeQueueService operations use a Guava Retryer:

RetryerBuilder.newBuilder()
    .retryIfExceptionOfType(AerospikeException.class)
    .withStopStrategy(StopStrategies.stopAfterAttempt(configuration.getRetries()))
    .withWaitStrategy(WaitStrategies.fixedWait(
        configuration.getSleepBetweenRetries(), TimeUnit.MILLISECONDS))
    .withBlockStrategy(BlockStrategies.threadSleepStrategy())
    .build();

Fire Timestamp Mechanism¶

When a consumer fires (dequeues) a message, addFireTimestamp() stamps the fireTS bin on the corresponding data record with the current epoch millisecond timestamp:

queueService.addFireTimestamp(magazineData, System.currentTimeMillis());

This timestamp is later used by the sweeper to identify "stuck" messages — those that were fired but never acknowledged within the configured sweepDuration.

Sweep Internals¶

The sweep process detects messages that were fired but not consumed within the allowed time window and reloads them into the sideline magazine.

Execution Flow¶

1. Sweeper.run() — Triggered as a TimerTask (only when this node is the active leader).
2. Fetches all active queues via queueService.getQueues(true).
3. Submits each queue to a fixed thread pool (Executors.newFixedThreadPool(64)).
4. Per queue: iterates over each shard 0..numShards-1 sequentially.
5. Per shard: calls queueService.sweep(queueName, shard, sweepTillFireTS, sidelineMagazine).

Sweep Algorithm (per shard)¶

flowchart TD
    A[Start sweep for shard] --> B[Read Magazine meta record]
    B --> C{Meta record exists?}
    C -- No --> Z[Skip — no data]
    C -- Yes --> D[Get currentFirePointer from meta]
    D --> E[Get sweepPointer from queue entity]
    E --> F{sweepPointer >= currentFirePointer?}
    F -- Yes --> Z2[Skip — already swept]
    F -- No --> G[Batch read 1000 records starting at sweepPointer]
    G --> H{For each record}
    H --> I{fireTS >= sweepTillFireTS?}
    I -- Yes --> J[Stop — reached time boundary]
    I -- No --> K[Load message into sideline magazine]
    K --> L[Delete record from main magazine]
    L --> H
    H -- All processed --> M[Update sweepPointer and sweptCounter via MapOperation.increment]
    M --> N{sweepPointer >= currentFirePointer?}
    N -- Yes --> O[Done]
    N -- No --> G
    J --> O

Key Details¶

• sweepTillFireTimestamp = System.currentTimeMillis() - sweepDuration
• Batch size: 1000 records per batch (constant SWEEP_BATCH_SIZE)
• Sideline sweep threshold: min(sweepTillFireTS, now - 2 * shovelTimeInterval) — prevents sweeping messages that are about to be shoveled
• Pointer updates: Uses MapOperation.increment to atomically advance per-shard sweep pointers
• Queue metadata TTL: queueExpiry * 2 (the TTL_FACTOR_FOR_QUEUE_EXPIRY constant)

Performance Tips¶