feat(cdp): Improved ingestion consumer event runner #27671
+1,329
−29
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# Conflicts: # plugin-server/src/main/pluginsServer.ts
There was a problem hiding this comment.
PR Summary
This PR implements a major refactor of the event processing pipeline, introducing a new EventPipelineRunnerV2 class that consolidates previously scattered logic into a more maintainable structure.
- Added new
EventPipelineRunnerV2class in/plugin-server/src/ingestion/event-pipeline-runner/event-pipeline-runner.tswith clear step-by-step processing - Created utility modules for timestamp handling (
timestamp-utils.ts), heatmap processing (heatmaps.ts), and element chain extraction (event-utils.ts) - Implemented comparison mode via
INGESTION_CONSUMER_V2_COMPARISON_PERCENTAGEconfig to safely validate the new implementation against the old one - Added comprehensive error handling with custom
EventDroppedErrorclass for expected failures - Improved AI event processing with better error handling and logging via structured logger
14 file(s) reviewed, 9 comment(s)
Edit PR Review Bot Settings | Greptile
plugin-server/src/cdp/hog-transformations/hog-transformer.service.ts
Outdated
Show resolved
Hide resolved
plugin-server/src/ingestion/event-pipeline-runner/event-pipeline-runner.test.ts
Outdated
Show resolved
Hide resolved
plugin-server/src/ingestion/event-pipeline-runner/event-pipeline-runner.ts
Outdated
Show resolved
Hide resolved
plugin-server/src/ingestion/event-pipeline-runner/event-pipeline-runner.ts
Outdated
Show resolved
Hide resolved
plugin-server/src/ingestion/event-pipeline-runner/utils/timestamp-utils.ts
Show resolved
Hide resolved
…ne-runner.ts Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>
# Conflicts: # plugin-server/src/cdp/hog-transformations/hog-transformer.service.ts # plugin-server/src/ingestion/ingestion-consumer.test.ts # plugin-server/src/ingestion/ingestion-consumer.ts # plugin-server/tests/main/ingestion-queues/each-batch.test.ts
eli-r-ph
approved these changes
Apr 16, 2025
# Conflicts: # plugin-server/src/ingestion/ingestion-consumer.ts
Comment on lines
+579
to
+596
| private async runEventRunnerV2(incomingEvent: IncomingEvent): Promise<RawKafkaEvent | undefined> { | ||
| const runner = this.getEventPipelineRunnerV2(incomingEvent.event) | ||
|
|
||
| try { | ||
| return await runner.run() | ||
| } catch (error) { | ||
| // NOTE: If we error at this point we want to handle it gracefully and continue to process the scheduled promises | ||
| await this.handleProcessingErrorV2(error, incomingEvent.message, incomingEvent.event) | ||
| } | ||
|
|
||
| runner?.getPromises().forEach((promise) => { | ||
| // Schedule each promise with their own error handling | ||
| // That way if all fail with ignoreable errors we continue but if any one fails with an unexpected error we can crash out | ||
| this.scheduleWork(promise).catch((error) => { | ||
| return this.handleProcessingErrorV2(error, incomingEvent.message, incomingEvent.event) | ||
| }) | ||
| }) | ||
| } |
There was a problem hiding this comment.
It appears the runEventRunnerV2 method doesn't return a value in all code paths. The method tries to return the result of runner.run(), but if an error is caught, it handles the error without returning anything. After error handling, the code processes promises but doesn't return a value.
Consider adding an explicit return statement or restructuring the method to ensure it consistently returns a value of type RawKafkaEvent | undefined in all code paths:
try {
return await runner.run();
} catch (error) {
await this.handleProcessingErrorV2(error, incomingEvent.message, incomingEvent.event);
return undefined; // Explicit return after error handling
}Spotted by Diamond
Is this helpful? React 👍 or 👎 to let us know.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Problem
The runner code is a nightmare. I wanted to refactor this for a while now but given we have a new ingestion consumer this felt like the right time as we can run both codebases in parallel
Changes
👉 Stay up-to-date with PostHog coding conventions for a smoother review.
Does this work well for both Cloud and self-hosted?
How did you test this code?
Adds tests to cover: