[PP] Handle IngestRequest message
#3974
Conversation
muhamadazmy
force-pushed
the
pr3973
branch
2 times, most recently
from
c093ea2 to
2a7342c
Compare
muhamadazmy
force-pushed
the
pr3973
branch
6 times, most recently
from
8b8db7f to
953b39a
Compare
muhamadazmy
force-pushed
the
pr3973
branch
5 times, most recently
from
966834c to
8ae259e
Compare
muhamadazmy
force-pushed
the
pr3973
branch
2 times, most recently
from
fc8efad to
5b6ea28
Compare
tillrohrmann
left a comment
tillrohrmann
left a comment
There was a problem hiding this comment.
Thanks a lot for creating this PR @muhamadazmy. The changes make a lot of sense to me. I left a few minor comments.
My main question was about the impact on the partition processor event loop if we are ingesting a lot of entries (potentially also large entries). Did you observe any negative effects?
It would be great to address the problem of deserializing the Envelope records in a follow-up PR since this is really unnecessary work that the system now needs to do.
| // sender | ||
| // .enqueue_many(records) | ||
| // .await | ||
| // .map_err(|_| Error::SelfProposer)?; |
There was a problem hiding this comment.
minor optimization: We might be able to use sender.try_enqueue_many and only fall back to enqueue individual records if the try one fails.
In the future we might have a LogSender::enqueue_many_optimized which checks for the available capacity and obtains as many permits as possible in a batch. But this is probably premature optimization.
| // is a way to pass the raw encoded data directly to the appender | ||
| let envelope = StorageCodec::decode(&mut record.record)?; |
There was a problem hiding this comment.
This is indeed a pity. Let's do this as a follow-up. I think there isn't a lot missing because Record already supports carrying Bytes and we only need a way to create a Record from (Keys, Bytes) that we have available here. This is literally wasted work that we are doing here.
| .propose_many_with_callback(records, callback) | ||
| .await; |
There was a problem hiding this comment.
Can this await become a problem for the partition processor event loop if there are too many records to ingest? Should we maybe think about a throttling mechanism to not starve the other select branches?
There was a problem hiding this comment.
Good point! I will look into it
There was a problem hiding this comment.
This is now mainly controlled by the ingestion client "batch_size" (in bytes) This put an upper limit on the size of a single ingest request. Default is 50kB.
muhamadazmy
force-pushed
the
pr3973
branch
13 times, most recently
from
a16b71e to
ee27f2d
Compare
There was a problem hiding this comment.
Thanks for updating this PR @muhamadazmy. I think we shouldn't map ServiceStopped to LostLeadership as this has a slightly different semantical meaning than before. Instead mapping it to NotLeader should be fine, I believe. Apart from this, +1 for merging :-)
It's really nice that we no longer need to do the deserialization serialization steps when handling the Envelope 👏
muhamadazmy
force-pushed
the
pr3973
branch
7 times, most recently
from
0cfe4d9 to
f188c5d
Compare
- `ingestion-client` implements the runtime layer that receives WAL envelopes, fans it out to the correct partition, and tracks completion. It exposes: - `IngestionClient`, enforces inflight budgets, and resolves partition IDs before sending work downstream. - The session subsystem that batches `IngestRecords`, retries connections, and reports commit status to callers. - `ingestion-client` only ingests records and notify the caller once the record is "committed" to bifrost by the PP. This makes it useful to implement kafka ingress and other external ingestion
Summary: Handle the incoming `IngestRequest` messages sent by the `ingestion-client`
2 participants
[PP] Handle
IngestRequestmessageSummary:
Handle the incoming
IngestRequestmessages sent by theingestion-clientStack created with Sapling. Best reviewed with ReviewStack.
Shuffler#4024IngestionClientfor invocation and state mgmt #3980ingestion-client#3975IngestRequestmessage #3974CommitTokenback from notify_committed() #3968ingestion-clientcrate #3976