Kinesis Firehose Advanced Patterns for MCP Server Probe Pipelines: Dynamic Partitioning, Lambda Transformation, S3 Delivery, and Error Handling

Pattern 1: Dynamic Partitioning — per-tenant S3 routing without a Lambda preprocessor

Without dynamic partitioning, every probe result from every tenant of every MCP endpoint lands in the same hourly S3 prefix: probe-results/2026/10/30/14/. An Athena query asking for a specific tenant's probe failures over seven days must scan every record from every tenant over that period. At five million probe records per day across a thousand tenants, that is a 35-million-record scan for a query that conceptually touches 35,000 rows. Dynamic partitioning routes each record to its own probe-results/tenant=acme/tool=search/2026/10/30/14/ prefix at ingest time using values extracted from the record — no Lambda preprocessor, no additional fan-out queue. The same query now scans only the files under tenant=acme/, reducing data scanned by 1,000× and Athena cost proportionally.

The enable-once-at-creation constraint

Dynamic partitioning must be configured when the Firehose delivery stream is created. It cannot be enabled on an existing stream — the UpdateDestination API returns InvalidArgumentException if you attempt to add DynamicPartitioningConfiguration to a stream that was created without it. To add dynamic partitioning to an existing pipeline, you must create a new delivery stream, update the upstream producer (or Kinesis Data Streams Firehose source) to write to the new stream name, drain the old stream, and delete it. Plan for this constraint at stream design time; retrofitting dynamic partitioning to a stream handling millions of records per day is a non-trivial migration.

JQ-1.6 expression syntax and partition key limits

Firehose uses JQ 1.6 to evaluate the MetadataExtractionQuery expression against each JSON record. The expression must return an object where each key becomes a partition key name available in the S3 prefix. Common patterns for MCP probe pipelines:

Hard limits on partition keys: maximum 50 per delivery stream; key values truncated to 2,048 bytes; key values are not URL-encoded in the S3 object key path — avoid /, ?, #, or spaces in partition key fields. If the JQ expression fails to extract a key (null result or JSON parse error on the record), the record is routed to the error prefix. This is the most common cause of unexpected error prefix growth: teams add a new field to probe results, deploy the producer change, and forget that older records without the new field now fail JQ extraction and route to the error prefix instead of S3.

Buffer semantics per partition — the billing trap

The buffer hints (SizeInMBs and IntervalInSeconds) apply per active partition, not globally. A delivery stream with 1,000 tenant partitions receiving one record per tenant per minute will flush 1,000 independent 60-second buffers, not one global buffer. Each flush is one S3 PUT. S3 PUT costs $0.005 per 1,000 requests — a minor figure for a handful of partitions that becomes a significant cost at scale:

The right mitigation is tiered partitioning: route free-tier tenants to a shared coarse-grained partition (tier=free/date=2026-10-30/) and reserve per-tenant partitioning for paid tiers with sufficient traffic to produce reasonably sized S3 files. A paid tenant with 500 monitored endpoints generating 500 probe results per minute fills a 64 MB buffer in about 2 hours (500 × ~2 KB × 60 min = 60 MB/hour) — the file size makes the PUT worthwhile. A free-tier user with one endpoint generates one 2 KB file per 60-second buffer flush — a 2 KB S3 object with the same storage and metadata overhead as a 64 MB object.

Processor ordering for deaggregation

When dynamic partitioning is enabled, Firehose must deaggregate multi-record batches before extracting partition keys — because different records in the same batch may belong to different partitions. If your producer uses KPL (Kinesis Producer Library) aggregation, add RecordDeAggregation before MetadataExtraction. The ordering is strict:

// Correct processor ordering for KPL-aggregated records
Processors: [
  { Type: 'RecordDeAggregation', Parameters: [
    { ParameterName: 'SubRecordType', ParameterValue: 'DELIMITED' },
  ]},
  { Type: 'MetadataExtraction', Parameters: [
    { ParameterName: 'MetadataExtractionQuery', ParameterValue: '{tenantId: .tenantId, toolName: .toolName}' },
    { ParameterName: 'JsonParsingEngine', ParameterValue: 'JQ-1.6' },
  ]},
  { Type: 'AppendDelimiterToRecord', Parameters: [
    { ParameterName: 'Delimiter', ParameterValue: '\\n' },
  ]},
]

