batch

Documentation

Overview

Package batch processes Lambda record batches and reports partial failures.

NewSQS, NewSQSFIFO, NewKinesis and NewDynamoDB create typed processors for AWS event records. New accepts a custom record Source. Configure concurrency and failure behavior with Options.

Handler composition

WrapSQS, WrapKinesis and WrapDynamoDB return Lambda handlers with a Response containing failed item identifiers. Enable ReportBatchItemFailures on the Lambda event source mapping so AWS can apply the partial response. WithParser composes payload validation without changing record identifiers.

Processor.Process returns a Report with individual outcomes. When all records fail, the report still contains failure identifiers alongside FullBatchFailureError. FIFO short-circuit rules preserve ordering; consult the guide before choosing concurrency or suppressing full-batch errors.

For usage and compatibility details, see the user guide.

Index

• func WrapDynamoDB[R any](processor RecordProcessor[events.DynamoDBEventRecord, R], ...) func(context.Context, events.DynamoDBEvent) (Response, error)
• func WrapHandler[E, T, R any](processor RecordProcessor[T, R], records func(E) []T, handler Handler[T, R]) func(context.Context, E) (Response, error)
• func WrapKinesis[R any](processor RecordProcessor[events.KinesisEventRecord, R], ...) func(context.Context, events.KinesisEvent) (Response, error)
• func WrapSQS[R any](processor RecordProcessor[events.SQSMessage, R], ...) func(context.Context, events.SQSEvent) (Response, error)
• type ConfigurationError
  • func (e *ConfigurationError) Error() string
• type FIFOGroupShortCircuitError
  • func (*FIFOGroupShortCircuitError) Error() string
• type FIFOShortCircuitError
  • func (*FIFOShortCircuitError) Error() string
• type FullBatchFailureError
  • func (e *FullBatchFailureError) Error() string
  • func (e *FullBatchFailureError) Unwrap() []error
• type Handler
  • func WithParser[T, P, R any](parse func(context.Context, T) (P, error), handler Handler[P, R]) Handler[T, R]
• type ItemFailure
• type Options
• type PanicError
  • func (e *PanicError) Error() string
  • func (e *PanicError) Unwrap() error
• type ParsingError
  • func (e *ParsingError) Error() string
  • func (e *ParsingError) Unwrap() error
• type Processor
  • func New[T, R any](source Source[T], options Options) (*Processor[T, R], error)
  • func NewDynamoDB[R any](options Options) (*Processor[events.DynamoDBEventRecord, R], error)
  • func NewKinesis[R any](options Options) (*Processor[events.KinesisEventRecord, R], error)
  • func NewSQS[R any](options Options) (*Processor[events.SQSMessage, R], error)
  • func NewSQSFIFO[R any](options Options) (*Processor[events.SQSMessage, R], error)
  • func (p *Processor[T, R]) Process(ctx context.Context, records []T, handler Handler[T, R]) (Report[T, R], error)
• type RecordProcessor
• type RecordResult
• type Report
• type Response
• type Source
• type UnexpectedBatchTypeError
  • func (*UnexpectedBatchTypeError) Error() string

Examples

• WrapSQS

Functions

func WrapDynamoDB

func WrapDynamoDB[R any](processor RecordProcessor[events.DynamoDBEventRecord, R], handler Handler[events.DynamoDBEventRecord, R]) func(context.Context, events.DynamoDBEvent) (Response, error)

func WrapHandler

func WrapHandler[E, T, R any](processor RecordProcessor[T, R], records func(E) []T, handler Handler[T, R]) func(context.Context, E) (Response, error)

WrapHandler adapts any typed event envelope and reuses shared invocation context.

func WrapKinesis

func WrapKinesis[R any](processor RecordProcessor[events.KinesisEventRecord, R], handler Handler[events.KinesisEventRecord, R]) func(context.Context, events.KinesisEvent) (Response, error)

func WrapSQS

func WrapSQS[R any](processor RecordProcessor[events.SQSMessage, R], handler Handler[events.SQSMessage, R]) func(context.Context, events.SQSEvent) (Response, error)

Example

package main

