mjsax commented on a change in pull request #10102:
URL: https://github.com/apache/kafka/pull/10102#discussion_r574183054
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -776,9 +776,10 @@ void runOnce() {
log.debug("{} punctuators ran.", punctuated);
+ final long beforeCommitTs = now;
final int committed = maybeCommit();
totalCommittedSinceLastSummary += committed;
- final long commitLatency = advanceNowAndComputeLatency();
+ final long commitLatency = Math.max(now - beforeCommitTs, 0);
Review comment:
> so there shouldn't be a noticeable difference between advancing now at
the end of maybeCommit vs advancing it immediately after maybeCommit returns,
right?
Well, not from a runtime point of view.
> (and similarly move the lastCommitMs = now to after maybeCommit returns)
Exactly: we need to first advance `now` and than update `lastCommitMs` --
but I would rather not update `lastCommitMs` outside of `maybeCommit`, because
it breaks encapsulation?
> I'm also ok with just advancing now inside maybeCommit. The main thing
that felt off was just that we compute the latency for no reason inside
maybeCommit, and ignore the result
Yes, that is a little odd -- as I said, happy to change
`advancedNowAndComputeLatency` to `now = time.milliseconds()` within
`maybeCommit()`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