Reversing the order — MetadataExtraction before RecordDeAggregation — causes the extractor to see the KPL aggregated envelope rather than the individual JSON records, which fails every key extraction and routes all records to the error prefix. The AppendDelimiterToRecord processor (which adds the newline delimiter that makes S3 files parseable as NDJSON) must always run last.

S3 prefix expression reference

Use Hive-compatible prefixes (tenant=value/date=2026-10-30 format) when the S3 location will be registered as an Athena external table — they enable partition projection so Athena detects new daily or hourly partitions automatically without operator action. The timestamp format follows Java DateTimeFormatter patterns; yyyy/MM/dd/HH produces the Hive-standard hourly partition layout.

Pattern 2: Lambda Transformation — normalizing probe results before archiving

Lambda transformation inserts a synchronous preprocessing step between Kinesis Firehose buffering and S3 delivery. The canonical uses for an MCP probe pipeline: normalize field names across schema versions (probe results from older agents may use responseTime where newer agents emit responseTimeMs), filter out test-mode probe events and internal health check synthetics before they enter the queryable archive, add a processing timestamp for audit trail purposes, and validate that required fields are present before the record reaches S3. The Firehose transformation contract is deceptively specific, and violating any of its three main invariants produces silent data routing to the error prefix.

The FirehoseTransformationEvent base64 contract

Firehose sends records to the transformation Lambda base64-encoded and expects base64-encoded output. This bidirectional encoding is non-negotiable — it is not a transport artifact that Firehose handles transparently. The full event and result shape:

// 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 KDS
      sequenceNumber: string;
      subsequenceNumber: number;
      partitionKey: string;
      shardId: string;
      approximateArrivalTimestamp: number;
    };
  }>;
}

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

The three invariants Firehose enforces: (1) exactly one output record per input record, matched by recordId; (2) the total size of all transformed records in the response must not exceed 6 MB; (3) the data field must be base64-encoded, not a plain string. Violating invariant 1 by omitting a recordId causes Firehose to treat the entire batch as failed and retry it. Violating invariant 3 by returning a plain UTF-8 string routes every record to the error prefix — Firehose decodes the data field from base64 before writing to S3; if it cannot decode, the record fails.

Complete production transformation Lambda

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

export const handler = async (
  event: FirehoseTransformationEvent
): Promise<FirehoseTransformationResult> => {
  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 — Dropped = no S3, no error prefix, no metric
      if (probe.isTest === true || probe.serverSlug?.startsWith('test-')) {
        return {
          recordId: record.recordId,
          result: 'Dropped',
          data: record.data,  // data field required even for Dropped (AWS quirk)
        };
      }

      // Step 4: Normalize — handle responseTime vs responseTimeMs schema versions
      const normalized = {
        serverSlug: probe.serverSlug,
        tenantId: probe.tenantId ?? 'public',
        status: probe.status,                          // 'healthy' | 'degraded' | 'down'
        responseTimeMs: probe.responseTimeMs ?? probe.responseTime ?? null,
        errorCode: probe.errorCode ?? null,
        probedAt: probe.probedAt,
        processedAt: new Date().toISOString(),
        schemaVersion: 2,
      };

      // Step 5: Append '\n' BEFORE base64 re-encoding
      // Without the newline, S3 files are one continuous blob — Athena cannot parse NDJSON
      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) {
      // Never throw — unhandled exception retries the entire 100-record batch
      console.error('Transform failed for record', record.recordId, err);
      return {
        recordId: record.recordId,
        result: 'ProcessingFailed',
        data: record.data,  // Preserve original base64 in error prefix for debugging
      };
    }
  });

  return { records };
};

ProcessingResult codes: Ok, Dropped, ProcessingFailed

Dropped is the correct result for events you deliberately do not want to archive. Unlike ProcessingFailed, dropped records do not appear in the error prefix and do not trigger the ProcessingFailedRecords metric — there is no alarm footprint. Use ProcessingFailed for records that should have been archivable but encountered an unrecoverable error; these records in the error prefix form the audit trail for data quality issues and can be replayed after fixing the Lambda.

