Feat: add support of XPENDING #2190
Conversation
| } | ||
|
|
||
| StreamConsumerMetadata consumer_metadata = decodeStreamConsumerMetadataValue(get_consumer_value); | ||
| consumer_metadata.pending_number -= 1; |
There was a problem hiding this comment.
@Yangsx-1 , It seems that we should also update pending_number in StreamConsumerMetadata, when xack id, currently, we only update pending_number in StreamConsumerGroupMetadata.
There was a problem hiding this comment.
Good catch! You're right.
There was a problem hiding this comment.
Has this patch about add support of XPENDING?
If not can you move the change to a new PR? @LiuYuHui
| } | ||
|
|
||
| StreamConsumerMetadata consumer_metadata = decodeStreamConsumerMetadataValue(get_consumer_value); | ||
| consumer_metadata.pending_number -= 1; |
There was a problem hiding this comment.
Good catch! You're right.
| StreamConsumerMetadata consumer_metadata = decodeStreamConsumerMetadataValue(get_consumer_value); | ||
| consumer_metadata.pending_number -= 1; | ||
|
|
||
| batch->Put(stream_cf_handle_, consumer_key, encodeStreamConsumerMetadataValue(consumer_metadata)); | ||
| batch->Delete(stream_cf_handle_, entry_key); |
There was a problem hiding this comment.
I think this line should be put after *acknowledged += 1 , because the entry_key may not be delete due to the early return.
| return s; | ||
| } | ||
| if (s.IsNotFound()) { | ||
| return rocksdb::Status::OK(); |
There was a problem hiding this comment.
I don't think it's a return here. The above line is the same. It should not affect other consumers and the final result.
| const std::string &group_name = options.group_name; | ||
| std::string ns_key = AppendNamespacePrefix(stream_name); | ||
|
|
||
| LockGuard guard(storage_->GetLockManager(), ns_key); |
| rocksdb::Status GetPendingEntries(StreamPendingOptions &options, StreamGetPendingEntryResult &pending_infos, | ||
| std::vector<StreamGetExtPendingEntryResult> &ext_results); |
There was a problem hiding this comment.
options here can be const.
The signature of this function is different from what we are accustomed to.
- We are accustomed to using pointers instead of references to represent return values.
- In general, we usually have only one return value.
- We usually place the key name(here is stream_name), group_name, consumer_name outside instead of inside the options.
| @@ -23,11 +23,14 @@ | |||
| #include <rocksdb/status.h> | |||
|
|
|||
| #include <memory> | |||
| #include <unordered_set> | |||
| } | ||
|
|
||
| StreamConsumerMetadata consumer_metadata = decodeStreamConsumerMetadataValue(get_consumer_value); | ||
| consumer_metadata.pending_number -= 1; |
There was a problem hiding this comment.
Has this patch about add support of XPENDING?
If not can you move the change to a new PR? @LiuYuHui
|
|
@LiuYuHui Hi, how's it going now? |
closes #1734