feat(aws_kinesis_streams sink): add optional record aggregation

[smolaon]

Summary

Adds optional record aggregation to the aws_kinesis_streams sink: many events are packed into one Kinesis record as newline-delimited JSON and compressed as a single unit, instead of one event per record each compressed independently.

Two things drive it. Kinesis bills every record rounded up to 1 KB, so small records waste most of what they are billed. And a per-record compression frame restarts the compressor every time, so redundancy across events — which is enormous in security event streams — is never exploited.

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%

Off by default. With aggregation.enabled = false the sink takes its original code path unchanged.

Design

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 already relies on.

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.

New module under streams/aggregation/ with 1-3 line registration edits elsewhere, plus a KinesisRequest::new constructor so metadata stays private.

Deliberate choices

encoding.codec must be json or native_json; the build errors otherwise. Newline framing is only sound if no event can contain a literal newline. JSON escapes them as the two characters \n, so a raw 0x0A never appears inside a serialized event — but text or raw_message passes bytes through, and one embedded newline would silently split one event into two on the consumer.

partition_key_field is ignored when aggregation is on, with a warn! rather than an error. 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. It is 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 defaults to 256 KiB and is capped at 900,000. The limit bounds the uncompressed input, because the compressed size is not known until the batch closes and RequestBuilder emits exactly one request per batch. Staying well under the 1 MB record limit means even wholly incompressible input still fits. Diminishing returns justify the default: 50 events/record already captures 31% of billed bytes against 30% at 100 and 28% at 400, so a larger record buys single digits while concentrating bytes against the 1 MB/s/shard write cap.

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.

Verification

cargo check and cargo check --tests clean, including with aws-kinesis-streams-integration-tests enabled — note src/lib.rs has #![deny(warnings)], so that is a warning-free build. cargo test ... aws_kinesis: 4 passed, including generate_config, so the new field does not break config-schema generation.

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,065 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. So the encoding transformer still applies per event, as it must.

Consumer requirement

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