High throughput SQS consumer using Spring Boot

The CEO wanted a dashboard showing overall inventory status: how it was moving, where it was, and how much was unaccounted for.

We didn't have that data, so we built it. Every inventory flow across B2B, B2C, and retail store applications was made to emit an event whenever a SKU item's status changed.

A consumer ingested those events, and a scheduled Glue job loaded them into Redshift, where the Data Analytics team built reports and dashboards.

The consumer's job, in order:

1. Read from SQS.
2. Deserialize the payload.
3. Parse and extract the fields that mattered.
4. Project and normalize the data into a stable document shape.
5. Write to MongoDB.

Steps 1 through 4 were ordinary functional requiremnts. Step 5, at the volume this pipeline saw, complicated things.

Naive approach and why it didn't work

The obvious way to avoid duplicate documents is to check before you write: look up the SKU Item ID in MongoDB, compare it to the incoming event, insert only if it's new.

That approach has a real cost. Every insert now costs two Mongo round trips instead of one. In a pipeline processing hundreds of thousands of messages per batch window, that extra read is blocking, it adds latency to every single record, and it puts sustained load on MongoDB's CPU for a check that succeeds the overwhelming majority of the time, since most records are genuinely new.

The obvious way to handle spikes is a single-threaded receive-and-process loop. Ack semantics stay simple: one message in flight, one ack out. It also can't keep up. Spikes of hundreds of thousands of messages arrived during batch windows, then the queue dropped to a trickle for the rest of the day. A design tuned for quiet midday load fails the moment a large inventory run dumps that volume in minutes.

The obvious fix for that is to throw an unbounded thread pool at it. That fails differently: bursts pile up in memory, the heap grows, and eventually you get an OutOfMemory crash or a visibility timeout storm as messages you haven't finished processing become eligible for redelivery.

All three were ruled out for the same underlying reason: they trade a hard requirement (throughput, or memory bounds) for something that's easy to reason about instead.

The design that replaced them

MongoDB was the primary store because the schema wasn't finalized and was still expected to change. Ordering wasn't required for this reporting pipeline, which meant aggressive parallelism was viable, and the solution had to maximize CPU use before reaching for horizontal scaling.

SQS hides a received message for a visibility timeout window. If the consumer pulls work faster than it can finish, a message can reappear and get processed again before the first attempt completes. Intake rate and processing capacity had to stay coupled. I used this as a starting formula for the timeout:

visibility_timeout = (queue_capacity / min_sustained_processing_rate)
                    + max_mongo_write_latency_p99
                    + gc_pause_headroom

queue_capacity and min_sustained_processing_rate come from the bounded executor described below.

Spring Cloud AWS's @SqsListener with default settings is fine for moderate traffic. The defaults were too opaque for this workload. I needed explicit control over three rates: how fast we poll, how fast we dispatch, and how fast we're allowed to fall behind before we push back on SQS itself. I kept @SqsListener as the entry point, and added a custom SqsMessageListenerContainerFactory, tuned concurrency and poll settings for batch overlap, manual acknowledgement for per-record acking, and a Spring-managed ThreadPoolTaskExecutor with CallerBlocksPolicy as backpressure.

Overlap batch receives, then parallelize per item

SQS ReceiveMessage is batch-oriented under the hood. A single poll at a time chains the network round trip in series with processing: the container waits on SQS, dispatches, then waits again. Container-level concurrency overlaps this, so while one batch is being normalized and written, another poll is already in flight.

Dependencies:

<dependency>
    <groupId>io.awspring.cloud</groupId>
    <artifactId>spring-cloud-aws-starter</artifactId>
    <version>3.2.0</version>
</dependency>
<dependency>
    <groupId>io.awspring.cloud</groupId>
    <artifactId>spring-cloud-aws-starter-sqs</artifactId>
    <version>3.2.0</version>
</dependency>

Configurable properties:

@ConfigurationProperties(prefix = "sqs")
record SqsProperties(
        String queueUrl,
        int batchSize,
        int longPollSeconds,
        int visibilityTimeoutSeconds,
        int maxConcurrentReceives) {
}

@ConfigurationProperties(prefix = "listener")
record BatchListenerProperties(
        int batchCorePoolSize,
        int batchMaxPoolSize,
        int batchQueueCapacity) {
}

