You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "mattisonchao (via GitHub)" <gi...@apache.org> on 2025/05/05 15:26:54 UTC
[PR] feat(stats): add consumer first consume/flow timestamp [pulsar]
mattisonchao opened a new pull request, #24247:
URL: https://github.com/apache/pulsar/pull/24247
### Motivation
This PR supports two stats for the consumer side to help identify the first message dispatch latency.
### Modifications
- Add stats `firstConsumedFlowTimestamp`
- Add stats `firstConsumedTimestamp`
### Verifying this change
- [x] Make sure that the change passes the CI checks.
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [x] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][broker] add consumer first consume/flow timestamp [pulsar]
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao closed pull request #24247: [feat][broker] add consumer first consume/flow timestamp
URL: https://github.com/apache/pulsar/pull/24247
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][broker] add consumer first consume/flow timestamp [pulsar]
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on code in PR #24247:
URL: https://github.com/apache/pulsar/pull/24247#discussion_r2137061415
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java:
##########
@@ -435,6 +437,9 @@ public Future<Void> sendMessages(final List<? extends Entry> entries,
writeAndFlushPromise.addListener(status -> {
// only increment counters after the messages have been successfully written to the TCP/IP connection
if (status.isSuccess()) {
+ if (firstConsumedTimestamp == 0) {
+ firstConsumedTimestamp = System.currentTimeMillis();
+ }
Review Comment:
let me rename it to `firstMessagesSentTimestamp`.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][broker] add consumer first consume/flow timestamp [pulsar]
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on code in PR #24247:
URL: https://github.com/apache/pulsar/pull/24247#discussion_r2074351862
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java:
##########
@@ -435,6 +437,9 @@ public Future<Void> sendMessages(final List<? extends Entry> entries,
writeAndFlushPromise.addListener(status -> {
// only increment counters after the messages have been successfully written to the TCP/IP connection
if (status.isSuccess()) {
+ if (firstConsumedTimestamp == 0) {
+ firstConsumedTimestamp = System.currentTimeMillis();
+ }
Review Comment:
> oh sorry, there are 2 separate fields. Just wondering where lastConsumedTimestamp is set. wouldn't it be set in a similar way as firstConsumedTimestamp?
IMO, we should also move `lastConsumedTimestamp ` to success sent. :/
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][broker] add consumer first consume/flow timestamp [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24247:
URL: https://github.com/apache/pulsar/pull/24247#discussion_r2074160492
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java:
##########
@@ -435,6 +437,9 @@ public Future<Void> sendMessages(final List<? extends Entry> entries,
writeAndFlushPromise.addListener(status -> {
// only increment counters after the messages have been successfully written to the TCP/IP connection
if (status.isSuccess()) {
+ if (firstConsumedTimestamp == 0) {
+ firstConsumedTimestamp = System.currentTimeMillis();
+ }
Review Comment:
oh sorry, there are 2 separate fields. Just wondering where `lastConsumedTimestamp` is set. wouldn't it be set in a similar way as `firstConsumedTimestamp`?
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][broker] add consumer first consume/flow timestamp [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24247:
URL: https://github.com/apache/pulsar/pull/24247#discussion_r2074158432
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java:
##########
@@ -435,6 +437,9 @@ public Future<Void> sendMessages(final List<? extends Entry> entries,
writeAndFlushPromise.addListener(status -> {
// only increment counters after the messages have been successfully written to the TCP/IP connection
if (status.isSuccess()) {
+ if (firstConsumedTimestamp == 0) {
+ firstConsumedTimestamp = System.currentTimeMillis();
+ }
Review Comment:
just wondering if this makes sense since `flowPermits` method would get called before messages get sent to a consumer?
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][broker] add consumer first consume/flow timestamp [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24247:
URL: https://github.com/apache/pulsar/pull/24247#issuecomment-2893725543
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/24247?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 39.48%. Comparing base [(`bbc6224`)](https://app.codecov.io/gh/apache/pulsar/commit/bbc62245c5ddba1de4b1e7cee4ab49334bc36277?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`55c8ee0`)](https://app.codecov.io/gh/apache/pulsar/commit/55c8ee0de86c76ba918926b3541b4367558a469c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 1107 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/24247?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #24247 +/- ##
=============================================
- Coverage 73.57% 39.48% -34.10%
+ Complexity 32624 13406 -19218
=============================================
Files 1877 1809 -68
Lines 139502 145458 +5956
Branches 15299 17067 +1768
=============================================
- Hits 102638 57430 -45208
- Misses 28908 80526 +51618
+ Partials 7956 7502 -454
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/24247/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/24247/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.71% <85.71%> (+2.13%)` | :arrow_up: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/24247/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `23.25% <85.71%> (-1.07%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/24247/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `35.59% <100.00%> (-37.25%)` | :arrow_down: |
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.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24247?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...ava/org/apache/pulsar/broker/service/Consumer.java](https://app.codecov.io/gh/apache/pulsar/pull/24247?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FConsumer.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL0NvbnN1bWVyLmphdmE=) | `63.91% <100.00%> (-22.76%)` | :arrow_down: |
| [.../common/policies/data/stats/ConsumerStatsImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24247?src=pr&el=tree&filepath=pulsar-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fcommon%2Fpolicies%2Fdata%2Fstats%2FConsumerStatsImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NvbW1vbi9wb2xpY2llcy9kYXRhL3N0YXRzL0NvbnN1bWVyU3RhdHNJbXBsLmphdmE=) | `100.00% <ø> (+2.38%)` | :arrow_up: |
... and [1675 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/24247/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][broker] add consumer first consume/flow timestamp [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24247:
URL: https://github.com/apache/pulsar/pull/24247#discussion_r2137047408
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java:
##########
@@ -435,6 +437,9 @@ public Future<Void> sendMessages(final List<? extends Entry> entries,
writeAndFlushPromise.addListener(status -> {
// only increment counters after the messages have been successfully written to the TCP/IP connection
if (status.isSuccess()) {
+ if (firstConsumedTimestamp == 0) {
+ firstConsumedTimestamp = System.currentTimeMillis();
+ }
Review Comment:
> IMO, we should also move `lastConsumedTimestamp ` to success sent. That will be more accurate, but I don't want to make this PR more complex. :/
This is problematic when the definition is different between `firstConsumedTimestamp` and `lastConsumedTimestamp`. Renaming `firstConsumedTimestamp` would resolve this. I think that "consumed" is very vague and it would be better to make it more specific such as `firstMessagesSentTimestamp`, `firstMessagesWrittenTimestamp` or `firstMessagesFlushedTimestamp`.
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][broker] add consumer first consume/flow timestamp [pulsar]
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on code in PR #24247:
URL: https://github.com/apache/pulsar/pull/24247#discussion_r2074351862
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java:
##########
@@ -435,6 +437,9 @@ public Future<Void> sendMessages(final List<? extends Entry> entries,
writeAndFlushPromise.addListener(status -> {
// only increment counters after the messages have been successfully written to the TCP/IP connection
if (status.isSuccess()) {
+ if (firstConsumedTimestamp == 0) {
+ firstConsumedTimestamp = System.currentTimeMillis();
+ }
Review Comment:
> oh sorry, there are 2 separate fields. Just wondering where lastConsumedTimestamp is set. wouldn't it be set in a similar way as firstConsumedTimestamp?
IMO, we should also move `lastConsumedTimestamp ` to success sent. That will be more accurate, but I don't want to make this PR more complex. :/
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][broker] add consumer first consume/flow timestamp [pulsar]
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao merged PR #24247:
URL: https://github.com/apache/pulsar/pull/24247
--
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@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org