enrich-kinesis

[benjben]

• enrich-pubsub: add reference.conf and provide minimal config example #505
• Add Github Action to scan Docker images with lacework #506
• common: Use schemas/nl.basjes/yauaa_context/jsonschema/1-0-3 #515
• Publish arm64 and amd64 docker images #491 Publish arm64 and amd64 docker images #492
• Enable ES6 by default in javascript enrichment #508 Enable ES6 by default in javascript enrichment (close #508) #509
• Beam Enrich: deprecate #530
• fix initialTimestamp
• Pubsub: use custom fs2-google-pubsub-grpc dependency #487 Do we wait for permutive release ? => 1) test throughput with Emit larger chunks in HTTP and gRPC consumers permutive-engineering/fs2-pubsub#330 2) if not successful test with our own 0.18.0.1-M3
• common: catch and handle errors in the CurrencyConversionEnrichment #542
• Validate enriched event against atomic schema before emitting #517 Enforce maximum field lengths (close #517) #521 => ⚠️ some fields in EnrichedEvent don't match the type in atomic schema (e.g. here and here)
• Add possibility to specify maxConcurrency and bufferSize and maybe others
• Enable AWS MSK IAM Authentication #544 and add % Runtime to sts
• make validation latest configurable
• enrich-kinesis: add integration test #531
• go over each change and think about what could break something

[istreeter]

I've only given this a quick look, but it looks very good indeed 🎉

[spenes]

Looks good to me in general 👍 Left a few small comments in there.

[chuwy]

Just noticed there's a few weeks old unsubmitted review from me, but most of the stuff is addressed already. Just one nit has left. Otherwise - looks great!

[voropaevp]

io is shadowing namespace of dependencies like io.circe. This confuses IDE when importing io.* in test.
I suggest changing it to inout or some other name that is not used as a top-level domain.

[istreeter]

How many collector payloads do you think it's possible to be holding in memory at the same time with these default settings? If you're processing, say, 5 kinesis shards per enrich instance.

I count 190,000 (admittedly a worse case scenario). And a further 10,000 more per additional shard.

Here's how:

Each polling batch is 10000 records. The fs2-aws library uses a buffer that holds 10 batches. Plus there can be extra batch per shard on this line waiting to enqueue.

Then you have the batch you're currently processing, plus three extra batches because of the prefetches here, here and here

This is just to explain why I've been questioning memory problems, especially for large collector payloads. Understanding the scope of this problem will be difficult, because we would need to explore regular processing, and also error-handling scenarios, e.g. kinesis problems, enrichment problems. The latter are harder to predict or simulate.

[benjben]

Very interesting!

I agree with your way of counting. There has been only a little change, I removed the .prefetch to now use >1 for sink.concurrency here so now it's concurrency - 1 additional shards instead of three.

Because of the enqueuing for each shard I'm afraid the memory used by the app is unbounded. So we need to make sure that our auto-scaling is good enough so that new instances are spawned before the app gets a chance to go OutOfMemory. I'll look at our scaling strategy in Terraform. But I think that whatever the strategy, it would be great to use the average record size of a customer to determine the memory allocated to the JVM. /cc @oguzhanunlu @jbeemster