You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@kvrocks.apache.org by "Beihao-Zhou (via GitHub)" <gi...@apache.org> on 2024/03/26 17:04:26 UTC
[PR] Feat: add support of XCLAIM [kvrocks]
Beihao-Zhou opened a new pull request, #2202:
URL: https://github.com/apache/kvrocks/pull/2202
close #1733
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add support of the command XCLAIM [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk merged PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Beihao-Zhou (via GitHub)" <gi...@apache.org>.
Beihao-Zhou closed pull request #2202: Feat: add support of XCLAIM
URL: https://github.com/apache/kvrocks/pull/2202
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Beihao-Zhou (via GitHub)" <gi...@apache.org>.
Beihao-Zhou commented on PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#issuecomment-2106307740
Also wondering does anyone know why the linting check couldn't pass as shown with errors (https://github.com/Beihao-Zhou/kvrocks/actions/runs/9052637137/job/24870567712#step:7:1493).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add support of the command XCLAIM [kvrocks]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#issuecomment-2107125887
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2202) **Quality Gate passed**
Issues
 [19 New issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=2202&resolved=false&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2202&metric=new_accepted_issues&view=list)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_kvrocks&pullRequest=2202&resolved=false&sinceLeakPeriod=true)
 [53.9% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2202&metric=new_coverage&view=list)
 [2.7% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2202&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2202)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "PragmaTwice (via GitHub)" <gi...@apache.org>.
PragmaTwice commented on code in PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#discussion_r1546125697
##########
src/types/redis_stream.cc:
##########
@@ -357,6 +357,134 @@ rocksdb::Status Stream::DeletePelEntries(const Slice &stream_name, const std::st
return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
}
+rocksdb::Status Stream::ClaimPelEntries(const Slice &stream_name, const std::string &group_name,
+ const std::string &consumer_name, const uint64_t min_idle_time,
Review Comment:
```suggestion
const std::string &consumer_name, uint64_t min_idle_time,
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Beihao-Zhou (via GitHub)" <gi...@apache.org>.
Beihao-Zhou commented on PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#issuecomment-2032346491
> @Beihao-Zhou Thanks for your patch. Can you fix the CI first? https://github.com/apache/kvrocks/actions/runs/8500498583/job/23284955556?pr=2202#step:7:1062
Sry for the late reply.
Sure thing, it's fixed now! I've ran both `./x.py check tidy` and ./x.py format` locally, but fails to run `./x.py check golangci-lint` as it kept hanging there and being killed eventually. I'm wondering if any maintainer can trigger the workflow for me in this PR? Thanks!!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Beihao-Zhou (via GitHub)" <gi...@apache.org>.
Beihao-Zhou commented on PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#issuecomment-2077036398
> You can fix the CI first.
Sure thing, will fix it and modify code this weekend!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "jihuayu (via GitHub)" <gi...@apache.org>.
jihuayu commented on PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#issuecomment-2028964796
@Beihao-Zhou Thanks for your patch.
Can you fix the CI first? https://github.com/apache/kvrocks/actions/runs/8500498583/job/23284955556?pr=2202#step:7:1062
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#issuecomment-2033512768
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2202) **Quality Gate passed**
Issues
 [15 New issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=2202&resolved=false&inNewCodePeriod=true)
 [0 Accepted issues](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2202&metric=new_accepted_issues&view=list)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_kvrocks&pullRequest=2202&resolved=false&inNewCodePeriod=true)
 [65.4% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2202&metric=new_coverage&view=list)
 [1.1% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2202&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2202)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Beihao-Zhou (via GitHub)" <gi...@apache.org>.