The data field for Dropped records is required despite not being written to S3 — the AWS documentation is ambiguous on this, but omitting it causes a 400 error from the Firehose internal validation. Pass record.data (the original base64 input) through unchanged for Dropped records.

Batch limits, timeout, and concurrency

Concurrency is determined by the upstream source: for a Kinesis Data Streams source, Firehose invokes the transformation Lambda synchronously once per shard. A one-shard KDS stream driving one Firehose delivery stream means at most one concurrent transformation Lambda invocation. If Lambda reserved concurrency is exhausted by other workloads in the same AWS account, Firehose receives a throttling response and retries — after 3 throttling-caused retries, the batch routes to the error prefix. Set reserved concurrency on the transformation Lambda explicitly if other high-throughput workloads share the account concurrency pool.

The retry model is per-batch: if one record in a 100-record batch throws an unhandled exception that escapes the handler, Firehose retries all 100 records. The 99 records that would have transformed successfully are retried unnecessarily, adding latency and increasing the risk that a non-idempotent transform (one that uses processedAt: new Date().toISOString()) produces duplicate records in the error prefix. Wrap per-record logic in try/catch and return ProcessingFailed for individual failures — this keeps the 99 healthy records on the Ok path and limits error prefix growth to the genuinely problematic record.

Pattern 3: S3 Delivery Configuration — buffer hints, Parquet, and Hive partitioning

The default Firehose S3 delivery configuration (5 MB buffer, 300-second interval, plain JSON, UNCOMPRESSED) is a reasonable development default and a poor production configuration for any table you plan to query repeatedly. Three independent optimization decisions each compound with the others: choosing the right buffer hints eliminates S3 small-file fragmentation; enabling Parquet format conversion reduces Athena scanned data by 5–10×; using Hive-compatible S3 prefixes with Athena partition projection eliminates maintenance operations on the query table. Together, these reduce the operational cost of the Firehose pipeline from "fine for occasional queries" to "practical for daily production Athena workloads."

Buffer hint presets by use case

Firehose flushes the buffer when either the size limit or the interval limit is reached — whichever comes first. Lower limits mean more frequent, smaller S3 files; higher limits mean fewer, larger files. The S3 PUT cost difference between 60-second and 900-second intervals is small in absolute terms ($0.001/day per partition at typical MCP probe volumes), but the file count difference has significant Athena performance implications: thousands of 5–10 KB Parquet files per prefix cause Athena to spend most of its execution time on S3 metadata operations rather than data scanning, inflating query duration and cost even without increasing data volume.

For AliveMCP's probe result pipeline: use 64 MB / 60-second hints for the operational status dashboard (probe failures must appear in Athena within 2 minutes to catch ongoing incidents before the next probe cycle fires) and 128 MB / 900-second hints for the long-term compliance archive (used for monthly tenant SLA reporting where 15-minute staleness is acceptable and cost efficiency matters more).

Hive-compatible prefixes and Athena partition projection

Athena partitioned tables require knowing which S3 prefixes contain data for each time range. Without partition projection, Athena queries only cover partitions registered in the Glue Data Catalog — which means running MSCK REPAIR TABLE after each delivery, or after each new hour of data arrives. Hive-compatible prefixes (key=value format) combined with Athena partition projection eliminate this: Athena computes which S3 prefixes to scan directly from the partition column ranges defined in the table DDL, with no manual partition registration required.

// Hive-compatible S3 prefix in delivery stream
Prefix: 'probe-results/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/',
ErrorOutputPrefix: 'probe-errors/type=!{firehose:error-output-type}/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/',

