feat(aws_kinesis): aggregate records, stop dropping partial failures

[smolaon]

Summary

• Kinesis bills every record rounded up to 1 KB and a per-record compression frame restarts the compressor each time, so small records waste most of what they are billed and cross-event redundancy -- enormous in security event streams -- is never exploited. Packing many events into one record as newline-delimited JSON and compressing them as a single unit takes measured billed bytes to 31% of today's on real CloudTrail traffic (4.99x compression vs 1.98x). Off by default.
• Also fixes a live silent data loss, independent of aggregation: PutRecords returns HTTP 200 with per-record errors, but event_status reported Delivered unconditionally, so a source with acknowledgements enabled deleted its message for records that were never written. request_retry_partial defaults to off, so they were not retried either. Applies to firehose too, which shares the response type.
• The loss is what makes the ordering matter: aggregation multiplies it by the number of events per record, so the fix belongs in front of any rollout.

Tests: cargo check --tests clean across streams + firehose + the integration-test feature (src/lib.rs has #![deny(warnings)]), cargo test aws_kinesis 9/9 passed, plus two end-to-end probes against a locally built binary.
Screenshots: n/a -- no visible change.

Details

A consumer must decompress and then split on newlines, and the rollout order is not symmetric.
An unaggregated record is a single-line payload, so a splitting consumer handles both formats and a
stream can carry both during a rollout. The reverse is not true: an old consumer receiving a
multi-line record fails to parse it, and depending on its error handling that can stall a shard
rather than skip a record. Deploy the consumer change everywhere before enabling this on any
producer.

Errored, not Rejected, is the correct status for a partial failure -- and the difference is
load-bearing. A per-record ProvisionedThroughputExceededException is transient, and only Errored
maps to BatchStatus::Errored, which the aws_s3 source turns into
ProcessingError::ErrorAcknowledgement and therefore leaves its SQS message in the queue.
Rejected would be deleted instead: delete_failed_message defaults to true
(src/sources/aws_s3/sqs.rs), so nacking as Rejected would have preserved the bug.

request_retry_partial is deliberately left defaulting to off here. Turning it on is the right
operational setting, but it is shared with firehose, whose should_retry_response retries the
whole request rather than the failed indices -- so flipping the shared default would start
duplicating successfully-written firehose records. Set it per sink in the config instead; the status
fix is the backstop that makes the default safe either way.

encoding.json.pretty is rejected under aggregation, not just non-JSON codecs. Newline framing
is only sound if no event can contain a literal newline, and compact JSON escapes them as the two
characters \n. to_writer_pretty emits real ones, which would split one event into several
unparseable fragments. The guard reads EncodingConfig::config() rather than the built Serializer,
because the built form does not expose the flag. It is scoped to aggregation, so pretty with
aggregation off still builds.

partition_key_field is ignored when aggregation is on, with a warn! rather than an error
(streams/config.rs). Events in one record may disagree on the field, so a random UUID is generated
per record, which keeps the MD5 hash distribution -- and therefore shard distribution -- uniform. A
warning rather than a hard failure because the option has no correctness impact here, and erroring
would crash-loop a sink on config that was previously valid.

aggregation.max_bytes bounds the uncompressed input (default 256 KiB, capped at 900,000).
The compressed size is not known until the batch closes and RequestBuilder emits exactly one
request per batch, so staying well under the 1 MB record limit means even wholly incompressible
input still fits.

Its own config table, not batch. batch is hard-capped to the PutRecords limits and its
units are records per API call; these are events per record. Reusing it would silently reinterpret
500 / 5 MB.

Implementation Plan

Format: NDJSON inside the existing zstd frame, not KPL. The wire format stays "a zstd frame", so
the consumer's codec detection needs no change -- it decompresses as today and then splits on
newlines. KPL would need protobuf plus an MD5 trailer, and on the Go side
awslabs/kinesis-aggregation fans out across ~75 go.mod files. NDJSON needs zero new deps on
either side, and we control both ends -- there is no interop requirement. It also reuses upstream
machinery wholesale: aws_s3 already does batch -> newline framing -> compress once.

Shape:

events
  |- inner batcher   -> Vec<Event>   (one record; max_events / max_bytes / timeout_secs)
  |- request builder -> NDJSON encode + compress ONCE
  |- outer batcher   -> Vec<record>  (one PutRecords; unchanged 500 / 5 MB)

The existing builder is RequestBuilder<KinesisProcessedEvent> with type Events = Event, and
RequestBuilder::encode_events constructs a fresh Compressor per call -- which is exactly why
every record is its own frame today. Taking Vec<Event> is the whole mechanism; it reuses the
upstream impl Encoder<Vec<Event>> for (Transformer, Encoder<Framer>) that aws_s3 relies on. Note
the encoder type changes too: the existing builder uses the unframed Encoder<()>, which has no
framer and hence no newline suffix. That is why a separate builder is cleaner than mutating the
shared one.

Everything downstream of the request builder is untouched: one aggregate is one KinesisRequest, so
the index-based partial-failure retry in KinesisRetryLogic keeps working, and finalizers are merged
via Vec::take_finalizers so an ack or nack applies to every event in the record. Acknowledgement
granularity is unchanged by aggregation, incidentally -- the driver already takes finalizers off the
whole BatchKinesisRequest before the call and stamps one status across all of them, so up to 500
records always shared one ack.

#	Layer	File	Change
1	Config	streams/aggregation/config.rs	KinesisAggregationConfig, BatcherSettings built directly (not via BatchConfig, which is hard-capped to PutRecords limits)
2	Encode	streams/aggregation/request_builder.rs	Events = Vec<Event>, Encoder<Framer>, merged finalizers, one UUID key per record
3	Pipeline	streams/aggregation/sink.rs	Two batching layers around the request builder
4	Build	streams/aggregation/build.rs	Newline-delimited encoder plus the compact-JSON-only serializer guard
5	Wiring	streams/config.rs, streams/mod.rs, request_builder.rs	aggregation field, build branch, KinesisRequest::new so metadata stays private
6	Fix	service.rs	event_status reports Errored when failure_count > 0, with unit tests

Not changed: record.rs (an aggregate is just a bigger Bytes), the base config.rs,
lib/codecs/*, Cargo.toml. Zero new crates.

Why not a codec-level BatchSerializer: that abstraction is for serializers producing one opaque
blob with no per-event framing (parquet, arrow). NDJSON is per-event serialize plus a framer, which
Encoder<Framer> already is. Going through lib/codecs would cross a crate boundary and mutate the
global serializer enum for every sink.

Measured, on 400 real CloudTrail events (3,680 B/event raw):

	B/event	Ratio	Billed B/event	vs today
Per-record (today)	1,858	1.98x	2,399	100%
Aggregated, 50/record	738	4.99x	748	31%
Aggregated, 100/record	702	5.24x	709	30%

Diminishing returns justify the 50-ish default: a larger record buys single digits while
concentrating bytes against the 1 MB/s/shard write cap.

Verification. End to end against a locally built binary and a fake Kinesis endpoint, 60 events
in:

	Records	Wire	Decoded	Lines/record
aggregation off	60	~300 B	492 B	1
aggregation on	1	1,078 B	29,717 B	60

Event bytes are 29,657 in both cases, and the events are byte-identical in frame order once the
stdin source's per-run host and timestamp are excluded. The decoded delta is exactly +60 for 60
events -- pure newline framing, no content change -- which confirms the encoding transformer still
applies per event, as it must.

A second probe drives the serializer guard through six configs: compact json and native_json
build; json with pretty: true, text, and raw_message are refused with the reason named; and
pretty with aggregation off still builds, confirming the guard cannot break an existing config.