@ConfigurationProperties(prefix = "processing")
record RecordProcessingProperties(
        int corePoolSize,
        int maxPoolSize,
        int queueCapacity,
        int keepAliveSeconds) {
}

Configuration values:

spring:
  cloud:
    aws:
      region:
        static: ap-south-1

sqs:
  queue-url: https://sqs.ap-south-1.amazonaws.com/123456789012/inventory-events
  batch-size: 10              # maxMessagesPerPoll, SQS API max per poll
  long-poll-seconds: 20
  max-concurrent-receives: 8  # multiplied by batch-size to set maxConcurrentMessages
  visibility-timeout-seconds: 90

listener:
  batch-core-pool-size: 16
  batch-max-pool-size: 32
  batch-queue-capacity: 100

processing:
  core-pool-size: 32
  max-pool-size: 64
  queue-capacity: 200          # bounded; CallerBlocksPolicy applies past this
  keep-alive-seconds: 30
  hash:
    version: V1

mongo:
  max-connection-pool-size: 50
  write-concern: majority       # matched to durability needs, not tuned for speed

Spring bean configuration:

@Configuration
@EnableConfigurationProperties({SqsProperties.class, BatchListenerProperties.class, RecordProcessingProperties.class})
class SqsListenerConfig {

    @Value("${spring.cloud.aws.region.static}")
    private String awsRegion;

    @Bean
    SqsAsyncClient sqsAsyncClient() {
        return SqsAsyncClient.builder()
                .region(Region.of(awsRegion))
                .credentialsProvider(DefaultCredentialsProvider.create())
                .build();
    }

    @Bean
    ThreadPoolTaskExecutor batchListenerExecutor(BatchListenerProperties props) {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(props.batchCorePoolSize());
        executor.setMaxPoolSize(props.batchMaxPoolSize());
        executor.setQueueCapacity(props.batchQueueCapacity());
        executor.setThreadNamePrefix("sqs-batch-listener-");
        executor.initialize();
        return executor;
    }

    @Bean
    ThreadPoolTaskExecutor recordExecutor(RecordProcessingProperties props) {
        MessageExecutionThreadFactory threadFactory = new MessageExecutionThreadFactory();
        threadFactory.setThreadNamePrefix("sqs-record-processor-");

        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(props.corePoolSize());
        executor.setMaxPoolSize(props.maxPoolSize());
        executor.setQueueCapacity(props.queueCapacity());
        executor.setKeepAliveSeconds(props.keepAliveSeconds());
        executor.setRejectedExecutionHandler(new CallerBlocksPolicy());
        executor.setThreadFactory(threadFactory);
        // carry MDC (sku/request context) across the executor boundary so
        // mongo_write_failed logs stay correlated once dispatch leaves the SQS thread
        executor.setTaskDecorator(runnable -> {
            var contextMap = MDC.getCopyOfContextMap();
            return () -> {
                try {
                    MDC.setContextMap(contextMap);
                    runnable.run();
                } finally {
                    MDC.clear();
                }
            };
        });
        executor.initialize();
        return executor;
    }

    @Bean
    SqsMessageListenerContainerFactory<Object> inventoryEventListenerContainerFactory(
            SqsAsyncClient sqsAsyncClient,
            ThreadPoolTaskExecutor batchListenerExecutor,
            SqsProperties sqsProperties) {
        return SqsMessageListenerContainerFactory.builder()
                .sqsAsyncClient(sqsAsyncClient)
                .configure(options -> options
                        // caps total in-flight messages across all polling threads,
                        // not just the number of concurrent poll calls
                        .maxConcurrentMessages(sqsProperties.maxConcurrentReceives() * sqsProperties.batchSize())
                        .maxMessagesPerPoll(sqsProperties.batchSize())
                        .pollTimeout(Duration.ofSeconds(sqsProperties.longPollSeconds()))
                        .messageVisibility(Duration.ofSeconds(sqsProperties.visibilityTimeoutSeconds()))
                        .listenerMode(ListenerMode.BATCH)
                        .acknowledgementMode(AcknowledgementMode.MANUAL)
                        .componentsTaskExecutor(batchListenerExecutor)
                        .build())
                .build();
    }
}

What each option is actually doing:

Overlapping polls only helps if processing is bounded. Too much concurrency without a matching processing bound just moves the backlog from SQS into memory. maxConcurrentMessages is that bound, and because permits free on ack rather than on dispatch, it stays meaningful even with fire-and-forget fan-out inside a batch.

