feat: add kafka support #56
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
brayniac
requested changes
Oct 16, 2023
There was a problem hiding this comment.
clippy found more than 10 potential problems in the proposed changes. Check the Files changed tab for more details.
brayniac
reviewed
Oct 16, 2023
brayniac
reviewed
Oct 16, 2023
brayniac
reviewed
Oct 18, 2023
brayniac
reviewed
Oct 18, 2023
brayniac
reviewed
Oct 18, 2023
brayniac
reviewed
Oct 18, 2023
brayniac
reviewed
Oct 18, 2023
brayniac
reviewed
Oct 18, 2023
brayniac
reviewed
Oct 18, 2023
brayniac
reviewed
Oct 18, 2023
brayniac
reviewed
Oct 18, 2023
brayniac
reviewed
Oct 18, 2023
brayniac
reviewed
Oct 18, 2023
brayniac
reviewed
Oct 18, 2023
|
That’s because you’re using the wrong counter name. Should be
PUBSUB_PUBLISH_COUNTER like it was before?
…On Tue, Oct 17, 2023 at 6:09 PM Xi Yang ***@***.***> wrote:
***@***.**** commented on this pull request.
------------------------------
In src/config/workload.rs
<#56 (comment)>:
> topic_len: usize,
message_len: usize,
+ #[serde(default = "four")]
+ key_len: usize,
Since we explicitly set the partition when producing one message, we don't
need to control the keys, as long as there is a key, it's OK. 4 is just a
random number, I can change it to at least one.
------------------------------
In src/metrics/mod.rs
<#56 (comment)>:
> @@ -589,7 +589,10 @@ request!(SORTED_SET_SCORE, "sorted_set_score");
* PUBSUB
*/
-request!(PUBSUB_PUBLISH, "publisher/publish");
+counter!(PUBSUB_PUBLISH, "publisher/publish/total");
But using request! doesn't work. The
snapshot.counter_rate(REQUEST_COUNTER) returns 0. Here is the output when
using request!(PUBSUB_PUBLISH):
2023-10-18T00:06:25.662+00:00 Publishers: Current: 1
2023-10-18T00:06:25.662+00:00 Publisher Publish: Success: NaN % Timeout:
NaN %
2023-10-18T00:06:25.662+00:00 Publisher Publish Rate (/s): Ok: 0.00 Error:
0.00 Timeout: 0.00
2023-10-18T00:06:25.662+00:00 Subscribers: Current: 1
2023-10-18T00:06:25.662+00:00 Subscriber Receive: Success: 100.00 %
Corrupted: 0.00 %
2023-10-18T00:06:25.662+00:00 Subscriber Receive Rate (/s): Ok: 99.65
Error: 0.00 Corrupt: 0.00 Invalid: 0.00
2023-10-18T00:06:25.662+00:00 Pubsub Publish Latency (us): p25: 5341 p50:
5373 p75: 5373 p90: 10944 p99: 19136 p999: 19136 p9999: 19136
2023-10-18T00:06:25.662+00:00 Pubsub End-to-End Latency (us): p25: 5373
p50: 5406 p75: 5406 p90: 10944 p99: 19136 p999: 19136 p9999: 19136
—
Reply to this email directly, view it on GitHub
<#56 (comment)>,
or unsubscribe
<https://github.com/notifications/unsubscribe-auth/ACP6HFYF7WZSATOD6SU7TWLX74T5BAVCNFSM6AAAAAA6CZMHYKVHI2DSMVQWIX3LMV43YUDVNRWFEZLROVSXG5CSMV3GSZLXHMYTMOBTG4YDSNZZGY>
.
You are receiving this because you commented.Message ID:
***@***.***>
|
brayniac
reviewed
Oct 18, 2023
src/pubsub/mod.rs
Outdated
| mod momento; | ||
|
|
||
| struct MessageStamp { |
There was a problem hiding this comment.
Should we call this a message validator?
src/pubsub/kafka.rs
Outdated
| } | ||
| } | ||
|
|
||
| pub fn create_topics(runtime: &mut Runtime, config: Config, workload_components: &Vec<Component>) { |
There was a problem hiding this comment.
Please make this &[Component]
src/config/workload.rs
Outdated
| @@ -4,6 +4,10 @@ fn one() -> usize { | |||
| 1 | |||
| } | |||
|
|
|||
| fn empty_names() -> Vec<String> { | |||
brayniac
reviewed
Oct 18, 2023
src/config/workload.rs
Outdated
| topic_len: usize, | ||
| #[serde(default = "empty_names")] |
There was a problem hiding this comment.
Just #[serde(default)] does what you want
brayniac
reviewed
Oct 18, 2023
src/pubsub/kafka.rs
Outdated
| Ok(m) => match m.payload_view::<[u8]>() { | ||
| Some(Ok(m)) => { | ||
| let mut v = Vec::new(); | ||
| v.extend_from_slice(m); |
There was a problem hiding this comment.
I think you can just do let mut v = m.to_owned();
brayniac
reviewed
Oct 18, 2023
src/pubsub/mod.rs
Outdated
| @@ -29,9 +100,20 @@ pub fn launch_pubsub( | |||
| work_receiver: Receiver<WorkItem>, | |||
| workload_components: Vec<Component>, | |||
| ) -> PubsubRuntimes { | |||
| if config.pubsub().is_some() { | |||
| if let Protocol::Kafka = config.general().protocol() { | |||
| let mut topic_rt = Builder::new_multi_thread() | |||
There was a problem hiding this comment.
Can we add some comment here explaining why we need an additional Tokio runtime for this. I'm unclear why we wouldn't handle this differently
There was a problem hiding this comment.
I moved kafka::create_topics to lauch_publishers, so it reuses the publisher_rt to create the topics.
brayniac
requested changes
Oct 18, 2023
brayniac
approved these changes
Oct 19, 2023
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
No description provided.