// Athena table DDL with partition projection — no MSCK REPAIR TABLE ever needed
CREATE EXTERNAL TABLE probe_results (
  server_slug   STRING,
  tenant_id     STRING,
  status        STRING,
  response_time_ms BIGINT,
  error_code    STRING,
  probed_at     STRING,
  schema_version INT
)
PARTITIONED BY (year STRING, month STRING, day STRING, hour STRING)
STORED AS PARQUET
LOCATION 's3://alivemcp-probe-archive/probe-results/'
TBLPROPERTIES (
  'projection.enabled'       = 'true',
  'projection.year.type'     = 'integer',
  'projection.year.range'    = '2026,2030',
  'projection.year.digits'   = '4',
  'projection.month.type'    = 'integer',
  'projection.month.range'   = '01,12',
  'projection.month.digits'  = '2',
  'projection.day.type'      = 'integer',
  'projection.day.range'     = '01,31',
  'projection.day.digits'    = '2',
  'projection.hour.type'     = 'integer',
  'projection.hour.range'    = '00,23',
  'projection.hour.digits'   = '2',
  'storage.location.template'= 's3://alivemcp-probe-archive/probe-results/year=${year}/month=${month}/day=${day}/hour=${hour}/'
);

JSON-to-Parquet format conversion via Glue Data Catalog

Format conversion transforms each JSON record to Parquet (columnar) before writing to S3. Athena scans 5–10× less data on a Parquet table because it reads only the columns referenced in the query — a query for SELECT status, COUNT(*) FROM probe_results WHERE tenant_id = 'acme' GROUP BY status reads only the status and tenant_id columns rather than deserializing the entire JSON record including responseTimeMs, errorCode, probedAt, and all other fields.

Requirements: a Glue Data Catalog database and table with schema matching the probe result records; IAM role with glue:GetTable and glue:GetTableVersions; source records must be valid JSON. One important incompatibility: format conversion cannot be combined with dynamic partitioning inline parsing — if you need both per-tenant partitioning and Parquet output, use a Lambda preprocessor that extracts the partition key as a top-level field, then configure MetadataExtraction on that normalized field. The Lambda transformation handles both concerns in a single pass.

// Format conversion configuration in delivery stream
DataFormatConversionConfiguration: {
  Enabled: true,
  InputFormatConfiguration: {
    Deserializer: { OpenXJsonSerDe: {} },  // Handles nested JSON objects and arrays
  },
  OutputFormatConfiguration: {
    Serializer: {
      ParquetSerDe: {
        Compression: 'SNAPPY',          // Parquet-native; fast decompression for Athena
        BlockSizeBytes: 268435456,      // 256 MB row group (default — good for 128 MB buffers)
        WriterVersion: 'V2',            // Improved null handling vs V1
        EnableDictionaryCompression: true,  // Efficient for low-cardinality columns (status, tier)
      },
    },
  },
  SchemaConfiguration: {
    RoleARN: 'arn:aws:iam::123456789012:role/firehose-glue-role',
    DatabaseName: 'alivemcp',
    TableName: 'probe_results',
    Region: 'us-east-1',
    VersionId: 'LATEST',  // Always picks up schema additions to the Glue table
  },
}

Parquet with Snappy compression is the default recommendation. Snappy is Parquet-native and provides fast decompression for Athena's parallel column reader — better query latency than GZIP Parquet, with only a slightly larger file size. Use GZIP with Parquet only when S3 storage cost optimization is the primary constraint and query latency is secondary. For plain JSON delivery (without format conversion), GZIP provides ~8:1 compression ratio on repetitive JSON — much better than Parquet Snappy's ~3:1 on equivalent data — and is fully compatible with Athena's JSON SerDe. Avoid ZIP for any Athena-queried table; Athena cannot read ZIP-compressed S3 files.

Format conversion adds approximately 3–5 seconds of processing overhead per buffer flush. This is negligible for batch archiving workloads but means the minimum effective delivery latency is ~65 seconds even with a 60-second buffer interval. Account for this when setting the DataFreshness alarm threshold.

IAM permissions required for Firehose S3 delivery

The Firehose service role requires s3:GetObject and s3:ListBucket in addition to s3:PutObject on the destination bucket. These read permissions are only used during stream creation for S3 accessibility validation — not during actual record delivery — but omitting them causes stream creation to fail with an access denied error. This is a common reason why delivery stream creation succeeds in one environment (where someone manually granted broader permissions) but fails in another (where the IaC role template only includes PutObject).

Pattern 4: Error Handling and Monitoring — detecting silent failures before data loss