Message listener:

@Component
class InventoryEventListener {

    private final ThreadPoolTaskExecutor recordExecutor;
    private final InventoryWriteErrorHandler errorHandler;
    private final String currentVersion;

    InventoryEventListener(ThreadPoolTaskExecutor recordExecutor, InventoryWriteErrorHandler errorHandler, @Value("${processing.hash.version}") String currentVersion) {
        this.recordExecutor = recordExecutor;
        this.errorHandler = errorHandler;
        this.currentVersion = currentVersion;
    }

    @SqsListener(
        queueNames = "${sqs.queue-url}",
        factory = "inventoryEventListenerContainerFactory"
    )
    void onBatch(List<Message<String>> messages) {
        for (Message<String> message : messages) {
            recordExecutor.submit(() -> processMessage(message));
        }
    }

    /**
    * This is a greatly simplified version of what went in production, for demo purposes
    */
    private void processMessage(Message<String> message) {
        Acknowledgement acknowledgement = MessageHeaderUtils.getHeader(
                message, SqsHeaders.MessageSystemAttributes.ACKNOWLEDGMENT_CALLBACK_HEADER, Acknowledgement.class);
        NormalizedRecord record = normalize(message);
        try {
            Document doc = buildDocument(record);
            doc.put("documentHash", computeDocumentHash(record, DocumentHashVersion.valueOf(currentVersion)));
            collection.insertOne(doc);
            acknowledgement.acknowledge();
        } catch (Exception e) {
            errorHandler.handle(acknowledgement, record, e);
        }
    }
}

A batch is not one unit of work, even though the container fetches it as one. onBatch fans each message out to recordExecutor and returns without waiting. It does not block on Mongo, and it does not ack anything itself. listenerMode(BATCH) plus acknowledgementMode(MANUAL) is what makes that split possible: the container hands the whole batch to the listener in one call, but leaves acknowledgement entirely up to the code, down to the level of a single message.

Error handler:

@Component
class InventoryWriteErrorHandler {

    void handle(Acknowledgement acknowledgement, NormalizedRecord record, Throwable t) {
        if (isDuplicateKey(t)) {
            acknowledgement.acknowledge();
            return;
        }
        if (t instanceof BusinessValidationException e) {
            log.error("business_validation_failed code={} sku={} ...", e.getErrorCode(), record.getSkuId(), e);
            acknowledgement.acknowledge();
            return;
        }
        if (t instanceof SchemaDriftException e) {
            log.error("schema_drift_detected code={} sku={} ...", e.getErrorCode(), record.getSkuId(), e);
            acknowledgement.acknowledge();
            return;
        }
        log.error("mongo_write_failed sku={} ...", record.getSkuId(), t);
    }

    private boolean isDuplicateKey(Throwable t) {
        return t instanceof MongoWriteException e
                && e.getError().getCategory() == ErrorCategory.DUPLICATE_KEY;
    }
}

InventoryWriteErrorHandler centralizes what "the write succeeded enough to ack" means, per record. It acks on a real insert, acks on a duplicate-key or a recognized business exception, and leaves the message unacked on anything else, so SQS redelivers it after the visibility timeout.

Backpressure that follows Mongo's actual write capacity

maxConcurrentMessages and CallerBlocksPolicy work together to provide backpressure. maxConcurrentMessages limits how many messages can be in flight at any time. CallerBlocksPolicy decides what happens when recordExecutor saturates: instead of accepting more work or rejecting it immediately, it blocks the submitting batchListenerExecutor thread until space frees up in the queue.

As MongoDB slows down, task submission slows down with it, which naturally reduces the rate at which new batches get accepted. Consumption stays coupled to MongoDB's actual write capacity instead of an unbounded backlog building up somewhere out of sight.

Deduping without a read before every write

For each normalized document, I computed a content hash over the business fields that defined its logical identity, stored the hash on the document, and enforced uniqueness with a unique index. SHA-256 was enough; for this workload, the collision risk was negligible.

enum DocumentHashVersion {
    V1(List.of("skuId", "locationId", "eventType", "quantity")),
    V2(List.of("skuId", "locationId", "eventType", "quantity", "unitOfMeasure"));

    final List<String> fields;

    DocumentHashVersion(List<String> fields) {
        this.fields = fields;
    }
}

