Consume from the earliest offset where a lost message is unrecoverable - #2794
Consume from the earliest offset where a lost message is unrecoverable#2794delthas wants to merge 1 commit into
Conversation
Hello delthas,My role is to assist you with the merge of this Available options
Available commands
Status report is not available. |
Codecov Report❌ Patch coverage is
❌ Your patch check has failed because the patch coverage (50.00%) is below the target coverage (80.00%). You can increase the patch coverage or adjust the target coverage. Additional details and impacted files
... and 4 files with indirect coverage changes
@@ Coverage Diff @@
## development/9.5 #2794 +/- ##
===================================================
- Coverage 75.50% 75.47% -0.03%
===================================================
Files 201 201
Lines 13928 13930 +2
===================================================
- Hits 10516 10514 -2
- Misses 3402 3406 +4
Partials 10 10
Flags with carried forward coverage won't be shown. Click here to find out more. 🚀 New features to boost your workflow:
|
Waiting for approvalThe following approvals are needed before I can proceed with the merge:
|
46741ad to
e93986d
Compare
SylvainSenechal
left a comment
SylvainSenechal
left a comment
There was a problem hiding this comment.
Approved but this looks again like quite a big oversight (maybe its mostly fixing issues with ci as if I understand correctly, 'earliest' will only apply to never commited partition which we have plenty in tests ci but probably not many in prod 🤔 )
You may wanna wait a few days for François to take a look though as he is back soon
Also, while its probably a bit annoying to do, I think this would really deserve a functional ci test in backbeat where we setup kafka, start a backbeat consumer then kill it and restart it, while manually producing a message in kafka in between, to make sure the behavior is correct 🤔
|
Agree this is mainly for CI, though this can happen when creating locations in production. For functional tests we already have functional tests for |
|
Waiting for another reviewer :) |
DarkIsDude
left a comment
DarkIsDude
left a comment
There was a problem hiding this comment.
Nothing for me. It's clear enough.
But I have a weird feeling that I can't explain: like we did a fix but it's the right one, something else must be done. I don't know why, I don't know what. Maybe something when we create a location for example 🤷. Anyway, code looks right. For example:
- backbeat-data-mover is one shared topic, but the group is ${groupId}-${this.site}, and every site filters by toLocation === this.site. Adding a location at runtime spawns a fresh QueueProcessor, hence a fresh group with no committed offsets — which under earliest now starts at offset 0 of the entire retained log.
- extensions/notification/queueProcessor/QueueProcessor.js : Same. Group is ${groupId}-${destinationId}; when internalTopic is unset, all destinations share one topic. Adding a destination creates a new group on that old topic, and processKafkaEntry filters by destination ID, prefix, suffix, and event type — never by time. So the full retention window of a bucket's events gets delivered to the new destination as if fresh, breaking the S3 guarantee that only post-configuration events are notified.
Also this can have an impact on Ring, maybe better to check with them and asking for a review ?
Only the replication status consumer pinned fromOffset (BB-826); every other BackbeatConsumer inherited librdkafka's default `latest`, so a message produced onto a partition with no committed offset while the consumer group had no member is skipped forever. Pin `earliest` on the three consumers where that loss is permanent and the group is not created per site or per destination: the lifecycle transition tasks topic, the per-location cold status topics and the gc topic. The lifecycle object (expiration) tasks topic, the data mover topic and the notification internal topic deliberately stay on the default -- see BB-831 for the per-consumer rationale and the replay-safety analysis behind each decision. Extract the gc consumer options into _getConsumerOptions() for unit coverage, following #2788. Issue: BB-831
e93986d to
891f7af
Compare
|
Thanks. I'm dropping notification and data-mover from this PR to keep it low-risk. They're not affecting CI anyway (no notification issues in the Zenko run census, and no data-mover tests). |
|
So only backbeat-lifecycle-transition-tasks, cold-status-, gc remain. Those should be safe. |
4 participants
Pins
fromOffset: 'earliest'on the three consumers where a message produced during a consumer group gap is permanently lost and the group is service-wide rather than created per site or per destination:backbeat-lifecycle-transition-taskscold-status-<location>GarbageCollectorFour consumers deliberately stay on the default: the lifecycle object (expiration) tasks topic (self-healing via the conductor's re-enumeration), the notification internal topic and the data-mover topic (both use a per-instance group id on a shared topic, so a new destination or location would replay the full retention window), and the replication topic itself (replay cost needs its own decision). A unit test pins the expiration exclusion so it is not "fixed" by accident.
The per-consumer rationale — why each loss is or is not recoverable, and the replay-safety analysis behind each decision — is in BB-831.
Validation
On a 30-run census with an earlier build of this fix, the Azure Archive location CRUD scenario went from 8/20 runs failing (40 %) to 1/30 (3.3 %), and no skip of this kind appears in any of the dumps inspected. Pre-fix, cold-status messages produced but never consumed ran at 15–34 % across two censuses, concentrated in the rollout window and absent outside it, and the mechanism was traced end to end in every case examined — all of them cold-status. Caveats: an unrelated fix touching cold transitions landed between censuses, and the validated build's consumer set is wider than what this PR now pins, so the attribution rests on the traced mechanism rather than on the rate alone.
Implementation note
The gc consumer options move into
_getConsumerOptions()for unit coverage, following #2788. GenericfromOffsetsemantics are already covered by the functional tests #2788 added to theBackbeatConsumersuite.Issue: BB-831