Firehose's silent failure model is its most dangerous operational characteristic for an MCP probe pipeline. PutRecord and PutRecordBatch always return HTTP 200 to the producer regardless of what happens to the records downstream. The delivery stream console shows no error state. CloudWatch metrics increment silently for failed records. The only visible signal is records accumulating in the error prefix — which nobody checks unless an alarm fires. A Lambda transformation bug introduced in a deployment can route every probe result to the processing-failed/ prefix for hours before anyone notices, because the producer is healthy, the stream appears active, and the S3 archive grows (just in the wrong prefix). Setting up proactive monitoring before the first production deployment is not optional — it is the difference between a 5-minute incident and a 4-hour data recovery exercise.

The error routing model

processing-failed is far more common than delivery-failed in MCP probe pipelines — every Lambda bug, every schema mismatch, every malformed record from the producer flows there. delivery-failed typically indicates a one-time infrastructure event (bucket policy change, accidental deletion) rather than an ongoing data quality problem. Monitoring DeliveryToS3.ProcessingFailedRecords catches Lambda bugs early; monitoring DeliveryToS3.DataFreshness catches delivery-failed events and any other delivery stall.

ErrorOutputPrefix configuration

Without a separate ErrorOutputPrefix, Firehose writes error records to the same S3 prefix as successful deliveries. For a Parquet-formatted table, this means non-Parquet error payloads (raw JSON from processing-failed records) land in the Parquet prefix and cause every Athena query to fail with a deserialization error on those objects. Always set ErrorOutputPrefix explicitly:

ExtendedS3DestinationConfiguration: {
  // Success prefix — Hive-compatible
  Prefix: 'probe-results/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/',

  // Error prefix — separate per error type, also Hive-compatible for queryability
  ErrorOutputPrefix: 'probe-errors/type=!{firehose:error-output-type}/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/',

  // ... buffer hints, format conversion, etc.
}

Error records in the error prefix are the original source records before transformation, base64-decoded, in the original format the producer sent. For AliveMCP probe results, this means the raw JSON probe result is preserved in full — you can inspect it directly from S3, fix the Lambda, and replay it.

Critical CloudWatch alarms

The DataFreshness alarm is the most important. It measures the age of the oldest record not yet delivered to S3 — when delivery stalls, the oldest undelivered record gets older every second, and the metric spikes from its normal value (near the buffer interval) to thousands of seconds. Set the threshold to buffer_interval + 120: for a 60-second buffer, alarm at >180 seconds; for a 900-second buffer, alarm at >1,020 seconds. Configure TreatMissingData: notBreaching — a stream with no records arriving reports no DataFreshness metric, which is not a delivery failure.

// CloudFormation: DataFreshness alarm for 60-second buffer delivery stream
{
  "Type": "AWS::CloudWatch::Alarm",
  "Properties": {
    "AlarmName": "alivemcp-firehose-delivery-stalled",
    "Namespace": "AWS/Firehose",
    "MetricName": "DeliveryToS3.DataFreshness",
    "Dimensions": [{ "Name": "DeliveryStreamName", "Value": "alivemcp-probe-results" }],
    "Statistic": "Maximum",
    "Period": 60,
    "EvaluationPeriods": 3,
    "Threshold": 180,
    "ComparisonOperator": "GreaterThanThreshold",
    "TreatMissingData": "notBreaching",
    "AlarmActions": ["arn:aws:sns:us-east-1:123456789012:alivemcp-ops-alerts"]
  }
}

Backup S3: FailedDataOnly as the safety net

The backup S3 destination captures source records before they enter the transformation and format conversion pipeline. FailedDataOnly captures only records that fail transformation or delivery — providing a guaranteed recovery path without doubling storage costs:

S3BackupMode: 'FailedDataOnly',  // 'Disabled' | 'FailedDataOnly' | 'Enabled' (all records)
S3BackupConfiguration: {
  BucketARN: 'arn:aws:s3:::alivemcp-probe-backup',
  RoleARN: 'arn:aws:iam::123456789012:role/firehose-delivery-role',
  Prefix: 'raw-backup/!{timestamp:yyyy/MM/dd/HH}/',
  BufferingHints: { SizeInMBs: 128, IntervalInSeconds: 900 },  // Backup is a safety net, not a query source
  CompressionFormat: 'GZIP',
}

