Option to Permanently Unlock SQS Trigger for Continuous Consumption

[surya9teja]

Feature description

Hi, The existing SQS trigger in Kestra implements a locking mechanism upon pipeline execution. While this aims to prevent reprocessing, the default behavior of the SQS trigger is to consume and delete messages. This makes the locking period, until the execution completes, potentially restrictive, especially in scalable environments that we deployed the Kestra such as Kubernetes. We propose adding an option to permanently unlock the SQS trigger, allowing it to continuously poll and trigger new executions for incoming SQS messages, thereby maximizing parallel processing capabilities.

[loicmathieu]

What you descrive is how the Realtime trigger will work: https://kestra.io/plugins/plugin-aws/sqs/io.kestra.plugin.aws.sqs.realtimetrigger

Except this trigger is always on, it didn't poll SQS for messages periodically, and will trigger an execution for each message.

The fact that the SQS Trigger is locked during the whole execution it triggered is the way all triggers work, this is not specific to SQS and is a core behavior of the Scheduler (the component behind all triggers).

[surya9teja]

Thank you for the explanation regarding the RealtimeTrigger. I understand its intended use case.

What I'm hoping for is a way to decouple the message batching capability of the standard SQS trigger from its execution locking behavior. Ideally, the trigger would continuously accumulate messages (up to maxRecords) into batches, making them ready for new executions as soon as the scheduler is available, without being blocked by the completion of previous executions.

Consider the following example

id: image_cropper_worker
namespace: microform.internal

triggers:
  - id: image_cropper_job_trigger
    type: io.kestra.plugin.aws.sqs.Trigger
    accessKeyId: "key"
    secretKeyId: "secret"
    region: "eu-west-2"
    serdeType: STRING
    queueUrl: "url"
    maxRecords: 500
    maxDuration: PT1M
concurrency:
  limit: 10

tasks:
  - id: to_json
    type: io.kestra.plugin.serdes.json.IonToJson
    from: "{{ trigger.uri }}"

  - id: log
    type: io.kestra.plugin.core.log.Log
    message: "{{read(outputs.to_json.uri)}}"

  - id: deskew_worker
    type: io.kestra.plugin.kubernetes.PodCreate
    namespace: dev
    metadata:
      annotations:
        karpenter.sh/do-not-disrupt: "true"
    waitRunning: PT4H
    waitUntilRunning: PT15M
    waitForLogInterval: PT1S
    resume: true
    inputFiles:
      deskew.py: "{{read('image_cropper/deskew.py')}}"
      data.jsonl: "{{outputs.to_json.uri}}"
    outputFiles:
      - "output.jsonl"
    spec:
      serviceAccount: kestra-sa
      containers:
        - name: deskew-worker
          image: microform/internal:microform_crop_images-v1.7
          command:
            - python
            - "{{workingDir}}/deskew.py"
            - "--file"
            - "{{workingDir}}/data.jsonl"
            - "--output_path"
            - "{{workingDir}}/output.jsonl"
          imagePullPolicy: IfNotPresent
          resources:
            requests:
              memory: "8Gi"
              cpu: "5"
            limits:
              memory: "8Gi"
              cpu: "5"
      nodeSelector:
        karpenter.sh/nodepool: tasks-cpu-node-pool
        resource-type: cpu
      restartPolicy: Never

In this pipeline, the deskew_worker task, which runs as a Kubernetes Pod, might take up to an hour to process a batch of 500 images (SQS message will have image uris) . Currently, theimage_cropper_job_trigger locks for this entire hour, preventing any new messages from being collected and processed, even if the queue has more messages.

By allowing the trigger to continuously poll for new batches, regardless of the current execution status, we could fully leverage Kestra's integration with Kubernetes and its scaling capabilities. As soon as one batch of tasks finishes (or even while it's running, up to the concurrency.limit), the scheduler could immediately initiate a new execution with the next available batch of messages, significantly improving the overall processing throughput.

If there are alternative patterns or configurations within Kestra that could help achieve a similar outcome that I haven't considered yet? Any insights or suggestions would be greatly appreciated.

[loicmathieu]

You can alternatively use a Subflow and don't wait for its completion.
So you have a parent flow that is triggered by a batch of messages, then launch a subflow with the batch of messages as input and don't wait for its completion.

It's more complex but it'll work.

Anyway, I'll open a discussion internally to check if we want to offer a way to 'not lock' an execution by a trigger or not at some point.

[surya9teja]

Thanks I think subflow might work for my scenario.

[anna-geller]

We'll follow @loicmathieu's suggestion. Also reopened kestra-io/kestra#311 as it's related