Beihao-Zhou commented on code in PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#discussion_r1581170841
##########
src/commands/cmd_stream.cc:
##########
@@ -242,6 +242,100 @@ class CommandXDel : public Commander {
std::vector<redis::StreamEntryID> ids_;
};
+class CommandXClaim : public Commander {
+ public:
+ Status Parse(const std::vector<std::string> &args) override {
+ if (args.size() < 6) {
+ return {Status::RedisParseErr, errWrongNumOfArguments};
+ }
+
+ CommandParser parser(args, 1);
+ stream_name_ = GET_OR_RET(parser.TakeStr());
+ group_name_ = GET_OR_RET(parser.TakeStr());
+ consumer_name_ = GET_OR_RET(parser.TakeStr());
+ min_idle_time_ = GET_OR_RET(parser.TakeInt<uint64_t>());
Review Comment:
I see I see, just wondering if that's the case, for `idle`, `time` and `retrycount`, are they all going to be int64_t, followed by some parse result checking?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Beihao-Zhou (via GitHub)" <gi...@apache.org>.
Beihao-Zhou commented on code in PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#discussion_r1597668269
##########
src/types/redis_stream_base.h:
##########
@@ -161,6 +161,18 @@ struct StreamXGroupCreateOptions {
std::string last_id;
};
+struct StreamClaimOptions {
+ uint64_t idle_time = 0;
Review Comment:
`idle_time` is initialized as 0 if the client doesn't pass any value.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Beihao-Zhou (via GitHub)" <gi...@apache.org>.
Beihao-Zhou commented on code in PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#discussion_r1597668646
##########
src/types/redis_stream.cc:
##########
@@ -357,6 +357,134 @@ rocksdb::Status Stream::DeletePelEntries(const Slice &stream_name, const std::st
return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
}
+rocksdb::Status Stream::ClaimPelEntries(const Slice &stream_name, const std::string &group_name,
+ const std::string &consumer_name, const uint64_t min_idle_time,
+ const std::vector<StreamEntryID> &entry_ids, const StreamClaimOptions &options,
+ StreamClaimResult *result) {
+ std::string ns_key = AppendNamespacePrefix(stream_name);
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ StreamMetadata metadata(false);
+ rocksdb::Status s = GetMetadata(GetOptions{}, ns_key, &metadata);
+ if (!s.ok()) return s;
+
+ std::string group_key = internalKeyFromGroupName(ns_key, metadata, group_name);
+ std::string get_group_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, group_key, &get_group_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) {
+ return rocksdb::Status::InvalidArgument("NOGROUP No such consumer group " + group_name + " for key name " +
+ stream_name.ToString());
+ }
+ StreamConsumerGroupMetadata group_metadata = decodeStreamConsumerGroupMetadataValue(get_group_value);
+
+ std::string consumer_key = internalKeyFromConsumerName(ns_key, metadata, group_name, consumer_name);
+ std::string get_consumer_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, consumer_key, &get_consumer_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) {
+ int created_number = 0;
+ s = createConsumerWithoutLock(stream_name, group_name, consumer_name, &created_number);
+ if (!s.ok()) {
+ return s;
+ }
+ group_metadata.consumer_number += created_number;
+ }
+ StreamConsumerMetadata consumer_metadata;
+ if (!s.IsNotFound()) {
+ consumer_metadata = decodeStreamConsumerMetadataValue(get_consumer_value);
+ }
+ auto now = util::GetTimeStampMS();
+ consumer_metadata.last_idle = now;
+ consumer_metadata.last_active = now;
+
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisStream);
+ batch->PutLogData(log_data.Encode());
+
+ for (const auto &id : entry_ids) {
+ std::string raw_value;
+ rocksdb::Status s = getEntryRawValue(ns_key, metadata, id, &raw_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) continue;
+
+ std::string entry_key = internalPelKeyFromGroupAndEntryId(ns_key, metadata, group_name, id);
+ std::string value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, entry_key, &value);
+ StreamPelEntry pel_entry;
+
+ if (!s.ok() && s.IsNotFound() && options.force) {
+ pel_entry = {0, 0, ""};
+ group_metadata.pending_number += 1;
+ }
+
+ if (s.ok()) {
+ pel_entry = decodeStreamPelEntryValue(value);
+ }
+
+ if (s.ok() || (s.IsNotFound() && options.force)) {
+ if (now - pel_entry.last_delivery_time < min_idle_time) continue;
+
+ std::vector<std::string> values;
+ if (options.just_id) {
+ result->ids.emplace_back(id.ToString());
+ } else {
+ auto rv = DecodeRawStreamEntryValue(raw_value, &values);
+ if (!rv.IsOK()) {
+ return rocksdb::Status::InvalidArgument(rv.Msg());
+ }
+ result->entries.emplace_back(id.ToString(), std::move(values));
+ }
+
+ if (pel_entry.consumer_name != "") {
+ std::string original_consumer_key =
+ internalKeyFromConsumerName(ns_key, metadata, group_name, pel_entry.consumer_name);
+ std::string get_original_consumer_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, original_consumer_key,
+ &get_original_consumer_value);
+ if (!s.ok()) {
+ return s;
+ }
+ StreamConsumerMetadata original_consumer_metadata =
+ decodeStreamConsumerMetadataValue(get_original_consumer_value);
+ original_consumer_metadata.pending_number -= 1;
+ batch->Put(stream_cf_handle_, original_consumer_key,
+ encodeStreamConsumerMetadataValue(original_consumer_metadata));
+ }
+
+ pel_entry.consumer_name = consumer_name;
+ consumer_metadata.pending_number += 1;
+ if (options.with_time) {
+ pel_entry.last_delivery_time = options.last_delivery_time;
+ } else {
+ pel_entry.last_delivery_time = now - options.idle_time;
Review Comment:
Thanks for pointing it out! The `last_delivery_time` boundary check has been added now. But for `idle_time`, as commented below, it's been initialized as 0 if the client doesn't pass any value in struct `StreamClaimOptions`, so I'm wondering if we still need to check it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Beihao-Zhou (via GitHub)" <gi...@apache.org>.
Beihao-Zhou commented on PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#issuecomment-2021009215
Hi @Yangsx-1! I have a question about `MakeCmdAttr`. What are `first_key`, `last_key`, and `key_step` for here? Also for the `description` field, should I have `no-dbsize-check` for `XCLAIM`? Thanks a lot!!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "jihuayu (via GitHub)" <gi...@apache.org>.
jihuayu commented on PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#issuecomment-2033461782
@Beihao-Zhou You can enable action in your [fork repo](https://github.com/Beihao-Zhou/kvrocks/actions),it is free!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Yangsx-1 (via GitHub)" <gi...@apache.org>.
Yangsx-1 commented on code in PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#discussion_r1575884298
##########
src/types/redis_stream.cc:
##########
@@ -357,6 +357,134 @@ rocksdb::Status Stream::DeletePelEntries(const Slice &stream_name, const std::st
return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
}
+rocksdb::Status Stream::ClaimPelEntries(const Slice &stream_name, const std::string &group_name,
+ const std::string &consumer_name, const uint64_t min_idle_time,
+ const std::vector<StreamEntryID> &entry_ids, const StreamClaimOptions &options,
+ StreamClaimResult *result) {
+ std::string ns_key = AppendNamespacePrefix(stream_name);
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ StreamMetadata metadata(false);
+ rocksdb::Status s = GetMetadata(GetOptions{}, ns_key, &metadata);
+ if (!s.ok()) return s;
+
+ std::string group_key = internalKeyFromGroupName(ns_key, metadata, group_name);
+ std::string get_group_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, group_key, &get_group_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) {
+ return rocksdb::Status::InvalidArgument("NOGROUP No such consumer group " + group_name + " for key name " +
+ stream_name.ToString());
+ }
+ StreamConsumerGroupMetadata group_metadata = decodeStreamConsumerGroupMetadataValue(get_group_value);
+
+ std::string consumer_key = internalKeyFromConsumerName(ns_key, metadata, group_name, consumer_name);
+ std::string get_consumer_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, consumer_key, &get_consumer_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) {
+ int created_number = 0;
+ s = createConsumerWithoutLock(stream_name, group_name, consumer_name, &created_number);
+ if (!s.ok()) {
+ return s;
+ }
+ group_metadata.consumer_number += created_number;
+ }
+ StreamConsumerMetadata consumer_metadata;
+ if (!s.IsNotFound()) {
+ consumer_metadata = decodeStreamConsumerMetadataValue(get_consumer_value);
+ }
+ auto now = util::GetTimeStampMS();
+ consumer_metadata.last_idle = now;
+ consumer_metadata.last_active = now;
+
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisStream);
+ batch->PutLogData(log_data.Encode());
+
+ for (const auto &id : entry_ids) {
+ std::string raw_value;
+ rocksdb::Status s = getEntryRawValue(ns_key, metadata, id, &raw_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) continue;
+
+ std::string entry_key = internalPelKeyFromGroupAndEntryId(ns_key, metadata, group_name, id);
+ std::string value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, entry_key, &value);
+ StreamPelEntry pel_entry;
+
+ if (!s.ok() && s.IsNotFound() && options.force) {
+ pel_entry = {0, 0, ""};
+ group_metadata.pending_number += 1;
+ }
+
+ if (s.ok()) {
+ pel_entry = decodeStreamPelEntryValue(value);
+ }
+
+ if (s.ok() || (!s.ok() && s.IsNotFound() && options.force)) {
Review Comment:
Maybe !s.ok() is not necessary here.
##########
src/types/redis_stream.cc:
##########
@@ -357,6 +357,134 @@ rocksdb::Status Stream::DeletePelEntries(const Slice &stream_name, const std::st
return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
}
+rocksdb::Status Stream::ClaimPelEntries(const Slice &stream_name, const std::string &group_name,
+ const std::string &consumer_name, const uint64_t min_idle_time,
+ const std::vector<StreamEntryID> &entry_ids, const StreamClaimOptions &options,
+ StreamClaimResult *result) {
+ std::string ns_key = AppendNamespacePrefix(stream_name);
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ StreamMetadata metadata(false);
+ rocksdb::Status s = GetMetadata(GetOptions{}, ns_key, &metadata);
+ if (!s.ok()) return s;
+
+ std::string group_key = internalKeyFromGroupName(ns_key, metadata, group_name);
+ std::string get_group_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, group_key, &get_group_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) {
+ return rocksdb::Status::InvalidArgument("NOGROUP No such consumer group " + group_name + " for key name " +
+ stream_name.ToString());
+ }
+ StreamConsumerGroupMetadata group_metadata = decodeStreamConsumerGroupMetadataValue(get_group_value);
+
+ std::string consumer_key = internalKeyFromConsumerName(ns_key, metadata, group_name, consumer_name);
+ std::string get_consumer_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, consumer_key, &get_consumer_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) {
+ int created_number = 0;
+ s = createConsumerWithoutLock(stream_name, group_name, consumer_name, &created_number);
+ if (!s.ok()) {
+ return s;
+ }
+ group_metadata.consumer_number += created_number;
+ }
+ StreamConsumerMetadata consumer_metadata;
+ if (!s.IsNotFound()) {
+ consumer_metadata = decodeStreamConsumerMetadataValue(get_consumer_value);
+ }
+ auto now = util::GetTimeStampMS();
+ consumer_metadata.last_idle = now;
+ consumer_metadata.last_active = now;
+
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisStream);
+ batch->PutLogData(log_data.Encode());
+
+ for (const auto &id : entry_ids) {
+ std::string raw_value;
+ rocksdb::Status s = getEntryRawValue(ns_key, metadata, id, &raw_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) continue;
+
+ std::string entry_key = internalPelKeyFromGroupAndEntryId(ns_key, metadata, group_name, id);
+ std::string value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, entry_key, &value);
+ StreamPelEntry pel_entry;
+
+ if (!s.ok() && s.IsNotFound() && options.force) {
+ pel_entry = {0, 0, ""};
+ group_metadata.pending_number += 1;
+ }
+
+ if (s.ok()) {
+ pel_entry = decodeStreamPelEntryValue(value);
+ }
+
+ if (s.ok() || (!s.ok() && s.IsNotFound() && options.force)) {
+ if (now - pel_entry.last_delivery_time < min_idle_time) continue;
+
+ std::vector<std::string> values;
+ if (options.just_id) {
+ result->ids.emplace_back(id.ToString());
+ } else {
+ auto rv = DecodeRawStreamEntryValue(raw_value, &values);
+ if (!rv.IsOK()) {
+ return rocksdb::Status::InvalidArgument(rv.Msg());
+ }
+ result->entries.emplace_back(id.ToString(), std::move(values));
+ }
+
+ if (pel_entry.consumer_name != "") {
+ std::string original_consumer_key =
+ internalKeyFromConsumerName(ns_key, metadata, group_name, pel_entry.consumer_name);
+ std::string get_original_consumer_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, original_consumer_key,
+ &get_original_consumer_value);
+ if (!s.ok()) {
+ return s;
+ }
+ StreamConsumerMetadata original_consumer_metadata =
+ decodeStreamConsumerMetadataValue(get_original_consumer_value);
+ original_consumer_metadata.pending_number -= 1;
+ batch->Put(stream_cf_handle_, original_consumer_key,
+ encodeStreamConsumerMetadataValue(original_consumer_metadata));
+ }
+
+ pel_entry.consumer_name = consumer_name;
+ consumer_metadata.pending_number += 1;
+ if (options.with_time) {
+ pel_entry.last_delivery_time = options.last_delivery_time;
+ } else {
+ pel_entry.last_delivery_time = now - options.idle_time;
+ }
+
+ if (options.with_retry_count) {
+ pel_entry.last_delivery_count = options.last_delivery_count;
+ } else if (!options.just_id) {
+ pel_entry.last_delivery_count += 1;
+ }
+
+ std::string pel_value = encodeStreamPelEntryValue(pel_entry);
+ batch->Put(stream_cf_handle_, entry_key, pel_value);
+ }
+ }
+
+ if (options.with_last_id) {
+ group_metadata.last_delivered_id = options.last_delivered_id;
Review Comment:
Only when options.last_deliverd_id > group_metadata.last_delivered_id can we update it.
##########
src/commands/cmd_stream.cc:
##########
@@ -242,6 +242,100 @@ class CommandXDel : public Commander {
std::vector<redis::StreamEntryID> ids_;
};
+class CommandXClaim : public Commander {
+ public:
+ Status Parse(const std::vector<std::string> &args) override {
+ if (args.size() < 6) {
+ return {Status::RedisParseErr, errWrongNumOfArguments};
+ }
+
+ CommandParser parser(args, 1);
+ stream_name_ = GET_OR_RET(parser.TakeStr());
+ group_name_ = GET_OR_RET(parser.TakeStr());
+ consumer_name_ = GET_OR_RET(parser.TakeStr());
+ min_idle_time_ = GET_OR_RET(parser.TakeInt<uint64_t>());
Review Comment:
Here should be int64_t, and if the min_idle_time < 0, we need to reset it to 0.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Yangsx-1 (via GitHub)" <gi...@apache.org>.
Yangsx-1 commented on code in PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#discussion_r1593311060
##########
src/types/redis_stream.cc:
##########
@@ -357,6 +357,134 @@ rocksdb::Status Stream::DeletePelEntries(const Slice &stream_name, const std::st
return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
}
+rocksdb::Status Stream::ClaimPelEntries(const Slice &stream_name, const std::string &group_name,
+ const std::string &consumer_name, const uint64_t min_idle_time,
+ const std::vector<StreamEntryID> &entry_ids, const StreamClaimOptions &options,
+ StreamClaimResult *result) {
+ std::string ns_key = AppendNamespacePrefix(stream_name);
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ StreamMetadata metadata(false);
+ rocksdb::Status s = GetMetadata(GetOptions{}, ns_key, &metadata);
+ if (!s.ok()) return s;
+
+ std::string group_key = internalKeyFromGroupName(ns_key, metadata, group_name);
+ std::string get_group_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, group_key, &get_group_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) {
+ return rocksdb::Status::InvalidArgument("NOGROUP No such consumer group " + group_name + " for key name " +
+ stream_name.ToString());
+ }
+ StreamConsumerGroupMetadata group_metadata = decodeStreamConsumerGroupMetadataValue(get_group_value);
+
+ std::string consumer_key = internalKeyFromConsumerName(ns_key, metadata, group_name, consumer_name);
+ std::string get_consumer_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, consumer_key, &get_consumer_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) {
+ int created_number = 0;
+ s = createConsumerWithoutLock(stream_name, group_name, consumer_name, &created_number);
+ if (!s.ok()) {
+ return s;
+ }
+ group_metadata.consumer_number += created_number;
+ }
+ StreamConsumerMetadata consumer_metadata;
+ if (!s.IsNotFound()) {
+ consumer_metadata = decodeStreamConsumerMetadataValue(get_consumer_value);
+ }
+ auto now = util::GetTimeStampMS();
+ consumer_metadata.last_idle = now;
+ consumer_metadata.last_active = now;
+
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisStream);
+ batch->PutLogData(log_data.Encode());
+
+ for (const auto &id : entry_ids) {
+ std::string raw_value;
+ rocksdb::Status s = getEntryRawValue(ns_key, metadata, id, &raw_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) continue;
+
+ std::string entry_key = internalPelKeyFromGroupAndEntryId(ns_key, metadata, group_name, id);
+ std::string value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, entry_key, &value);
+ StreamPelEntry pel_entry;
+
+ if (!s.ok() && s.IsNotFound() && options.force) {
+ pel_entry = {0, 0, ""};
+ group_metadata.pending_number += 1;
+ }
+
+ if (s.ok()) {
+ pel_entry = decodeStreamPelEntryValue(value);
+ }
+
+ if (s.ok() || (s.IsNotFound() && options.force)) {
+ if (now - pel_entry.last_delivery_time < min_idle_time) continue;
+
+ std::vector<std::string> values;
+ if (options.just_id) {
+ result->ids.emplace_back(id.ToString());
+ } else {
+ auto rv = DecodeRawStreamEntryValue(raw_value, &values);
+ if (!rv.IsOK()) {
+ return rocksdb::Status::InvalidArgument(rv.Msg());
+ }
+ result->entries.emplace_back(id.ToString(), std::move(values));
+ }
+
+ if (pel_entry.consumer_name != "") {
+ std::string original_consumer_key =
+ internalKeyFromConsumerName(ns_key, metadata, group_name, pel_entry.consumer_name);
+ std::string get_original_consumer_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, original_consumer_key,
+ &get_original_consumer_value);
+ if (!s.ok()) {
+ return s;
+ }
+ StreamConsumerMetadata original_consumer_metadata =
+ decodeStreamConsumerMetadataValue(get_original_consumer_value);
+ original_consumer_metadata.pending_number -= 1;
+ batch->Put(stream_cf_handle_, original_consumer_key,
+ encodeStreamConsumerMetadataValue(original_consumer_metadata));
+ }
+
+ pel_entry.consumer_name = consumer_name;
+ consumer_metadata.pending_number += 1;
+ if (options.with_time) {
+ pel_entry.last_delivery_time = options.last_delivery_time;
+ } else {
+ pel_entry.last_delivery_time = now - options.idle_time;
Review Comment:
`If no IDLE/TIME option was passed, we want the last delivery time to be now, so that the idle time of the message will be zero.`
I think we should check this little proplem. And if the options.last_delivery_time is larger than now, we should modify it equal to now:
`if (deliverytime < 0 || deliverytime > now) deliverytime = now`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "jihuayu (via GitHub)" <gi...@apache.org>.
jihuayu commented on PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#issuecomment-2021694706
Hi @Beihao-Zhou.
First_key, last_key, and key_step are Redis key range. For this command, it will be 1 1 1.
In my opinion, this command could not add the `no-dbsize-check` flag, `no-dbsize-check` is only added to the command which can make the db smaller.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Beihao-Zhou (via GitHub)" <gi...@apache.org>.
Beihao-Zhou commented on code in PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#discussion_r1581170841
##########
src/commands/cmd_stream.cc:
##########
@@ -242,6 +242,100 @@ class CommandXDel : public Commander {
std::vector<redis::StreamEntryID> ids_;
};
+class CommandXClaim : public Commander {
+ public:
+ Status Parse(const std::vector<std::string> &args) override {
+ if (args.size() < 6) {
+ return {Status::RedisParseErr, errWrongNumOfArguments};
+ }
+
+ CommandParser parser(args, 1);
+ stream_name_ = GET_OR_RET(parser.TakeStr());
+ group_name_ = GET_OR_RET(parser.TakeStr());
+ consumer_name_ = GET_OR_RET(parser.TakeStr());
+ min_idle_time_ = GET_OR_RET(parser.TakeInt<uint64_t>());
Review Comment:
I see I see, just wondering if that's the case, for `idle`, `time` and `retrycount`, are they all going to be int64_t, followed by parse result checking such that only non-negative values are allowed?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Yangsx-1 (via GitHub)" <gi...@apache.org>.
Yangsx-1 commented on code in PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#discussion_r1581994485
##########
src/commands/cmd_stream.cc:
##########
@@ -242,6 +242,100 @@ class CommandXDel : public Commander {
std::vector<redis::StreamEntryID> ids_;
};
+class CommandXClaim : public Commander {
+ public:
+ Status Parse(const std::vector<std::string> &args) override {
+ if (args.size() < 6) {
+ return {Status::RedisParseErr, errWrongNumOfArguments};
+ }
+
+ CommandParser parser(args, 1);
+ stream_name_ = GET_OR_RET(parser.TakeStr());
+ group_name_ = GET_OR_RET(parser.TakeStr());
+ consumer_name_ = GET_OR_RET(parser.TakeStr());
+ min_idle_time_ = GET_OR_RET(parser.TakeInt<uint64_t>());
Review Comment:
Yes, I believe it's the same way Redis uses.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Yangsx-1 (via GitHub)" <gi...@apache.org>.
Yangsx-1 commented on code in PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#discussion_r1597876159
##########
src/types/redis_stream.cc:
##########
@@ -357,6 +357,134 @@ rocksdb::Status Stream::DeletePelEntries(const Slice &stream_name, const std::st
return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
}
+rocksdb::Status Stream::ClaimPelEntries(const Slice &stream_name, const std::string &group_name,
+ const std::string &consumer_name, const uint64_t min_idle_time,
+ const std::vector<StreamEntryID> &entry_ids, const StreamClaimOptions &options,
+ StreamClaimResult *result) {
+ std::string ns_key = AppendNamespacePrefix(stream_name);
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ StreamMetadata metadata(false);
+ rocksdb::Status s = GetMetadata(GetOptions{}, ns_key, &metadata);
+ if (!s.ok()) return s;
+
+ std::string group_key = internalKeyFromGroupName(ns_key, metadata, group_name);
+ std::string get_group_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, group_key, &get_group_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) {
+ return rocksdb::Status::InvalidArgument("NOGROUP No such consumer group " + group_name + " for key name " +
+ stream_name.ToString());
+ }
+ StreamConsumerGroupMetadata group_metadata = decodeStreamConsumerGroupMetadataValue(get_group_value);
+
+ std::string consumer_key = internalKeyFromConsumerName(ns_key, metadata, group_name, consumer_name);
+ std::string get_consumer_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, consumer_key, &get_consumer_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) {
+ int created_number = 0;
+ s = createConsumerWithoutLock(stream_name, group_name, consumer_name, &created_number);
+ if (!s.ok()) {
+ return s;
+ }
+ group_metadata.consumer_number += created_number;
+ }
+ StreamConsumerMetadata consumer_metadata;
+ if (!s.IsNotFound()) {
+ consumer_metadata = decodeStreamConsumerMetadataValue(get_consumer_value);
+ }
+ auto now = util::GetTimeStampMS();
+ consumer_metadata.last_idle = now;
+ consumer_metadata.last_active = now;
+
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisStream);
+ batch->PutLogData(log_data.Encode());
+
+ for (const auto &id : entry_ids) {
+ std::string raw_value;
+ rocksdb::Status s = getEntryRawValue(ns_key, metadata, id, &raw_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) continue;
+
+ std::string entry_key = internalPelKeyFromGroupAndEntryId(ns_key, metadata, group_name, id);
+ std::string value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, entry_key, &value);
+ StreamPelEntry pel_entry;
+
+ if (!s.ok() && s.IsNotFound() && options.force) {
+ pel_entry = {0, 0, ""};
+ group_metadata.pending_number += 1;
+ }
+
+ if (s.ok()) {
+ pel_entry = decodeStreamPelEntryValue(value);
+ }
+
+ if (s.ok() || (s.IsNotFound() && options.force)) {
+ if (now - pel_entry.last_delivery_time < min_idle_time) continue;
+
+ std::vector<std::string> values;
+ if (options.just_id) {
+ result->ids.emplace_back(id.ToString());
+ } else {
+ auto rv = DecodeRawStreamEntryValue(raw_value, &values);
+ if (!rv.IsOK()) {
+ return rocksdb::Status::InvalidArgument(rv.Msg());
+ }
+ result->entries.emplace_back(id.ToString(), std::move(values));
+ }
+
+ if (pel_entry.consumer_name != "") {
+ std::string original_consumer_key =
+ internalKeyFromConsumerName(ns_key, metadata, group_name, pel_entry.consumer_name);
+ std::string get_original_consumer_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, original_consumer_key,
+ &get_original_consumer_value);
+ if (!s.ok()) {
+ return s;
+ }
+ StreamConsumerMetadata original_consumer_metadata =
+ decodeStreamConsumerMetadataValue(get_original_consumer_value);
+ original_consumer_metadata.pending_number -= 1;
+ batch->Put(stream_cf_handle_, original_consumer_key,
+ encodeStreamConsumerMetadataValue(original_consumer_metadata));
+ }
+
+ pel_entry.consumer_name = consumer_name;
+ consumer_metadata.pending_number += 1;
+ if (options.with_time) {
+ pel_entry.last_delivery_time = options.last_delivery_time;
+ } else {
+ pel_entry.last_delivery_time = now - options.idle_time;
Review Comment:
I think we don't need to check it anymore.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#issuecomment-2032360387
@Beihao-Zhou Done, thanks for your efforts.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Beihao-Zhou (via GitHub)" <gi...@apache.org>.
Beihao-Zhou commented on code in PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#discussion_r1548099517
##########
src/types/redis_stream.cc:
##########
@@ -357,6 +357,134 @@ rocksdb::Status Stream::DeletePelEntries(const Slice &stream_name, const std::st
return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
}
+rocksdb::Status Stream::ClaimPelEntries(const Slice &stream_name, const std::string &group_name,
+ const std::string &consumer_name, const uint64_t min_idle_time,
Review Comment:
Thanks!! Fixed :))
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Beihao-Zhou (via GitHub)" <gi...@apache.org>.
Beihao-Zhou commented on code in PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#discussion_r1597668646
##########
src/types/redis_stream.cc:
##########
@@ -357,6 +357,134 @@ rocksdb::Status Stream::DeletePelEntries(const Slice &stream_name, const std::st
return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
}
+rocksdb::Status Stream::ClaimPelEntries(const Slice &stream_name, const std::string &group_name,
+ const std::string &consumer_name, const uint64_t min_idle_time,
+ const std::vector<StreamEntryID> &entry_ids, const StreamClaimOptions &options,
+ StreamClaimResult *result) {
+ std::string ns_key = AppendNamespacePrefix(stream_name);
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ StreamMetadata metadata(false);
+ rocksdb::Status s = GetMetadata(GetOptions{}, ns_key, &metadata);
+ if (!s.ok()) return s;
+
+ std::string group_key = internalKeyFromGroupName(ns_key, metadata, group_name);
+ std::string get_group_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, group_key, &get_group_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) {
+ return rocksdb::Status::InvalidArgument("NOGROUP No such consumer group " + group_name + " for key name " +
+ stream_name.ToString());
+ }
+ StreamConsumerGroupMetadata group_metadata = decodeStreamConsumerGroupMetadataValue(get_group_value);
+
+ std::string consumer_key = internalKeyFromConsumerName(ns_key, metadata, group_name, consumer_name);
+ std::string get_consumer_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, consumer_key, &get_consumer_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) {
+ int created_number = 0;
+ s = createConsumerWithoutLock(stream_name, group_name, consumer_name, &created_number);
+ if (!s.ok()) {
+ return s;
+ }
+ group_metadata.consumer_number += created_number;
+ }
+ StreamConsumerMetadata consumer_metadata;
+ if (!s.IsNotFound()) {
+ consumer_metadata = decodeStreamConsumerMetadataValue(get_consumer_value);
+ }
+ auto now = util::GetTimeStampMS();
+ consumer_metadata.last_idle = now;
+ consumer_metadata.last_active = now;
+
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisStream);
+ batch->PutLogData(log_data.Encode());
+
+ for (const auto &id : entry_ids) {
+ std::string raw_value;
+ rocksdb::Status s = getEntryRawValue(ns_key, metadata, id, &raw_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) continue;
+
+ std::string entry_key = internalPelKeyFromGroupAndEntryId(ns_key, metadata, group_name, id);
+ std::string value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, entry_key, &value);
+ StreamPelEntry pel_entry;
+
+ if (!s.ok() && s.IsNotFound() && options.force) {
+ pel_entry = {0, 0, ""};
+ group_metadata.pending_number += 1;
+ }
+
+ if (s.ok()) {
+ pel_entry = decodeStreamPelEntryValue(value);
+ }
+
+ if (s.ok() || (s.IsNotFound() && options.force)) {
+ if (now - pel_entry.last_delivery_time < min_idle_time) continue;
+
+ std::vector<std::string> values;
+ if (options.just_id) {
+ result->ids.emplace_back(id.ToString());
+ } else {
+ auto rv = DecodeRawStreamEntryValue(raw_value, &values);
+ if (!rv.IsOK()) {
+ return rocksdb::Status::InvalidArgument(rv.Msg());
+ }
+ result->entries.emplace_back(id.ToString(), std::move(values));
+ }
+
+ if (pel_entry.consumer_name != "") {
+ std::string original_consumer_key =
+ internalKeyFromConsumerName(ns_key, metadata, group_name, pel_entry.consumer_name);
+ std::string get_original_consumer_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, original_consumer_key,
+ &get_original_consumer_value);
+ if (!s.ok()) {
+ return s;
+ }
+ StreamConsumerMetadata original_consumer_metadata =
+ decodeStreamConsumerMetadataValue(get_original_consumer_value);
+ original_consumer_metadata.pending_number -= 1;
+ batch->Put(stream_cf_handle_, original_consumer_key,
+ encodeStreamConsumerMetadataValue(original_consumer_metadata));
+ }
+
+ pel_entry.consumer_name = consumer_name;
+ consumer_metadata.pending_number += 1;
+ if (options.with_time) {
+ pel_entry.last_delivery_time = options.last_delivery_time;
+ } else {
+ pel_entry.last_delivery_time = now - options.idle_time;
Review Comment:
Thanks for pointing it out! The `last_delivery_time` boundary check has been added now. But for `idle_time`, as commented below, it's been initialized as 0 if the client doesn't pass any value in struct `StreamClaimOptions`, so I'm wondering if we still need to check it.
https://github.com/apache/kvrocks/pull/2202#discussion_r1597668269
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Beihao-Zhou (via GitHub)" <gi...@apache.org>.
Beihao-Zhou commented on code in PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#discussion_r1597668646
##########
src/types/redis_stream.cc:
##########
@@ -357,6 +357,134 @@ rocksdb::Status Stream::DeletePelEntries(const Slice &stream_name, const std::st
return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
}
+rocksdb::Status Stream::ClaimPelEntries(const Slice &stream_name, const std::string &group_name,
+ const std::string &consumer_name, const uint64_t min_idle_time,
+ const std::vector<StreamEntryID> &entry_ids, const StreamClaimOptions &options,
+ StreamClaimResult *result) {
+ std::string ns_key = AppendNamespacePrefix(stream_name);
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ StreamMetadata metadata(false);
+ rocksdb::Status s = GetMetadata(GetOptions{}, ns_key, &metadata);
+ if (!s.ok()) return s;
+
+ std::string group_key = internalKeyFromGroupName(ns_key, metadata, group_name);
+ std::string get_group_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, group_key, &get_group_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) {
+ return rocksdb::Status::InvalidArgument("NOGROUP No such consumer group " + group_name + " for key name " +
+ stream_name.ToString());
+ }
+ StreamConsumerGroupMetadata group_metadata = decodeStreamConsumerGroupMetadataValue(get_group_value);
+
+ std::string consumer_key = internalKeyFromConsumerName(ns_key, metadata, group_name, consumer_name);
+ std::string get_consumer_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, consumer_key, &get_consumer_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) {
+ int created_number = 0;
+ s = createConsumerWithoutLock(stream_name, group_name, consumer_name, &created_number);
+ if (!s.ok()) {
+ return s;
+ }
+ group_metadata.consumer_number += created_number;
+ }
+ StreamConsumerMetadata consumer_metadata;
+ if (!s.IsNotFound()) {
+ consumer_metadata = decodeStreamConsumerMetadataValue(get_consumer_value);
+ }
+ auto now = util::GetTimeStampMS();
+ consumer_metadata.last_idle = now;
+ consumer_metadata.last_active = now;
+
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisStream);
+ batch->PutLogData(log_data.Encode());
+
+ for (const auto &id : entry_ids) {
+ std::string raw_value;
+ rocksdb::Status s = getEntryRawValue(ns_key, metadata, id, &raw_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) continue;
+
+ std::string entry_key = internalPelKeyFromGroupAndEntryId(ns_key, metadata, group_name, id);
+ std::string value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, entry_key, &value);
+ StreamPelEntry pel_entry;
+
+ if (!s.ok() && s.IsNotFound() && options.force) {
+ pel_entry = {0, 0, ""};
+ group_metadata.pending_number += 1;
+ }
+
+ if (s.ok()) {
+ pel_entry = decodeStreamPelEntryValue(value);
+ }
+
+ if (s.ok() || (s.IsNotFound() && options.force)) {
+ if (now - pel_entry.last_delivery_time < min_idle_time) continue;
+
+ std::vector<std::string> values;
+ if (options.just_id) {
+ result->ids.emplace_back(id.ToString());
+ } else {
+ auto rv = DecodeRawStreamEntryValue(raw_value, &values);
+ if (!rv.IsOK()) {
+ return rocksdb::Status::InvalidArgument(rv.Msg());
+ }
+ result->entries.emplace_back(id.ToString(), std::move(values));
+ }
+
+ if (pel_entry.consumer_name != "") {
+ std::string original_consumer_key =
+ internalKeyFromConsumerName(ns_key, metadata, group_name, pel_entry.consumer_name);
+ std::string get_original_consumer_value;
+ s = storage_->Get(rocksdb::ReadOptions(), stream_cf_handle_, original_consumer_key,
+ &get_original_consumer_value);
+ if (!s.ok()) {
+ return s;
+ }
+ StreamConsumerMetadata original_consumer_metadata =
+ decodeStreamConsumerMetadataValue(get_original_consumer_value);
+ original_consumer_metadata.pending_number -= 1;
+ batch->Put(stream_cf_handle_, original_consumer_key,
+ encodeStreamConsumerMetadataValue(original_consumer_metadata));
+ }
+
+ pel_entry.consumer_name = consumer_name;
+ consumer_metadata.pending_number += 1;
+ if (options.with_time) {
+ pel_entry.last_delivery_time = options.last_delivery_time;
+ } else {
+ pel_entry.last_delivery_time = now - options.idle_time;
Review Comment:
Thanks for pointing it out! The `last_delivery_time` boundary check has been added now. But for `idle_time`, as commented below (https://github.com/apache/kvrocks/pull/2202#discussion_r1597668269), it's been initialized as 0 if the client doesn't pass any value in struct `StreamClaimOptions`, so I'm wondering if we still need to check it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Beihao-Zhou (via GitHub)" <gi...@apache.org>.
Beihao-Zhou commented on PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#issuecomment-2083757794
@Yangsx-1 Hiii! Code and CI are fixed (CI succeeds on my fork). Could you please take a look at the PR again? Thanks!!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Beihao-Zhou (via GitHub)" <gi...@apache.org>.
Beihao-Zhou commented on PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#issuecomment-2022663063
> Hi @Beihao-Zhou. First_key, last_key, and key_step are Redis key range. For this command, it will be 1 1 1. In my opinion, this command could not add the `no-dbsize-check` flag, `no-dbsize-check` is only added to the command which can make the db smaller.
I see I seem thank you very much!!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Beihao-Zhou (via GitHub)" <gi...@apache.org>.
Beihao-Zhou commented on PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#issuecomment-2034505020
> @Beihao-Zhou You can enable action in your [fork repo](https://github.com/Beihao-Zhou/kvrocks/actions),it is free!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Beihao-Zhou (via GitHub)" <gi...@apache.org>.
Beihao-Zhou commented on PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#issuecomment-2106765802
@Yangsx-1 Thanks for reviewing! Everything is fixed then :))
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "Yangsx-1 (via GitHub)" <gi...@apache.org>.
Yangsx-1 commented on PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#issuecomment-2106650482
> Also wondering does anyone know why the linting check couldn't pass as shown with errors (https://github.com/Beihao-Zhou/kvrocks/actions/runs/9052637137/job/24870567712#step:7:1493).
@Beihao-Zhou The PR #2276 modified the variable name and the clang-tidy told you the variable name is wrong.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat: add support of XCLAIM [kvrocks]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #2202:
URL: https://github.com/apache/kvrocks/pull/2202#issuecomment-2084500235
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2202) **Quality Gate passed**
Issues
 [18 New issues](https://sonarcloud.io/project/issues?id=apache_kvrocks&pullRequest=2202&resolved=false&inNewCodePeriod=true)
 [0 Accepted issues](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2202&metric=new_accepted_issues&view=list)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_kvrocks&pullRequest=2202&resolved=false&inNewCodePeriod=true)
 [53.9% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2202&metric=new_coverage&view=list)
 [2.7% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=2202&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=2202)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org