Repository navigation
[Issue 1154][consumer] Fix bogus values in the processing time metric - #1550
Merged
nodece merged 1 commit intoOct 24, 2026
Merged
Conversation
nodece
approved these changes
Oct 21, 2026
The pulsar_client_consumer_processing_time_seconds histogram records values of about 8.5e9 seconds. On an individual ack, ackIDCommon replaces trackingID with a new ID that has no receive time and then observes the metric from that ID, so it measures from the zero time.Time. Cumulative acks of chunked messages hit the same problem, because toTrackingMessageID does not keep the receive time of a chunk ID. Individual acks of chunked messages recorded one such value per chunk, because each chunk is acked by its plain message ID. Individual and cumulative acks also did this for IDs that the consumer never received, such as ones built by the application. Add observeProcessingTime, which reads the receive time from the acked tracking or chunk ID and skips IDs that have none. Use it in the three places that observed the metric, and observe a chunked message once in the individual ack and AckIDList paths, so chunked messages are recorded once per message on every ack path. Signed-off-by: Yashvardhan Thanvi <[email protected]>
Thanatos9404
force-pushed
the
fix/consumer-processing-time-metric
branch
from
October 21, 2026 05:51
3c37764 to
c18e596
Compare
Contributor
Author
|
Thanks for the review! I rebased onto master so the integration tests pick up #1545: the three failures in the last run (TestTxn_TCClient, TestSetTopicAutoCreation, TestPartitionState) are the ones that change fixed. The commit itself is unchanged, and the new run is waiting for workflow approval. |
This file contains hidden or 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
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.
Fixes #1154
Motivation
pulsar_client_consumer_processing_time_secondsshould measure the time from receiving a message to acking it, but almost every observation is about 8.5e9 seconds.ackIDCommonobserves the metric from a rebuilt tracking ID whosereceivedTimeis zero. Chunked messages, and IDs the consumer never received, produced bogus values too.Modifications
partitionConsumer.observeProcessingTime(msgID), which reads the receive time from the acked tracking or chunk message ID and records nothing when there is none.ackIDCommon(with the acked ID),AckIDListandinternalAckIDCumulative, so a chunked message is observed once on every ack path.time.Since. A batch is still observed once.Verifying this change
This change added tests and can be verified as follows:
TestProcessingTimeMetric,TestProcessingTimeMetricBatchMessageandTestProcessingTimeMetricSkipsIDsWithoutReceiveTimeinpulsar/consumer_partition_test.go. They don't need a broker. The new cases fail without the fix and pass with it, except the plainAckIDListandAckIDCumulativeones, which guard paths that already worked.Does this pull request potentially affect one of the following parts:
Documentation