@driimus/lambda-batch-processor

Concurrently process batch records with partial failure support.

Installation

[!WARNING] This is an ES only package. Before installing, make sure that your project's configuration supports ECMAScript modules.

pnpm add @driimus/lambda-batch-processor

Type hints

For types to work as expected, @types/aws-lambda must be installed:

pnpm add --save-dev @types/aws-lambda

Usage

[!WARNING] > ReportBatchItemFailures must be enabled to allow retrying failed messages.

import { SQSBatchProcessor } from '@driimus/lambda-batch-processor';

const processor = new SQSBatchProcessor(async (record) => {
  /** do stuff */
});

export const handler = processor.process;

Supported event sources:

• DynamoDB Streams

  import { DynamoDBBatchProcessor } from '@driimus/lambda-batch-processor';

• Kinesis Data Streams

  import { KinesisBatchProcessor } from '@driimus/lambda-batch-processor';

• SQS

  import { SQSBatchProcessor } from '@driimus/lambda-batch-processor';

Non-retryable errors

Exceptions that occur during batch processing can be treated as permanent failures.

This feature is inspired from the AWS Lambda Powertools for Java, with one key difference: By default, messages that trigger permanent failures will not be reported. In the case of SQS messages, the result will be their deletion from the queue. To send SQS messages to a dead-letter queue, you can use @driimus/sqs-permanent-failure-dlq.

Logging

You can enable logging by providing a logger compatible with the Logger interface, which is modelled after pino's function signatures.

[!NOTE] The provided logger should support serialising AggregateError objects.

If using pino, it might be worth adding pino-lambda, to preserve Lambda's standard log message format.

import { SQSBatchProcessor } from '@driimus/lambda-batch-processor';
import pino from 'pino';
import { lambdaRequestTracker, pinoLambdaDestination } from 'pino-lambda';

const destination = pinoLambdaDestination();
const withRequest = lambdaRequestTracker();

const logger = pino({}, destination);

const processor = new SQSBatchProcessor(
  async (record) => {
    /** do stuff */
  },
  {
    logger,
  },
);

export const handler = async (event, context) => {
  withRequest(event, context);

  return await processor.process(event, context);
};