Add SQS fair queue support (#975)

* Add SQS fair queue support

* Replace bitnami/kafka with official apache/kafka image

Author: ish-bindra

.changeset/puny-paws-rule.md

@@ -0,0 +1,5 @@
+---
+"steveo": minor
+---
+
+Steveo - Add sqs fair queues support

.github/workflows/main.yml

@@ -63,7 +63,7 @@ jobs:
         run: |
           pip install localstack awscli-local[ver1]
           docker pull localstack/localstack
-          docker run --hostname 127.0.0.1 --name kafka -d -p 9092:9092 -e KAFKA_ENABLE_KRAFT=yes -e KAFKA_CFG_NODE_ID=0 -e KAFKA_CFG_PROCESS_ROLES=controller,broker -e KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 -e KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT -e KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@127.0.0.1:9093 -e KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER bitnami/kafka:3.5
+          docker run --name kafka --network=host -d -e KAFKA_NODE_ID=1 -e KAFKA_PROCESS_ROLES=broker,controller -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092 -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@127.0.0.1:9093 -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 -e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 -e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 -e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 -e KAFKA_NUM_PARTITIONS=1 apache/kafka:latest
           localstack start -d
           localstack wait -t 30
           echo "Startup complete"

packages/steveo/src/common.ts

@@ -170,16 +170,22 @@ export interface KafkaMessageRoutingOptions {
 }
 
 /**
- * @description SQS FIFO message routing options
+ * @description SQS message routing options
  */
 export interface SQSMessageRoutingOptions {
   /**
-   * @description Groups messages with the same key.
-   *         See https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/using-messagegroupid-property.html
+   * @description Groups messages with the same key (MessageGroupId).
+   *
+   * Behavior depends on queue type:
+   * - **FIFO queues**: Messages with the same key are processed in strict order, one at a time.
+   * - **Standard queues**: Enables Fair Queue behavior - messages with the same key can be processed
+   *   in parallel while maintaining fair distribution across different keys/tenants.
+   *
+   * See https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/using-messagegroupid-property.html
    */
   key?: string;
   /**
-   * @description The message deduplication ID.
+   * @description The message deduplication ID (FIFO queues only).
    * SQS FIFO engine uses content-based deduplication by default if no message deduplication ID is provided.
    */
   deDuplicationId?: string;

packages/steveo/src/producers/sqs.ts

@@ -242,7 +242,7 @@ class SqsProducer extends BaseProducer implements IProducer {
       MessageAttributes: messageAttributes,
       MessageBody: JSON.stringify({ ...msg, _meta: messageMetadata }),
       QueueUrl: this.sqsUrls[sqsTopic],
-      MessageGroupId: fifo && options.key ? options.key : undefined,
+      MessageGroupId: options.key || undefined,
       MessageDeduplicationId:
         fifo && options.deDuplicationId ? options.deDuplicationId : undefined,
     } as SendMessageCommandInput;

packages/steveo/test/producer/sqs_test.ts

@@ -3,7 +3,7 @@ import sinon from 'sinon';
 import Producer from '../../src/producers/sqs';
 import Registry from '../../src/runtime/registry';
 import Task from '../../src/runtime/task';
-import {ITask, SQSMessageRoutingOptions} from '../../src/common';
+import { ITask, SQSMessageRoutingOptions } from '../../src/common';
 import { createMessageMetadata } from '../../src/lib/context';
 import { TaskOptions } from '../../src/types/task-options';
 
@@ -77,12 +77,12 @@ describe('SQS Producer', () => {
       const subscribeStub = sandbox.stub().resolves({ some: 'success' });
       const anotherRegistry = {
         registeredTasks: [],
-        addNewTask: () => {},
-        removeTask: () => {},
+        addNewTask: () => { },
+        removeTask: () => { },
         getTopics: () => [],
         getTaskTopics: () => [],
         getTask: () => ({
-          publish: () => {},
+          publish: () => { },
           subscribe: subscribeStub,
           options: {
             deadLetterQueue: true,
@@ -217,7 +217,7 @@ describe('SQS Producer', () => {
 
       const sentAttributes =
         sendMessageStub.getCall(0).args[0].MessageAttributes[
-          attributes[0].name
+        attributes[0].name
         ];
       const sentDataType = sentAttributes.DataType;
       const sentStringValue = sentAttributes.StringValue;
@@ -260,7 +260,7 @@ describe('SQS Producer', () => {
         },
         MessageBody: JSON.stringify(expectedMessageBody),
         QueueUrl: undefined,
-        MessageGroupId: undefined,
+        MessageGroupId: 'context',
         MessageDeduplicationId: undefined
       };
 
@@ -408,5 +408,83 @@ describe('SQS Producer', () => {
       );
       sinon.assert.calledWith(sendMessageStub, expectedPayload);
     });
+
+    it('should set MessageGroupId on standard queue when key is provided', async () => {
+      const task: ITask = new Task(
+        //@ts-expect-error
+        { engine: 'sqs' },
+        registry,
+        producer,
+        'test-task-standard-with-key',
+        'test-topic',
+        () => undefined,
+        {} // Standard queue (no fifo flag)
+      );
+      registry.addNewTask(task);
+
+      const messagePayload: any = { a: 'payload' };
+      const messageContext: SQSMessageRoutingOptions = { key: 'tenant-123' };
+      const expectedMessageBody = {
+        ...messagePayload,
+        _meta: { ...createMessageMetadata(messagePayload), ...messageContext },
+      };
+
+      const expectedPayload = {
+        MessageAttributes: {
+          Timestamp: { DataType: 'Number', StringValue: clock.now.toString() },
+        },
+        MessageBody: JSON.stringify(expectedMessageBody),
+        QueueUrl: undefined,
+        MessageGroupId: messageContext.key, // Should be set on standard queue
+        MessageDeduplicationId: undefined
+      };
+
+      const sendMessageStub = sandbox.stub(producer.producer, 'sendMessage');
+      await producer.send(
+        'test-topic',
+        messagePayload,
+        messageContext
+      );
+      sinon.assert.calledWith(sendMessageStub, expectedPayload);
+    });
+
+    it('should not set MessageGroupId when key is not provided', async () => {
+      const task: ITask = new Task(
+        //@ts-expect-error
+        { engine: 'sqs' },
+        registry,
+        producer,
+        'test-task-no-key',
+        'test-topic',
+        () => undefined,
+        {}
+      );
+      registry.addNewTask(task);
+
+      const messagePayload = { a: 'payload' };
+      const messageContext: SQSMessageRoutingOptions = {}; // No key
+      const expectedMessageBody = {
+        ...messagePayload,
+        _meta: { ...createMessageMetadata(messagePayload), ...messageContext },
+      };
+
+      const expectedPayload = {
+        MessageAttributes: {
+          Timestamp: { DataType: 'Number', StringValue: clock.now.toString() },
+        },
+        MessageBody: JSON.stringify(expectedMessageBody),
+        QueueUrl: undefined,
+        MessageGroupId: undefined, // Should be undefined without key
+        MessageDeduplicationId: undefined
+      };
+
+      const sendMessageStub = sandbox.stub(producer.producer, 'sendMessage');
+      await producer.send(
+        'test-topic',
+        messagePayload,
+        messageContext
+      );
+      sinon.assert.calledWith(sendMessageStub, expectedPayload);
+    });
   });
 });