The Enabled (all records) mode doubles storage costs and is appropriate only during initial pipeline validation or when performing a live schema migration where a guaranteed copy of every record is required to reconstruct the archive after a format conversion failure. In steady-state production, FailedDataOnly provides the recovery path at roughly 1–5% of the storage cost of Enabled, assuming a well-functioning Lambda that processes the vast majority of records successfully.

Replaying failed records from the error prefix

When a Lambda transformation bug routes records to processing-failed/, the recovery procedure is: fix the Lambda, deploy the fix, then replay the failed records through the corrected pipeline:

import { FirehoseClient, PutRecordBatchCommand } from '@aws-sdk/client-firehose';
import { S3Client, GetObjectCommand, ListObjectsV2Command } from '@aws-sdk/client-s3';

async function replayFailedRecords(
  bucket: string,
  prefix: string,    // e.g., 'probe-errors/type=processing-failed/year=2026/month=10/day=30/'
  deliveryStreamName: string
): Promise<void> {
  const s3 = new S3Client({ region: 'us-east-1' });
  const firehose = new FirehoseClient({ region: 'us-east-1' });

  const list = await s3.send(new ListObjectsV2Command({ Bucket: bucket, Prefix: prefix }));

  for (const obj of list.Contents ?? []) {
    const get = await s3.send(new GetObjectCommand({ Bucket: bucket, Key: obj.Key! }));
    const body = await get.Body!.transformToString('utf-8');

    // Error prefix records are NDJSON — one original record per line
    const records = body.trim().split('\n').map(line => ({
      Data: Buffer.from(line, 'utf-8'),
    }));

    // Replay in batches of 500 (PutRecordBatch limit)
    for (let i = 0; i < records.length; i += 500) {
      const batch = records.slice(i, i + 500);
      await firehose.send(new PutRecordBatchCommand({
        DeliveryStreamName: deliveryStreamName,
        Records: batch,
      }));
      // Stay below 1,000 records/second Firehose throttle
      await new Promise(r => setTimeout(r, 500));
    }
  }
}

Replay considerations: replayed records receive a new approximateArrivalTimestamp (the replay time, not the original probe time). Ensure downstream Athena queries use probedAt from the record payload as the event time, not the Firehose arrival timestamp. If the error prefix contains duplicate records from multiple retry attempts before routing to error, deduplicate on serverSlug + probedAt before replaying.

Cross-pattern failure modes reference

Putting it together: a complete AliveMCP Firehose pipeline

A production-grade Firehose pipeline for AliveMCP probe results combines all four patterns: dynamic partitioning routes records to per-tenant S3 prefixes using JQ; a Lambda transformation normalizes schema versions, filters test events, and ensures the base64/newline contract; S3 delivery is configured for Parquet output via Glue Data Catalog with Hive-compatible prefixes; and four CloudWatch alarms cover the full failure surface.

The patterns interact at two points. First, format conversion (Parquet via Glue) is incompatible with dynamic partitioning inline MetadataExtraction — the two features cannot coexist in the same processor chain. The resolution is to use the Lambda transformation for both: the Lambda extracts the partition key field as a top-level normalized field in the JSON output (e.g., adding a tier field based on isPaid), and a separate MetadataExtraction processor reads the pre-extracted field. Second, the DataFreshness alarm threshold must account for format conversion latency: with a 60-second buffer interval, format conversion adds ~5 seconds, so set the threshold to 185+ seconds rather than 180.

For teams building the KDS-to-Firehose pipeline described in the KDS Advanced arc — where KDS provides the source stream, Enhanced Fan-Out feeds the batch archiver, and Lambda ESM feeds the real-time alerter — Firehose sits between KDS and S3, receiving records from the KDS batch archiver consumer via the Kinesis-sourced Firehose delivery stream. The archiver consumer writes records to KDS; Firehose reads from KDS, transforms via Lambda, converts to Parquet, and delivers to S3 with per-tenant partitioning. CloudWatch alarms across both the KDS stream (IteratorAgeMilliseconds, WriteProvisionedThroughputExceeded) and the Firehose delivery stream (DataFreshness, ProcessingFailedRecords) give complete pipeline visibility with a single-page alarm dashboard.