[aws-stepfunctions-tasks] How to set MessageGroupId for SNS Publish Task?

[srachit]

Hello,
I am using AWS CDK V1 and trying to create the following AWS Resources:

1. Step function
2. FIFO SNS Topic connected to above Step function
3. FIFO SQS Queue subscribed to above topic

I am able to create all these individual resources and connect them, but when I try to execute the step function it fails with this error:
Invalid parameter: The MessageGroupId parameter is required for FIFO topics (Service: AmazonSNS; Status Code: 400; Error Code: InvalidParameter; Request ID: ee83f0fd-3451-5ae6-bafa-aa098b5b5f0d; Proxy: null)

Looking at the SNS Publish task documentation I do not see any property named MessageGroupID that I can define.

How do I proceed with using a FIFO SNS Topic with my Step function?

---

Adding code for reference purposes.

Creation of SQS, SNS and subscription.

const sqsQueue = new sqs.Queue(this, "SQSQueue", {
            queueName: sqsQueueName,
            fifo: true,
        });

const snsTopic = new sns.Topic(this, "SNSTopic", {
    topicName: snsTopicName,
    fifo: true,
    contentBasedDeduplication: true,
});

snsTopic.addSubscription(new subs.SqsSubscription(sqsQueue));

Task creation:

get_sns_task(taskName: string, topic: sns.ITopic) {
        const task = new tasks.SnsPublish(this, taskName, {
            topic,
            integrationPattern: sfn.IntegrationPattern.WAIT_FOR_TASK_TOKEN,
            message: sfn.TaskInput.fromObject({
                TaskToken: sfn.JsonPath.taskToken,
                TaskName: taskName,
                AssetId: sfn.JsonPath.stringAt("$.AssetId"),
                tid: sfn.JsonPath.stringAt("$.tid"),
            }),
        });
        return task;
    }

[srachit]

I was able to solve this issue by extending tasks.SNSPublish:

export interface SnsPublishProps extends tasks.SnsPublishProps {
    readonly messageGroupId: string;
}

export class FifoSnsPublish extends tasks.SnsPublish {
  private readonly messageGroupId: string;

    constructor(scope: Construct, id: string, props: SnsPublishProps) {
        super(scope, id, props);
        this.messageGroupId = props.messageGroupId;
    }

    protected _renderTask(): any {
        const result = super._renderTask()
        result.Parameters.MessageGroupId = this.messageGroupId;
        return result;
    }
}

Would be helpful if this was natively supported

[jeffgardnerdev]

Thank you @srachit! I was able to use this, and modify slightly to enable setting the values as a JSONPath (see _renderTask():

import { 
  aws_stepfunctions as sfn,
  aws_stepfunctions_tasks as sfnTasks,
} from 'aws-cdk-lib';
import { Construct } from 'constructs';

export interface SnsFifoPublishSfnTaskProps extends sfnTasks.SnsPublishProps {
  readonly messageGroupId: string;
  readonly messageDeduplicationId: string;
}

export class SnsFifoPublishSfnTask extends sfnTasks.SnsPublish {
  private readonly messageGroupId: string;
  private readonly messageDeduplicationId: string;

    constructor(scope: Construct, id: string, props: SnsFifoPublishSfnTaskProps) {
        super(scope, id, props);
        this.messageGroupId = props.messageGroupId;
        this.messageDeduplicationId = props.messageDeduplicationId;
    }

    protected _renderTask(): any {
        const result = super._renderTask();
        result.Parameters = sfn.FieldUtils.renderObject({
          ...result.Parameters,
          MessageGroupId: this.messageGroupId,
          MessageDeduplicationId: this.messageDeduplicationId,
        });
        return result;
    }
}

[github-actions[bot]]

Hello! Reopening this discussion to make it searchable.