EventBridge Pipes Error Handling

Two DLQ layers — source DLQ vs pipe DLQ

Understanding which layer handles which failures is the foundation of Pipes error handling:

// CDK: full error handling configuration for SQS-sourced pipe
import { Queue } from "aws-cdk-lib/aws-sqs";
import { aws_pipes as pipes } from "aws-cdk-lib";

// Layer 1: SQS source DLQ (catches messages that fail at the queue level)
const sourceDlq = new Queue(this, "SourceDlq", {
  queueName: "mcp-tool-calls-source-dlq",
  retentionPeriod: cdk.Duration.days(14),
});

const toolCallQueue = new Queue(this, "ToolCallQueue", {
  queueName: "mcp-tool-calls",
  visibilityTimeout: cdk.Duration.seconds(120),
  deadLetterQueue: {
    queue: sourceDlq,
    maxReceiveCount: 3,  // Move to DLQ after 3 failed receive cycles
  },
});

// Layer 2: Pipe DLQ (catches enrichment/target failures after Pipes consumed the message)
const pipeDlq = new Queue(this, "PipeDlq", {
  queueName: "mcp-tool-calls-pipe-dlq",
  retentionPeriod: cdk.Duration.days(14),
});

const pipe = new pipes.CfnPipe(this, "McpEventPipe", {
  // ...
  source: toolCallQueue.queueArn,
  sourceParameters: {
    sqsQueueParameters: {
      batchSize: 10,
      maximumBatchingWindowInSeconds: 5,
    },
    // Note: destinationConfig is NOT available for SQS sources in Pipes
    // For SQS, the SQS-level DLQ + pipe RUNNING_FAILED state handles failures
    // Pipe DLQ is configured via the KINESIS/DYNAMODB parameters instead
  },
  // ...
});

Important nuance: the pipe-level destinationConfig.onFailure is configured inside the sourceParameters for streaming sources (Kinesis, DynamoDB) but is not available for SQS sources in Pipes. For SQS-sourced pipes, failed batches are retried based on the SQS visibility timeout and maxReceiveCount — the message becomes visible again after visibility timeout expires and gets re-polled. If it exceeds maxReceiveCount, it goes to the SQS source DLQ. This is why the SQS visibility timeout must exceed total pipe processing time: a visibility timeout that's too short causes the message to become visible before Pipes finishes, leading to duplicate processing before the message eventually hits maxReceiveCount and is dead-lettered.

Kinesis/DynamoDB retry configuration

// CDK: Kinesis source with bounded retry configuration
const pipe = new pipes.CfnPipe(this, "KinesisPipe", {
  source: eventStream.streamArn,
  sourceParameters: {
    kinesisStreamParameters: {
      startingPosition: "LATEST",
      batchSize: 100,
      maximumBatchingWindowInSeconds: 5,

      // Retry config — CRITICAL: don't leave at -1 (indefinite)
      maximumRetryAttempts: 3,        // Retry failed batches up to 3 times
      maximumRecordAgeInSeconds: 3600, // Drop records older than 1h (don't retry stale events)

      // Partial failure: AUTOMATIC_BISECT splits a failing batch in half
      // to isolate the record(s) causing failures
      onPartialBatchItemFailure: "AUTOMATIC_BISECT",

      // Pipe-level DLQ: receives the failed records after retry exhaustion
      destinationConfig: {
        onFailure: {
          destination: pipeDlq.queueArn,  // SQS or SNS
        },
      },
    },
  },
  // ...
});

// Grant pipe role to write to the DLQ
pipeRole.addToPolicy(new PolicyStatement({
  actions: ["sqs:SendMessage"],
  resources: [pipeDlq.queueArn],
}));

AUTOMATIC_BISECT — partial batch failure isolation

When onPartialBatchItemFailure: "AUTOMATIC_BISECT" is set, and a batch fails (enrichment or target throws for the batch), Pipes splits the batch in half and retries each half independently. This isolates the failing record(s) to progressively smaller batches until the failing record is alone in a batch of 1, at which point it can be sent to the DLQ without affecting neighboring records.

