fix(stream): make consumer decrement pending number when message is acknowledged. #2352
Conversation
LindaSummer
force-pushed
the
bugfix/consumer_xack_pending
branch
from
266bbff to
566e419
Compare
LindaSummer
changed the title
|
Hi @torwig , Thanks for your suggestion. I have updated the related code. Please take a look. Thanks very much. |
|
CI failed is unrelated, rerun |
|
Maybe I need to rebase on unstable branch rather than merge, it involved merge commit in this PR. |
We can squash and merge it in one patch, don't worry about that |
|
Thanks very much for your warm help. 😊 The merge commit seems to make coverage calculated more code blocks than my commits. Do I need to solve it with rebase in this PR to re-trigger the coverage workflow? |
src/types/redis_stream.cc
Outdated
| } | ||
| } | ||
| if (*acknowledged > 0) { | ||
| StreamConsumerGroupMetadata group_metadata = decodeStreamConsumerGroupMetadataValue(get_group_value); | ||
| group_metadata.pending_number -= *acknowledged; | ||
| std::string group_value = encodeStreamConsumerGroupMetadataValue(group_metadata); | ||
| batch->Put(stream_cf_handle_, group_key, group_value); | ||
|
|
||
| for (const auto &consumer : consumer_acknowledges) { |
There was a problem hiding this comment.
| for (const auto &consumer : consumer_acknowledges) { | |
| for (const auto &[consumer_name, count] : consumer_acknowledges) { |
There was a problem hiding this comment.
This is reported by sonar, maybe we should fix that
There was a problem hiding this comment.
got it, I will update the loop with this syntax.
| auto consumer_meta_key = internalKeyFromConsumerName(ns_key, metadata, group_name, consumer.first); | ||
| std::string consumer_meta_original; | ||
| s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, consumer_meta_key, &consumer_meta_original); | ||
| if (s.ok()) { |
There was a problem hiding this comment.
what if !s.ok() && !s.IsNotFound()?
There was a problem hiding this comment.
Thanks very for your correction.
It should be an internal error and I think that I need to return the error. This function's return type is a rocksdb::Status so I don't need to wrap it to another format.
Do I need to add golang integration test for this error handling case? This may be hard to produce a case for it.
There was a problem hiding this comment.
Do I need to add golang integration test for this error handling case?
Currently it's a bit hard to modify this, especially in integration tests. We can just handle this without testing.
There was a problem hiding this comment.
Got it! I have corrected my wording above.
This function returns rocksdb::Status and I think I should return the status immediately in the function when !s.ok() && !s.IsNotFound().
There was a problem hiding this comment.
There're two gets here, right? if (s.ok()) in the loop should also adds this checks(Though errors merely happens)
There was a problem hiding this comment.
Thanks for your correction!
Sorry for missing another error handling. I have added it with a new commit.
LindaSummer
dismissed stale reviews from mapleFU, Yangsx-1, and torwig
via
777f504
LindaSummer
force-pushed
the
bugfix/consumer_xack_pending
branch
from
712a717 to
777f504
Compare
git-hulk
changed the title
|
Would merge if ci passes |
|
No worry about this flaky test failure, can rerun to avoid blocking the PR. |
|
Issue
close #2350
Proposed Changes
pending_numberwhen consumer group has acknowledged entry.XINFO.