Optimize Kafka queue offset handling - #123
Conversation
mariusandra
left a comment
There was a problem hiding this comment.
I feel like this code is way too complex for what it should do. It also leaves the kafka_offset inside event.
I tried another way to get the same result with #124 ... though I haven't run the code yet.
| import { Kafka, Consumer, Message, EachBatchPayload } from 'kafkajs' | ||
| import { PluginsServer, Queue, RawEventMessage } from 'types' | ||
| import { KAFKA_EVENTS_INGESTION_HANDOFF } from './topics' | ||
| import { KAFKA_EVENTS_WAL } from './topics' |
There was a problem hiding this comment.
Why the change from KAFKA_EVENTS_INGESTION_HANDOFF?
|
Oh, sry, actually I was doing some debugging due to tests failing, so this is not in a reviewable state ATM (was reviewable at 6feda22).
|
|
Which cases would #124 not cover? |
|
just saw your reply :), will give it a thought after dinner |
| const singleIngestionTimer = new Date() | ||
| await this.saveEvent(event) | ||
| resolveOffset(event.kafka_offset!) | ||
| resolveOffset(eventUuidToKafkaOffset.get(event.uuid!)!) |
There was a problem hiding this comment.
TODO: don't run this if the event was not present in batch from Kafka (meaning it was inserted by a plugin's processEventBatch)
Changes
A way to optimize offsets and to avoid weird situations.
Checklist