String computeDocumentHash(NormalizedRecord r, DocumentHashVersion version) {
    String canonical = version.fields.stream()
            .map(r::getFieldValue)
            .collect(Collectors.joining("|"));
    String digest = Hashing.sha256()
            .hashString(canonical, StandardCharsets.UTF_8)
            .toString();
    return version.name() + ":" + digest;
}

// Mongo unique index, created once
db.inventoryEvents.createIndex(
  { documentHash: 1 },
  { unique: true, name: "uniq_content_hash" }
)

A compound unique index on the raw fields would also have worked. I preferred a fixed-size hash because it kept the index small, gave every document a single stable key, and turned identity changes into a matter of bumping the hash version instead of reshaping the index.

There's no "find before insert" anywhere in this pipeline. InventoryWriteErrorHandler is what makes insert-first safe: a duplicate key exception is a normal, expected outcome, not a failure. Since most records were genuinely new, an extra read on every insert would have added latency and load to MongoDB for no benefit. This worked because the pipeline was append-only and idempotent through the unique hash, so a duplicate could be identified after the fact instead of before it.

Tuning against a throughput target

I tuned pool sizes, queue capacity, concurrent receives, visibility timeout, and MongoDB connection settings against a target of roughly 500 messages/sec per instance, using representative spike traffic.

I tuned one bottleneck at a time. If CPUs were idle while receive latency was visible, I increased concurrent receives. If receiving outpaced processing, I adjusted pool and queue sizes until MongoDB became the bottleneck. Once MongoDB saturated, adding more concurrency stopped improving throughput. It only increased the amount of in-flight work.

Asynchronous logging

Logging is a blocking operation, and at this throughput it was measurable enough to matter.

I switched Logback to an asynchronous appender. Application threads hand log events to an in-memory queue and continue immediately, while a dedicated background thread flushes to disk. Logging wasn't part of the business transaction, so favoring throughput over guaranteeing every log entry during overload was a reasonable trade.

<appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
    <file>logs/inventory-consumer.log</file>

    <encoder>
        <pattern>%d %-5level [%thread] %logger - %msg%n</pattern>
    </encoder>

    <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
        <fileNamePattern>logs/inventory-consumer.%d{yyyy-MM-dd}.%i.log.gz</fileNamePattern>
        <maxFileSize>100MB</maxFileSize>
        <maxHistory>14</maxHistory>
    </rollingPolicy>
</appender>

<appender name="ASYNC" class="ch.qos.logback.classic.AsyncAppender">
    <queueSize>8192</queueSize>
    <discardingThreshold>0</discardingThreshold>
    <appender-ref ref="FILE"/>
</appender>

<root level="INFO">
    <appender-ref ref="ASYNC"/>
</root>

Why this matters beyond one SQS consumer

The specifics here are Spring Cloud AWS and MongoDB, but none of the decisions were really about those tools. A few lessons carried over from this pipeline that I'd apply to any high-throughput consumer:

• Bound concurrency where you can actually enforce it, not where it's easiest to configure.
• Make deduplication a write-time constraint instead of a read-before-write check.
• Decouple acknowledgement from the framework's callback boundary.

Bound concurrency where you can actually enforce it

The receive loop and the processing pool are two different rates, and only one of them, recordExecutor, has direct visibility into what's actually slow, which is Mongo. Putting the bound there, with CallerBlocksPolicy, means backpressure propagates from the true bottleneck outward: Mongo slows down, recordExecutor fills, submission blocks, and polling naturally throttles. Bounding the receive loop instead would have meant guessing at Mongo's capacity from a layer that has no way to observe it.

Make deduplication a write-time constraint instead of a read-before-write check

A read-then-compare check is intuitive, but it charges every single insert a round trip for a check that almost always passes. A unique index on a content hash moves that cost into the one case that actually needs handling: the rare duplicate, which now surfaces as an exception instead of a branch you evaluate on every record. The write path gets faster because the common case stopped paying for the rare one.

Decouple acknowledgement from the framework's callback boundary

onBatch returning has nothing to do with whether the work is done. Treating acknowledgement as a property of the individual record's outcome, not of the listener method's return, is what let batch dispatch and per-record completion move at different speeds without losing correctness. Any framework that lets you separate "the callback returned" from "the unit of work finished" is worth using that way.