import (
	"context"
	"errors"
	"fmt"

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

func main() {
	processor, err := batch.NewSQS[string](batch.Options{Sequential: true})
	if err != nil {
		panic(err)
	}
	handler := batch.WrapSQS(processor, func(_ context.Context, record events.SQSMessage) (string, error) {
		if record.Body == "retry" {
			return "", errors.New("temporary failure")
		}
		return record.Body, nil
	})
	response, err := handler(context.Background(), events.SQSEvent{Records: []events.SQSMessage{
		{MessageId: "accepted", Body: "order-123"},
		{MessageId: "failed", Body: "retry"},
	}})
	fmt.Println(response.BatchItemFailures[0].ItemIdentifier, err)
	// Enable ReportBatchItemFailures on the event source mapping in production.
	

Types

type ConfigurationError

type ConfigurationError struct{ Message string }

func (*ConfigurationError) Error

func (e *ConfigurationError) Error() string

type FIFOGroupShortCircuitError

type FIFOGroupShortCircuitError struct{ GroupID string }

func (*FIFOGroupShortCircuitError) Error

func (*FIFOGroupShortCircuitError) Error() string

type FIFOShortCircuitError

type FIFOShortCircuitError struct{}

func (*FIFOShortCircuitError) Error

func (*FIFOShortCircuitError) Error() string

type FullBatchFailureError

type FullBatchFailureError struct{ RecordErrors []error }

func (*FullBatchFailureError) Error

func (e *FullBatchFailureError) Error() string

func (*FullBatchFailureError) Unwrap

func (e *FullBatchFailureError) Unwrap() []error

type Handler

type Handler[T, R any] func(context.Context, T) (R, error)

func WithParser

func WithParser[T, P, R any](parse func(context.Context, T) (P, error), handler Handler[P, R]) Handler[T, R]

WithParser composes record parsing and processing while keeping original record identifiers in the processor report. The parser owns schema/transform policy.

type ItemFailure

type ItemFailure struct {
	ItemIdentifier string `json:"itemIdentifier"`
}

type Options

type Options struct {
	Sequential bool
	// Zero permits all records concurrently, matching the reference default.
	MaxConcurrency           int
	SuppressFullBatchFailure bool
	SkipGroupOnError         bool
}

Options are fixed at construction and never shared as mutable invocation state.

type PanicError

type PanicError struct{ Value any }

func (*PanicError) Error

func (e *PanicError) Error() string

func (*PanicError) Unwrap

func (e *PanicError) Unwrap() error

type ParsingError

type ParsingError struct{ Err error }

func (*ParsingError) Error

func (e *ParsingError) Error() string

func (*ParsingError) Unwrap

func (e *ParsingError) Unwrap() error

type Processor

type Processor[T, R any] struct {
	// contains filtered or unexported fields
}

Processor is immutable and safe for warm reuse and concurrent invocations.

func New

func New[T, R any](source Source[T], options Options) (*Processor[T, R], error)

func NewDynamoDB

func NewDynamoDB[R any](options Options) (*Processor[events.DynamoDBEventRecord, R], error)

func NewKinesis

func NewKinesis[R any](options Options) (*Processor[events.KinesisEventRecord, R], error)

func NewSQS

func NewSQS[R any](options Options) (*Processor[events.SQSMessage, R], error)

func NewSQSFIFO

func NewSQSFIFO[R any](options Options) (*Processor[events.SQSMessage, R], error)

func (*Processor[T, R]) Process

func (p *Processor[T, R]) Process(ctx context.Context, records []T, handler Handler[T, R]) (Report[T, R], error)

Process waits for all started handlers. Cancellation prevents new handler calls and marks unstarted records as failures; handlers must honor context. A handler panic becomes a record-level PanicError, like a thrown JS exception.

type RecordProcessor

type RecordProcessor[T, R any] interface {
	Process(context.Context, []T, Handler[T, R]) (Report[T, R], error)
}

RecordProcessor allows custom processors without subclassing or shared stores.

type RecordResult

type RecordResult[T, R any] struct {
	Record T
	Result R
	Err    error
	// Skipped marks a record not delivered to the handler (FIFO or cancellation).
	Skipped bool
}

type Report

type Report[T, R any] struct {
	Response  Response
	Results   []RecordResult[T, R]
	Successes []T
	Failures  []T
	Errors    []error
}

Report belongs to one call. Results retain input order; success/failure lists retain completion order. Records and handler return values remain caller-owned.

type Response

type Response struct {
	BatchItemFailures []ItemFailure `json:"batchItemFailures"`
}

Response is the Lambda ReportBatchItemFailures response for all three sources.

type Source

type Source[T any] struct {
	Identifier func(T) string
	GroupID    func(T) string
	// DynamoDB's reference collector omits absent/empty sequence identifiers.
	OmitEmptyIdentifier bool
}

Source defines record identity and, optionally, FIFO message grouping. A non-nil GroupID enables FIFO processing. Callbacks must not panic or mutate records and must be safe for concurrent invocations.

type UnexpectedBatchTypeError

type UnexpectedBatchTypeError struct{}

func (*UnexpectedBatchTypeError) Error

func (*UnexpectedBatchTypeError) Error() string