Ractor-based deserialization engine #2938
Draft
+1,249
−291
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
mensfeld
added
the
enhancement
Contributor
There was a problem hiding this comment.
Pull request overview
This PR introduces a Ractor-based parallel deserialization engine for Karafka, targeting Ruby 4.0+. It renames the Deserializers feature to Deserializing throughout the codebase while maintaining backward compatibility through aliases. The implementation adds configurable parallel deserialization that dispatches work to a Ractor pool when message batches meet defined thresholds.
Key Changes:
- Renamed
DeserializerstoDeserializingwith backward-compatible aliases - Added parallel deserialization infrastructure using Ractor workers with configurable pool size, batch thresholds, and payload size thresholds
- Implemented early dispatch mechanism in executor to allow Ractors to work while jobs wait in the queue
Reviewed changes
Copilot reviewed 41 out of 45 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
lib/karafka/routing/features/deserializing.rb |
Renamed feature class from Deserializers to Deserializing, added post_setup to initialize Ractor pool |
lib/karafka/routing/features/deserializing/topic.rb |
Updated topic routing API to use deserializing method with backward-compatible deserializers alias |
lib/karafka/routing/features/deserializing/config.rb |
Added parallel field to Config struct and parallel? method for checking if parallel deserialization is enabled |
lib/karafka/routing/features/deserializing/contracts/topic.rb |
Updated validation contract to require parallel boolean field |
lib/karafka/deserializing/parallel/pool.rb |
Implements Ractor worker pool for parallel deserialization with work distribution via ports |
lib/karafka/deserializing/parallel/future.rb |
Represents pending async deserialization, collects results from Ractor workers |
lib/karafka/deserializing/parallel/immediate.rb |
No-op implementation when parallel deserialization is skipped |
lib/karafka/deserializing/parallel/injector.rb |
Injects pre-deserialized payloads back into Message objects |
lib/karafka/deserializing/deserializers/*.rb |
Moved deserializers from Karafka::Deserializers to Karafka::Deserializing::Deserializers with Base class that freezes instances |
lib/karafka/setup/config.rb |
Added deserializing.parallel configuration namespace with pool settings |
lib/karafka/setup/contracts/config.rb |
Added validation rules for parallel deserialization configuration |
lib/karafka/processing/executor.rb |
Added early dispatch logic in before_schedule_consume to send work to Ractor pool |
lib/karafka/processing/strategies/default.rb |
Added result retrieval and injection in handle_before_consume |
lib/karafka/messages/message.rb |
Added payload= setter for injecting pre-deserialized payloads |
lib/karafka/messages/batch_metadata.rb |
Added deserialization field to store Future/Immediate objects |
lib/karafka/messages/builders/batch_metadata.rb |
Initialize deserialization field with Immediate instance by default |
config/locales/errors.yml |
Updated error messages from deserializers to deserializing, added parallel config validation errors |
spec/**/* |
Updated all spec files to use new Deserializing namespace and added comprehensive tests for parallel components |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
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.
No description provided.