Batch processing¶

Process SQS, Kinesis and DynamoDB Streams records individually and return the IDs that Lambda should retry.

Install¶

Use Go 1.27 or newer:

CGO_ENABLED=0 go get github.com/rambow-cloud/powertools-lambda-go/batch@v1.2.0
CGO_ENABLED=0 go get github.com/rambow-cloud/powertools-lambda-go/logger@v1.2.0
CGO_ENABLED=0 go get github.com/aws/aws-lambda-go@v1.55.0

Retry only failed SQS messages¶

One message contains a valid order; the other has the wrong field type. Enable ReportBatchItemFailures on the SQS event source mapping to use the partial response.

Input

{
  "Records": [
    {
      "messageId": "m1",
      "body": "{\"id\":\"ORD-123\"}"
    },
    {
      "messageId": "m2",
      "body": "{\"id\":123}"
    }
  ]
}

Code

package main

import (
    "context"
    json "encoding/json/v2"
    "os"

    "github.com/aws/aws-lambda-go/events"
    "github.com/aws/aws-lambda-go/lambda"
    "github.com/rambow-cloud/powertools-lambda-go/batch"
    "github.com/rambow-cloud/powertools-lambda-go/logger"
)

type Order struct {
    ID string `json:"id"`
}

func main() {
    appLog := logger.New(logger.WithServiceName("orders"))
    if err := run(appLog); err != nil {
        if logErr := appLog.Error("Application failed", err); logErr != nil {
            panic(logErr)
        }
        os.Exit(1)
    }
}

func run(appLog *logger.Logger) error {
    processor, err := batch.NewSQS[string](batch.Options{})
    if err != nil {
        return err
    }
    handler := func(_ context.Context, record events.SQSMessage) (string, error) {
        var order Order
        if err := json.Unmarshal([]byte(record.Body), &order); err != nil {
            return "", err
        }
        // Replace this with your business operation.
        return order.ID, nil
    }
    lambda.Start(batch.WrapSQS(processor, handler))
    return nil
}

Output

{
  "batchItemFailures": [
    {
      "itemIdentifier": "m2"
    }
  ]
}

Only m2 is reported for retry. The successful order ID is a business result, not part of the SQS response.

Limit concurrency¶

processor, err := batch.NewSQS[string](batch.Options{MaxConcurrency: 5})

Use Sequential: true when callbacks must run one at a time. The processor waits for all started callbacks before returning.

Process records directly¶

Use Process when you need successful business results as well as the Lambda response:

report, err := processor.Process(ctx, records, handler)

report.Response contains retry identifiers; report.Results keeps results in input order. Handle both the report and err.

Choose a source¶

Source	Constructor	Retry identifier
SQS Standard	`NewSQS`	Message ID
SQS FIFO	`NewSQSFIFO`	Message ID, with FIFO stop/skip behavior
Kinesis	`NewKinesis`	Sequence number
DynamoDB Streams	`NewDynamoDB`	Sequence number

For FIFO queues, review FIFO processing before changing ordering or group policies.

Things to know¶

A returned callback error marks that record as failed. Successful calls are not retried by this response.
By default, all records failing produces FullBatchFailureError as well as a populated report. SuppressFullBatchFailure keeps the response without that error.
A Lambda invocation that returns an error still fails the whole batch, regardless of a partial response.
Create the processor once; use the invocation context inside callbacks.

Next steps¶

See the API and behavior notes for advanced options, and module installation for dependencies.