This is available for Kinesis and DynamoDB Streams sources. It is not available for SQS sources in Pipes (for SQS, use Lambda ESM with reportBatchItemFailures / batchItemFailures response format instead).

// Target Lambda: how to handle partial batch failures
// For Kinesis/DynamoDB sources with AUTOMATIC_BISECT, the Lambda target
// does NOT need to use batchItemFailures response format (unlike Lambda ESM).
// Pipes handles bisection automatically when the Lambda throws.

// Pattern: throw on any failure — let Pipes bisect
exports.handler = async (events) => {
  const results = await Promise.allSettled(
    events.map(evt => writeToTarget(evt))
  );

  const failures = results.filter(r => r.status === "rejected");
  if (failures.length > 0) {
    // Throw to signal batch failure — Pipes bisects automatically
    throw new Error(`${failures.length}/${events.length} events failed`);
  }
  // No return value needed for Lambda targets in FIRE_AND_FORGET mode
};

// Note: if using invocationType: "REQUEST_RESPONSE", target Lambda return
// value is ignored by Pipes (enrichment Lambda return IS used — target is not)

RUNNING_FAILED — what it means and how to recover

RUNNING_FAILED is a terminal error state where the pipe stops polling the source entirely. It occurs when:

• Enrichment or target failures exhaust maximumRetryAttempts (for Kinesis/DynamoDB sources)
• A batch that cannot be bisected further (single-record batch) still fails
• The pipe DLQ is not configured and records fail — Pipes enters RUNNING_FAILED rather than discarding events

Recovery steps:

1. Run aws pipes describe-pipe --name YOUR_PIPE_NAME and read the StateReason field — it names the specific failure (enrichment exception message, target error code).
2. Fix the root cause (Lambda bug, DynamoDB capacity, enrichment timeout).
3. If the pipe DLQ has the failed records, decide whether to replay or discard them.
4. Call aws pipes start-pipe --name YOUR_PIPE_NAME to resume polling from where it stopped.

# CLI: diagnose RUNNING_FAILED and resume
aws pipes describe-pipe \
  --name mcp-tool-call-router \
  --query '{State: State, StateReason: StateReason, LastModifiedTime: LastModifiedTime}'

# After fix, resume:
aws pipes start-pipe --name mcp-tool-call-router

# Check pipe DLQ for failed records:
aws sqs get-queue-attributes \
  --queue-url https://sqs.us-east-1.amazonaws.com/123456789012/mcp-tool-calls-pipe-dlq \
  --attribute-names ApproximateNumberOfMessages

Self-referential pipe loops — the EventBridge Bus target trap

If your pipe target is an EventBridge Bus, and you have a Rule on that bus that routes events matching your pipe's output back to the same SQS queue that is the pipe's source, you have created an infinite loop:

SQS → Pipe → EventBridge Bus → Rule → SQS → Pipe → ...

Each cycle puts a message back on the queue. The loop amplifies with every round: if 10 messages are processed, each produces 10 more, resulting in 100 → 1,000 → 10,000 messages in minutes. This can overwhelm your SQS queue, hit Pipes throughput limits, and generate significant unexpected costs.

// Safe pattern: use dead-letter routing to a DIFFERENT queue
// or use an event detail-type prefix to stop the loop

// Option 1: add a "processed" flag to the target event and filter it out
const noLoopFilter = {
  body: {
    // Only process events that haven't been routed through this pipe before
    processedByPipe: [{ "exists": false }],
  },
};

// Option 2: use different event detail-type on the bus to prevent re-routing
// PutEvents: set detail-type = "McpToolCallProcessed"
// Rule: only match detail-type = "McpToolCallRaw" (not "McpToolCallProcessed")
// → loop is broken because processed events don't match the rule

// Option 3: target a different downstream queue (not the pipe's source)
// SQS-source → Pipe → EventBridge Bus → Rule → DIFFERENT_SQS_QUEUE
// No loop possible — different queue isn't the pipe's source

Monitor for loops using a CloudWatch alarm on the SQS queue's ApproximateNumberOfMessagesVisible metric — a sudden exponential spike is the signature of an event loop. Set an alarm at 10x the expected maximum queue depth.

Error handling decision tree