mjsax commented on code in PR #21201:
URL: https://github.com/apache/kafka/pull/21201#discussion_r2692737619
##########
docs/streams/upgrade-guide.md:
##########
@@ -61,6 +61,10 @@ Starting in Kafka Streams 2.6.x, a new processing mode is
available, named EOS v
Since 2.6.0 release, Kafka Streams depends on a RocksDB version that requires
MacOS 10.14 or higher.
+## Streams API changes in 4.3.0
+
+The streams thread metrics `commit-ratio`, `process-ratio`, `punctuate-ratio`,
and `poll-ratio`, along with streams state updater metrics
`active-restore-ratio`, `standby-restore-ratio`, `idle-ratio`, and
`checkpoint-ratio` have been updated. Each metric now reports, over a rolling
measurement window, the ratio of time this thread spends performing the given
action (`{action}`) to the total elapsed time in that window. The effective
window duration is determined by the metrics configuration:
`metrics.sample.window.ms` (per-sample window length)and `metrics.num.samples`
(number of rolling windows).
Review Comment:
```suggestion
The streams thread metrics `commit-ratio`, `process-ratio`,
`punctuate-ratio`, and `poll-ratio`, along with streams state updater metrics
`active-restore-ratio`, `standby-restore-ratio`, `idle-ratio`, and
`checkpoint-ratio` have been updated. Each metric now reports, over a rolling
measurement window, the ratio of time this thread spends performing the given
action (`{action}`) to the total elapsed time in that window. The effective
window duration is determined by the metrics configuration:
`metrics.sample.window.ms` (per-sample window length) and `metrics.num.samples`
(number of rolling windows).
```
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -713,19 +726,63 @@ private void measureCheckpointLatency(final Runnable
actionToMeasure) {
private void recordMetrics(final long now, final long totalLatency,
final long totalWaitLatency) {
final long totalRestoreLatency = Math.max(0L, totalLatency -
totalWaitLatency - totalCheckpointLatency);
- updaterMetrics.idleRatioSensor.record((double) totalWaitLatency /
totalLatency, now);
- updaterMetrics.checkpointRatioSensor.record((double)
totalCheckpointLatency / totalLatency, now);
+ recordWindowedSum(
+ now,
+ (double) totalWaitLatency,
Review Comment:
Do we need to cast explicitly? The parameter of `recordWindowedSum(...)` is
defined as `double`, so it seems the explicit cast is unnecessary? Same for the
rows below?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -1035,10 +1092,10 @@ private Stream<Task> streamOfNonPausedTasks() {
private class StateUpdaterMetrics {
private static final String STATE_LEVEL_GROUP =
"stream-state-updater-metrics";
- private static final String IDLE_RATIO_DESCRIPTION = RATIO_DESCRIPTION
+ "being idle";
- private static final String RESTORE_RATIO_DESCRIPTION =
RATIO_DESCRIPTION + "restoring active tasks";
- private static final String UPDATE_RATIO_DESCRIPTION =
RATIO_DESCRIPTION + "updating standby tasks";
- private static final String CHECKPOINT_RATIO_DESCRIPTION =
RATIO_DESCRIPTION + "checkpointing tasks restored progress";
+ private static final String IDLE_RATIO_DESCRIPTION =
WINDOWED_RATIO_DESCRIPTION_PREFIX + THREAD_TIME_UNIT_DESCRIPTION + "being
idle";
Review Comment:
```suggestion
private static final String IDLE_RATIO_DESCRIPTION =
WINDOWED_RATIO_DESCRIPTION_PREFIX + THREAD_TIME_UNIT_DESCRIPTION + "being idle";
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
