Kinesis Firehose Lambda Transformation for MCP Servers — Base64 Encoding, ProcessingResult Codes, Newline Delimiters

The Firehose transformation contract

Firehose calls your Lambda synchronously with a batch of records. The event shape is fixed:

// Input event from Firehose
interface FirehoseTransformEvent {
  invocationId: string;
  deliveryStreamArn: string;
  region: string;
  records: Array<{
    recordId: string;                    // Opaque ID — must be echoed back in output
    approximateArrivalTimestamp: number; // Epoch milliseconds
    data: string;                        // BASE64-encoded record payload
    kinesisRecordMetadata?: {            // Present when source is Kinesis Data Streams
      sequenceNumber: string;
      subsequenceNumber: number;
      partitionKey: string;
      shardId: string;
      approximateArrivalTimestamp: number;
    };
  }>;
}

// Output event your Lambda must return
interface FirehoseTransformResult {
  records: Array<{
    recordId: string;   // Must match the input recordId exactly
    result: 'Ok' | 'Dropped' | 'ProcessingFailed';
    data: string;       // BASE64-encoded transformed payload (required for 'Ok' and 'ProcessingFailed')
  }>;
}

Three invariants Firehose enforces on the output:

1. The output must contain exactly one record per input record (by recordId). Missing a recordId causes Firehose to treat the batch as failed.
2. The recordId in each output record must exactly match the recordId from the corresponding input record.
3. The total size of all transformed records in the response must not exceed 6 MB. If the transformation inflates records beyond this limit, use ProcessingFailed for oversized records and handle them separately.

Complete transformation Lambda example

A production-grade transformation Lambda for AliveMCP probe results: normalizes field names, adds a processing timestamp, discards test probe events, and routes malformed records to the error prefix:

import type { FirehoseTransformationEvent, FirehoseTransformationResult,
               FirehoseTransformationResultRecord } from 'aws-lambda';

export const handler = async (
  event: FirehoseTransformationEvent
): Promise => {
  const records: FirehoseTransformationResultRecord[] = event.records.map(record => {
    try {
      // Step 1: Decode base64 input
      const payload = Buffer.from(record.data, 'base64').toString('utf-8');

      // Step 2: Parse JSON
      const probe = JSON.parse(payload);

      // Step 3: Drop test probes (result: Dropped — no error prefix, no S3 delivery)
      if (probe.isTest === true || probe.serverSlug?.startsWith('test-')) {
        return {
          recordId: record.recordId,
          result: 'Dropped',
          data: record.data,  // data field required even for Dropped records (AWS quirk)
        };
      }

      // Step 4: Normalize and enrich
      const normalized = {
        serverSlug: probe.serverSlug,
        tenantId: probe.tenantId ?? 'public',
        status: probe.status,                          // 'healthy' | 'degraded' | 'down'
        responseTimeMs: probe.responseTimeMs ?? null,
        errorCode: probe.errorCode ?? null,
        probedAt: probe.probedAt,
        processedAt: new Date().toISOString(),         // Add processing timestamp
        schemaVersion: 2,                              // Version your schema
      };

      // Step 5: Re-encode as base64 — MUST append '\n' before encoding
      // Without the newline, S3 files are one continuous blob with no record boundaries
      const transformedPayload = JSON.stringify(normalized) + '\n';
      const transformedData = Buffer.from(transformedPayload, 'utf-8').toString('base64');

      return {
        recordId: record.recordId,
        result: 'Ok',
        data: transformedData,
      };

    } catch (err) {
      // Step 6: Route unparseable or schema-invalid records to error prefix
      console.error('Transform failed for record', record.recordId, err);
      return {
        recordId: record.recordId,
        result: 'ProcessingFailed',
        data: record.data,  // Original base64 data — preserved in error prefix for debugging
      };
    }
  });

  return { records };
};

Key implementation notes:

• The \n before re-encoding is not optional. When Firehose writes a batch of records to S3, it concatenates the decoded payloads byte-for-byte. Without a newline terminator, {"a":1}{"b":2} is what lands in S3 — valid JSON for each individual record, but not newline-delimited JSON (NDJSON) that Athena, S3 Select, or Glue can parse as separate records.
• The data field is required even for Dropped records. The AWS documentation is ambiguous on this, but omitting data for Dropped records causes a 400 error from the Firehose API. Pass the original record.data through.
• Never throw from the handler. An unhandled exception causes Firehose to retry the entire batch (up to 3 times) before routing all records to the error prefix — even records that would have transformed successfully. Wrap per-record logic in try/catch and return ProcessingFailed for individual record errors.

Batch size, timeout, and concurrency

Firehose controls the batch size and invocation frequency — your Lambda does not control when it is called:

Concurrency model: Firehose invokes the transformation Lambda synchronously and waits for each batch to complete before delivering the transformed records to S3. Multiple Firehose shards (when source is KDS) may invoke the Lambda in parallel — one concurrent Lambda invocation per Firehose shard. For AliveMCP with 1 KDS shard, you will have at most 1 concurrent transformation Lambda invocation at any moment.

If Lambda concurrency is exhausted (because another workload has consumed all reserved concurrency), Firehose receives a throttling error and retries the batch. After 3 retries, the batch routes to the error prefix. Set reserved concurrency on the transformation Lambda if other workloads in the same account could exhaust the account concurrency limit.

ProcessingResult codes: when to use each

Dropped is the right choice for events you deliberately do not want to archive — test probes, internal health checks, rate-limiter synthetic requests. Unlike ProcessingFailed, dropped records do not appear in your error prefix and do not trigger alerts. Use ProcessingFailed only for records that should have been archivable but encountered an unrecoverable error — these records in the error prefix are your audit trail for data quality issues.

Retry behavior and error routing

When the transformation Lambda invocation fails — either a Lambda error (unhandled exception, timeout, out-of-memory) or a throttling error — Firehose retries the entire batch up to 3 times with exponential backoff before routing all records in that batch to the error prefix. The retry duration is controlled by the DurationInSeconds field on RetryOptions (0 to 7,200 seconds; default 300 seconds).

// Configure retry duration in the delivery stream
ProcessingConfiguration: {
  Enabled: true,
  Processors: [
    {
      Type: 'Lambda',
      Parameters: [
        {
          ParameterName: 'LambdaArn',
          ParameterValue: 'arn:aws:lambda:us-east-1:123456789012:function:firehose-transform:$LATEST',
        },
        {
          ParameterName: 'BufferSizeInMBs',
          ParameterValue: '3',   // Max 3 MB per Lambda invocation
        },
        {
          ParameterName: 'BufferIntervalInSeconds',
          ParameterValue: '60',  // Invoke Lambda at least every 60 seconds
        },
        {
          ParameterName: 'NumberOfRetries',
          ParameterValue: '3',   // 3 retries before error prefix routing
        },
        {
          ParameterName: 'RoleArn',
          ParameterValue: 'arn:aws:iam::123456789012:role/firehose-lambda-invoke-role',
        },
      ],
    },
  ],
}

The retry behavior is per-batch: if 99 out of 100 records in a batch transform successfully but 1 causes an exception, the entire batch of 100 records is retried. This makes idempotent transforms critical — a record that was successfully transformed in a prior attempt and is now being re-processed must produce the same output. The processedAt timestamp in the example above would differ between retries; use probedAt from the source record as the authoritative timestamp if deduplication matters.

Common failure modes