The CEO wanted a dashboard to show the overall inventory status: how it was moving, where it was, and how much was unaccounted for. But that was not simple, because we did not have that data. So we implemented inventory audit across all B2B, B2C, and retail store applications. Every inventory flow was made to emit an event whenever a SKU item's status changed. The events were simply serialized entities plus metadata. Finally, we built a consumer application to ingest the event data and an ETL job to load it into Redshift, where the Data Analytics team could generate reports and dashboards.

Consumer responsibilities

The consumers:

  1. Read from SQS.
  2. Deserialized the payload.
  3. Parsed and extracted the fields that mattered.
  4. Projected and normalized the data into a stable document shape.
  5. Wrote to MongoDB.

Challenges

Duplicates

Multiple upstream services could emit payloads that shared the same business markers. The payloads were not byte-identical, but the business data that mattered was. If the consumer wrote careless duplicates, Redshift would inherit them, and queries would then take longer to filter them out. So it was critical to remove duplicates before inserting them.

One approach was to read the existing record from MongoDB using the SKU Item ID and then compare it with the incoming event. The problem with this approach was that it required an additional database call. It was a blocking operation that slowed down processing, and in a high-throughput pipeline, every millisecond mattered. Additionally, it put real pressure on MongoDB CPU.

Performance

If the consumer lagged for some reason, reports showed discrepancies during the next reconciliation cycle.

Design choices

MongoDB

MongoDB was chosen as the primary database because the schema was not finalized and was still expected to change.

AWS Glue

A scheduled Glue job was created to read from MongoDB and load data into Redshift, and reports were then generated from Redshift.

Scaling

Ordering was not required for this reporting pipeline, so aggressive parallelism was viable.

Spikes of hundreds of thousands of messages arrived during batch operation windows, then the queue dropped to a trickle for the rest of the day. A design tuned for quiet midday load would fail when a large inventory run dumped hundreds of thousands of messages in minutes. So the solution had to maximize use of CPU first before utilizing horizontal scaling.

SQS Visibility Timeout & Queue Sizing

SQS hid a received message for a visibility timeout window. If the instance pulled work faster than it could finish, those messages could reappear and get processed again before the first attempt completed. Intake rate and processing capacity had to stay coupled. As a starting point, I used the below formula for timeout:

visibility_timeout = (queue_capacity / min_sustained_processing_rate)
                    + max_mongo_write_latency_p99
                    + gc_pause_headroom

with queue_capacity and min_sustained_processing_rate taken from the bounded executor described below.

What I ruled out

I ruled out the following due to various reasons:

  • Single-threaded receive + process keeps ack semantics simple, but it cannot keep up with spikes.
  • An unbounded executor queue absorbs bursts in memory, can result in heap growth, and eventually crashes due to OutOfMemory and visibility timeout storms.
  • Select-then-insert for dedupe looks clean, but it costs two Mongo hops, putting pressure on MongoDB CPU and added latency means ~50% reduction in performance.

Final Design

Spring Cloud AWS's @SqsListener with default settings was fine for moderate traffic. However, the defaults were too opaque and I needed explicit control over three rates to be able to optimize for my use case. I kept @SqsListener as the entry point but 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, and the listener container handles the actual polling. I configured it to long-poll, use the maximum practical batch size to cut API calls, and keep enough messages in flight concurrently to hide network latency.

A single poll at a time still chains the network round trip in series with processing: the container waits on SQS, dispatches, then waits again. The container-level concurrency setting 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:
  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();
    }
}
  • maxConcurrentMessages is the container-factory equivalent of a semaphore-bounded receive loop: it caps how many messages can be in flight across the container's polling threads at once, which is what keeps the held-message count inside what the visibility timeout and heap can handle.
  • maxMessagesPerPoll sets the SQS batch size.
  • pollTimeout is the long-poll wait
  • messageVisibility is where the visibility-timeout sizing from earlier actually gets applied, it's a container option, not a field on a raw ReceiveMessageRequest. Those permits are released on acknowledgement, not on listener return, so maxConcurrentMessages already tracks real per-record completion even though onBatch itself returns almost immediately.
  • componentsTaskExecutor points the container's dispatch at batchListenerExecutor, which is a separate pool from recordExecutor.

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 the 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 duplicate-key or a recognized business exception, and leaves the message unacked on anything else.

Backpressure

maxConcurrentMessages and CallerBlocksPolicy worked together to provide backpressure. maxConcurrentMessages limited the number of messages that could be in flight at any time. CallerBlocksPolicy handled what happened when recordExecutor became saturated. Instead of accepting more work or rejecting it immediately, it blocked the submitting batchListenerExecutor thread until space became available in the queue. As MongoDB slowed down, task submission slowed down too, which naturally reduced the rate at which new batches were accepted. This kept message consumption aligned with MongoDB's write capacity instead of allowing an unbounded backlog to build.

Handling duplicates

For each normalized document, I computed a content hash over the selected business fields that defined its logical identity. I stored that hash on the document and enforced uniqueness with a unique index. I used a SHA256 hash function because, 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 made identity changes a matter of updating the hash version instead of reshaping the index.

There was no "find before insert." InventoryWriteErrorHandler, shown earlier, made insert-first safe. Since most records were new, an extra read on every insert would have added unnecessary latency and load to MongoDB. This worked because the pipeline was append-only and idempotent through the unique hash.

How it runs

Below is a simplified view of the consumer that was implemented:

I tuned pool sizes, queue capacity, concurrent receives, visibility timeout, and MongoDB connection settings against the target throughput of ~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 did not improve throughput. It only increased the amount of in-flight work.

Asynchronous logging

I know, this is borderline over optimization, but trust me, it was needed and worth the effort. Logging is a blocking operation, and it can add up and drag the performance significantly down (like i said earlier, every ms count).

So I switched Logback to use an asynchronous appender. Application threads only handed log events to an in-memory queue and immediately continued processing, while a dedicated background thread flushed the logs to disk. Since logging was not part of the business transaction, it was a good tradeoff to favor throughput over guaranteeing every log entry during overload.

<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>