You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by "glcrazier (via GitHub)" <gi...@apache.org> on 2024/02/21 07:27:04 UTC
[PR] [rust] support handling telemetry command in Producer [rocketmq-clients]
glcrazier opened a new pull request, #682:
URL: https://github.com/apache/rocketmq-clients/pull/682
<!-- Please make sure the target branch is right. In most case, the target branch should be `master`. -->
### Which Issue(s) This PR Fixes
<!-- Please ensure that the related issue has already been created, and [link this pull request to that issue using keywords](<https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword>) to ensure automatic closure. -->
Fixes #issue_id
### Brief Description
<!-- Write a brief description for your pull request to help the maintainer understand the reasons behind your changes. -->
### How Did You Test This Change?
run cargo test and example to make sure it works
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] support handling settings command in Producer [rocketmq-clients]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #682:
URL: https://github.com/apache/rocketmq-clients/pull/682#issuecomment-1956115310
## [Codecov](https://app.codecov.io/gh/apache/rocketmq-clients/pull/682?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: `50 lines` in your changes are missing coverage. Please review.
> Comparison is base [(`65b70ed`)](https://app.codecov.io/gh/apache/rocketmq-clients/commit/65b70ed73259e8667e75a94e5162f75e91b4bf1b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) 55.89% compared to head [(`0871eb6`)](https://app.codecov.io/gh/apache/rocketmq-clients/pull/682?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) 47.73%.
> Report is 1 commits behind head on master.
| [Files](https://app.codecov.io/gh/apache/rocketmq-clients/pull/682?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [rust/src/producer.rs](https://app.codecov.io/gh/apache/rocketmq-clients/pull/682?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cnVzdC9zcmMvcHJvZHVjZXIucnM=) | 78.34% | [26 Missing and 8 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq-clients/pull/682?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [rust/src/simple\_consumer.rs](https://app.codecov.io/gh/apache/rocketmq-clients/pull/682?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cnVzdC9zcmMvc2ltcGxlX2NvbnN1bWVyLnJz) | 38.88% | [8 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq-clients/pull/682?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [rust/src/client.rs](https://app.codecov.io/gh/apache/rocketmq-clients/pull/682?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cnVzdC9zcmMvY2xpZW50LnJz) | 64.28% | [3 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq-clients/pull/682?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #682 +/- ##
============================================
- Coverage 55.89% 47.73% -8.17%
Complexity 668 668
============================================
Files 259 207 -52
Lines 15653 12336 -3317
Branches 3891 3693 -198
============================================
- Hits 8750 5889 -2861
+ Misses 6650 6180 -470
- Partials 253 267 +14
```
| [Flag](https://app.codecov.io/gh/apache/rocketmq-clients/pull/682/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [nodejs](https://app.codecov.io/gh/apache/rocketmq-clients/pull/682/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [rust](https://app.codecov.io/gh/apache/rocketmq-clients/pull/682/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `80.01% <73.54%> (+0.69%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/rocketmq-clients/pull/682?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] support handling settings command in Producer [rocketmq-clients]
Posted by "lizhanhui (via GitHub)" <gi...@apache.org>.
lizhanhui commented on code in PR #682:
URL: https://github.com/apache/rocketmq-clients/pull/682#discussion_r1497711161
##########
rust/src/producer.rs:
##########
@@ -93,23 +116,80 @@ impl Producer {
};
let logger = log::logger(option.logging_format());
let settings = build_producer_settings(&option, &client_option);
- let mut client = Client::new(&logger, client_option, settings)?;
- client.set_transaction_checker(transaction_checker);
+ let client = Client::new(&logger, client_option, settings)?;
+ let option = Arc::new(RwLock::new(option));
Ok(Producer {
option,
logger,
client,
+ transaction_checker: Some(transaction_checker),
+ shutdown_tx: None,
})
}
+ async fn get_resource_namespace(&self) -> String {
+ let option_guard = self.option.read();
+ let resource_namespace = option_guard.await.namespace().to_string();
+ resource_namespace
+ }
+
/// Start the producer
pub async fn start(&mut self) -> Result<(), ClientError> {
- self.client.start().await?;
- if let Some(topics) = self.option.topics() {
+ let (telemetry_command_tx, mut telemetry_command_rx) = mpsc::channel(16);
+ let telemetry_command_tx: mpsc::Sender<pb::telemetry_command::Command> =
+ telemetry_command_tx;
+ self.client.start(telemetry_command_tx).await?;
+ let option_guard = self.option.read().await;
+ let topics = option_guard.topics();
+ if let Some(topics) = topics {
for topic in topics {
self.client.topic_route(topic, true).await?;
}
}
+ drop(option_guard);
+ let transaction_checker = self.transaction_checker.take();
+ if transaction_checker.is_some() {
+ self.transaction_checker = Some(Box::new(|_, _| TransactionResolution::UNKNOWN));
Review Comment:
You are right.
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] support handling settings command in Producer [rocketmq-clients]
Posted by "lizhanhui (via GitHub)" <gi...@apache.org>.
lizhanhui commented on code in PR #682:
URL: https://github.com/apache/rocketmq-clients/pull/682#discussion_r1497660393
##########
rust/src/producer.rs:
##########
@@ -93,23 +116,80 @@ impl Producer {
};
let logger = log::logger(option.logging_format());
let settings = build_producer_settings(&option, &client_option);
- let mut client = Client::new(&logger, client_option, settings)?;
- client.set_transaction_checker(transaction_checker);
+ let client = Client::new(&logger, client_option, settings)?;
+ let option = Arc::new(RwLock::new(option));
Ok(Producer {
option,
logger,
client,
+ transaction_checker: Some(transaction_checker),
+ shutdown_tx: None,
})
}
+ async fn get_resource_namespace(&self) -> String {
+ let option_guard = self.option.read();
+ let resource_namespace = option_guard.await.namespace().to_string();
+ resource_namespace
+ }
+
/// Start the producer
pub async fn start(&mut self) -> Result<(), ClientError> {
- self.client.start().await?;
- if let Some(topics) = self.option.topics() {
+ let (telemetry_command_tx, mut telemetry_command_rx) = mpsc::channel(16);
+ let telemetry_command_tx: mpsc::Sender<pb::telemetry_command::Command> =
+ telemetry_command_tx;
+ self.client.start(telemetry_command_tx).await?;
+ let option_guard = self.option.read().await;
+ let topics = option_guard.topics();
+ if let Some(topics) = topics {
for topic in topics {
self.client.topic_route(topic, true).await?;
}
}
+ drop(option_guard);
+ let transaction_checker = self.transaction_checker.take();
+ if transaction_checker.is_some() {
+ self.transaction_checker = Some(Box::new(|_, _| TransactionResolution::UNKNOWN));
Review Comment:
Why not simply take the transaction checker and leave None afterward?
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] support handling settings command in Producer [rocketmq-clients]
Posted by "glcrazier (via GitHub)" <gi...@apache.org>.
glcrazier commented on code in PR #682:
URL: https://github.com/apache/rocketmq-clients/pull/682#discussion_r1497689223
##########
rust/src/producer.rs:
##########
@@ -93,23 +116,80 @@ impl Producer {
};
let logger = log::logger(option.logging_format());
let settings = build_producer_settings(&option, &client_option);
- let mut client = Client::new(&logger, client_option, settings)?;
- client.set_transaction_checker(transaction_checker);
+ let client = Client::new(&logger, client_option, settings)?;
+ let option = Arc::new(RwLock::new(option));
Ok(Producer {
option,
logger,
client,
+ transaction_checker: Some(transaction_checker),
+ shutdown_tx: None,
})
}
+ async fn get_resource_namespace(&self) -> String {
+ let option_guard = self.option.read();
+ let resource_namespace = option_guard.await.namespace().to_string();
+ resource_namespace
+ }
+
/// Start the producer
pub async fn start(&mut self) -> Result<(), ClientError> {
- self.client.start().await?;
- if let Some(topics) = self.option.topics() {
+ let (telemetry_command_tx, mut telemetry_command_rx) = mpsc::channel(16);
+ let telemetry_command_tx: mpsc::Sender<pb::telemetry_command::Command> =
+ telemetry_command_tx;
+ self.client.start(telemetry_command_tx).await?;
+ let option_guard = self.option.read().await;
+ let topics = option_guard.topics();
+ if let Some(topics) = topics {
for topic in topics {
self.client.topic_route(topic, true).await?;
}
}
+ drop(option_guard);
+ let transaction_checker = self.transaction_checker.take();
+ if transaction_checker.is_some() {
+ self.transaction_checker = Some(Box::new(|_, _| TransactionResolution::UNKNOWN));
Review Comment:
> Why not simply take the transaction checker and leave None afterward?
In `send_transaction_message` function, the producer uses `has_transaction_checker` function to check if it's legal to call this function. If we leave None, we have no idea whether it's a transaction producer.
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] support handling settings command in Producer [rocketmq-clients]
Posted by "lizhanhui (via GitHub)" <gi...@apache.org>.
lizhanhui merged PR #682:
URL: https://github.com/apache/rocketmq-clients/pull/682
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] support handling settings command in Producer [rocketmq-clients]
Posted by "glcrazier (via GitHub)" <gi...@apache.org>.
glcrazier commented on code in PR #682:
URL: https://github.com/apache/rocketmq-clients/pull/682#discussion_r1497177364
##########
rust/src/producer.rs:
##########
@@ -118,7 +196,53 @@ impl Producer {
Ok(())
}
- fn transform_messages_to_protobuf(
+ async fn handle_recover_orphaned_transaction_command<T: RPCClient + 'static>(
+ mut rpc_client: T,
+ command: pb::RecoverOrphanedTransactionCommand,
+ transaction_checker: &Option<Box<TransactionChecker>>,
+ endpoints: Endpoints,
+ ) -> Result<(), ClientError> {
+ let transaction_id = command.transaction_id;
+ let message = command.message.unwrap();
+ let message_id = message
+ .system_properties
+ .as_ref()
+ .unwrap()
Review Comment:
Got it, will fix
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] support handling settings command in Producer [rocketmq-clients]
Posted by "lizhanhui (via GitHub)" <gi...@apache.org>.
lizhanhui commented on code in PR #682:
URL: https://github.com/apache/rocketmq-clients/pull/682#discussion_r1497165946
##########
rust/src/producer.rs:
##########
@@ -118,7 +196,53 @@ impl Producer {
Ok(())
}
- fn transform_messages_to_protobuf(
+ async fn handle_recover_orphaned_transaction_command<T: RPCClient + 'static>(
+ mut rpc_client: T,
+ command: pb::RecoverOrphanedTransactionCommand,
+ transaction_checker: &Option<Box<TransactionChecker>>,
+ endpoints: Endpoints,
+ ) -> Result<(), ClientError> {
+ let transaction_id = command.transaction_id;
+ let message = command.message.unwrap();
+ let message_id = message
+ .system_properties
+ .as_ref()
+ .unwrap()
Review Comment:
Prefer propagating error to the upper layer over unwrap, which may panic and crash the whole process.
##########
rust/src/producer.rs:
##########
@@ -118,7 +196,53 @@ impl Producer {
Ok(())
}
- fn transform_messages_to_protobuf(
+ async fn handle_recover_orphaned_transaction_command<T: RPCClient + 'static>(
+ mut rpc_client: T,
+ command: pb::RecoverOrphanedTransactionCommand,
+ transaction_checker: &Option<Box<TransactionChecker>>,
+ endpoints: Endpoints,
+ ) -> Result<(), ClientError> {
+ let transaction_id = command.transaction_id;
+ let message = command.message.unwrap();
+ let message_id = message
+ .system_properties
+ .as_ref()
+ .unwrap()
+ .message_id
+ .clone();
+ let topic = message.topic.as_ref().unwrap().clone();
Review Comment:
Same here
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org