You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "mjsax (via GitHub)" <gi...@apache.org> on 2026/08/04 03:56:18 UTC
Re: [PR] KAFKA-14597: record-e2e-latency-max only considers consumption latencies but not processing delays [kafka]
mjsax commented on code in PR #22452:
URL: https://github.com/apache/kafka/pull/22452#discussion_r3708993547
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -1289,6 +1297,96 @@ void maybeRecordE2ELatency(final long recordTimestamp, final long now, final Str
}
}
+ /**
+ * Buffers a terminal node's record timestamp. The latency is recorded later by
+ * {@link #maybeFlushTerminalE2ELatency(long)}, so processing delay is captured without a
+ * wall-clock read per output record.
+ */
+ void maybeAddTerminalE2ELatency(final long recordTimestamp, final String nodeName) {
+ final Sensor e2eLatencySensor = e2eLatencySensors.get(nodeName);
+ if (e2eLatencySensor == null) {
+ throw new IllegalStateException("Requested to record e2e latency but could not find sensor for node " + nodeName);
+ } else if (e2eLatencySensor.shouldRecord() && e2eLatencySensor.hasMetrics()) {
+ terminalE2ELatencyAccumulators.get(nodeName).add(recordTimestamp);
+ }
+ }
+
+ /**
+ * Records terminal {@code record-e2e-latency} for all records buffered since the last flush,
+ * against a single already-read wall-clock time {@code endMs}. Emits the exact max and min
+ * latency and, for larger batches, fills the remaining samples with the interior-timestamp mean
+ * so the count-weighted average is exact as well. Emitting one sample per record keeps the
+ * metric's weighting unchanged; all records are attributed {@code endMs}, a conservative upper
+ * bound on their completion time.
+ */
+ @Override
+ public void maybeFlushTerminalE2ELatency(final long endMs) {
+ for (final Map.Entry<String, TerminalE2ELatencyAccumulator> entry : terminalE2ELatencyAccumulators.entrySet()) {
+ final TerminalE2ELatencyAccumulator accumulator = entry.getValue();
+ final long n = accumulator.count;
+ if (n > 0) {
+ final Sensor e2eLatencySensor = e2eLatencySensors.get(entry.getKey());
+ if (e2eLatencySensor.shouldRecord() && e2eLatencySensor.hasMetrics()) {
+ final double maxLatency = endMs - accumulator.minTs;
+ final double minLatency = endMs - accumulator.maxTs;
+ if (n == 1) {
+ e2eLatencySensor.record(maxLatency, endMs);
+ } else {
+ e2eLatencySensor.record(maxLatency, endMs);
+ e2eLatencySensor.record(minLatency, endMs);
+ if (n >= 3) {
+ final double interiorMeanTs = accumulator.baseTs
+ + (double) (accumulator.sumOffset
+ - (accumulator.minTs - accumulator.baseTs)
+ - (accumulator.maxTs - accumulator.baseTs)) / (n - 2);
+ // clamp against rounding so the filler can't move the sensor's max/min
+ final double filler = Math.min(maxLatency, Math.max(minLatency, endMs - interiorMeanTs));
+ for (long i = 0; i < n - 2; i++) {
+ e2eLatencySensor.record(filler, endMs);
+ }
+ }
+ }
+ }
+ accumulator.reset();
+ }
+ }
+ }
+
+ /**
+ * Per-terminal-node timestamp aggregates used to reconstruct e2e latency without keeping a
+ * sample per record. Timestamps are summed as offsets from the first one so the running sum
+ * stays small and overflow-safe.
+ */
+ private static final class TerminalE2ELatencyAccumulator {
+ private long count;
+ private long baseTs;
+ private long minTs;
+ private long maxTs;
+ private long sumOffset;
+
+ void add(final long recordTimestamp) {
+ if (count == 0) {
+ baseTs = recordTimestamp;
+ minTs = recordTimestamp;
+ maxTs = recordTimestamp;
+ sumOffset = 0L;
+ } else {
+ if (recordTimestamp < minTs) {
+ minTs = recordTimestamp;
+ }
+ if (recordTimestamp > maxTs) {
+ maxTs = recordTimestamp;
+ }
+ sumOffset += recordTimestamp - baseTs;
Review Comment:
This is clever. I like it!
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -1289,6 +1297,96 @@ void maybeRecordE2ELatency(final long recordTimestamp, final long now, final Str
}
}
+ /**
+ * Buffers a terminal node's record timestamp. The latency is recorded later by
+ * {@link #maybeFlushTerminalE2ELatency(long)}, so processing delay is captured without a
+ * wall-clock read per output record.
+ */
+ void maybeAddTerminalE2ELatency(final long recordTimestamp, final String nodeName) {
+ final Sensor e2eLatencySensor = e2eLatencySensors.get(nodeName);
+ if (e2eLatencySensor == null) {
+ throw new IllegalStateException("Requested to record e2e latency but could not find sensor for node " + nodeName);
+ } else if (e2eLatencySensor.shouldRecord() && e2eLatencySensor.hasMetrics()) {
+ terminalE2ELatencyAccumulators.get(nodeName).add(recordTimestamp);
+ }
+ }
+
+ /**
+ * Records terminal {@code record-e2e-latency} for all records buffered since the last flush,
+ * against a single already-read wall-clock time {@code endMs}. Emits the exact max and min
+ * latency and, for larger batches, fills the remaining samples with the interior-timestamp mean
+ * so the count-weighted average is exact as well. Emitting one sample per record keeps the
+ * metric's weighting unchanged; all records are attributed {@code endMs}, a conservative upper
+ * bound on their completion time.
+ */
+ @Override
+ public void maybeFlushTerminalE2ELatency(final long endMs) {
+ for (final Map.Entry<String, TerminalE2ELatencyAccumulator> entry : terminalE2ELatencyAccumulators.entrySet()) {
+ final TerminalE2ELatencyAccumulator accumulator = entry.getValue();
+ final long n = accumulator.count;
+ if (n > 0) {
+ final Sensor e2eLatencySensor = e2eLatencySensors.get(entry.getKey());
+ if (e2eLatencySensor.shouldRecord() && e2eLatencySensor.hasMetrics()) {
+ final double maxLatency = endMs - accumulator.minTs;
+ final double minLatency = endMs - accumulator.maxTs;
+ if (n == 1) {
+ e2eLatencySensor.record(maxLatency, endMs);
+ } else {
+ e2eLatencySensor.record(maxLatency, endMs);
+ e2eLatencySensor.record(minLatency, endMs);
+ if (n >= 3) {
+ final double interiorMeanTs = accumulator.baseTs
+ + (double) (accumulator.sumOffset
+ - (accumulator.minTs - accumulator.baseTs)
+ - (accumulator.maxTs - accumulator.baseTs)) / (n - 2);
+ // clamp against rounding so the filler can't move the sensor's max/min
+ final double filler = Math.min(maxLatency, Math.max(minLatency, endMs - interiorMeanTs));
+ for (long i = 0; i < n - 2; i++) {
+ e2eLatencySensor.record(filler, endMs);
+ }
+ }
+ }
+ }
+ accumulator.reset();
+ }
+ }
+ }
+
+ /**
+ * Per-terminal-node timestamp aggregates used to reconstruct e2e latency without keeping a
+ * sample per record. Timestamps are summed as offsets from the first one so the running sum
+ * stays small and overflow-safe.
+ */
+ private static final class TerminalE2ELatencyAccumulator {
+ private long count;
+ private long baseTs;
+ private long minTs;
+ private long maxTs;
+ private long sumOffset;
+
+ void add(final long recordTimestamp) {
+ if (count == 0) {
+ baseTs = recordTimestamp;
+ minTs = recordTimestamp;
+ maxTs = recordTimestamp;
+ sumOffset = 0L;
+ } else {
+ if (recordTimestamp < minTs) {
+ minTs = recordTimestamp;
+ }
+ if (recordTimestamp > maxTs) {
+ maxTs = recordTimestamp;
+ }
Review Comment:
Should we use `Math.min(...)`, and `Math.max(...)` -- does the same thing but might be easier to read?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -1006,6 +1012,8 @@ record = null;
droppedRecordsSensor.record();
}
} finally {
+ // flush e2e latency for any records the punctuator forwarded to terminal nodes
+ maybeFlushTerminalE2ELatency(punctuateWallClockMs);
Review Comment:
Claude pointed out, that this is the ts captured before the punctuator ran. -- So it seems we cannot re-use a cached value.
Given that punctuators don't run on every _input record_, but time based, and usually in the seconds or even minutes, it should be ok to just pass int `time.milliseconds()` here, and pay the system-call.
We also don't test that this flush is happening at all.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutor.java:
##########
@@ -124,6 +124,8 @@ private int processTask(final Task task, final int maxNumRecords, final long beg
} finally {
now = time.milliseconds();
task.recordProcessBatchTime(now - then);
+ // record terminal e2e latency for the batch against this single end-of-batch time
+ task.maybeFlushTerminalE2ELatency(now);
Review Comment:
Same. This is not covered by any test
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -928,7 +932,9 @@ public void punctuate(final ProcessorNode<?, ?, ?, ?> node,
null,
new RecordHeaders()
);
- updateProcessorContext(node, time.milliseconds(), recordContext);
+ // reused as the terminal e2e flush time below, so punctuation needs no extra clock read
+ final long punctuateWallClockMs = time.milliseconds();
+ updateProcessorContext(node, punctuateWallClockMs, recordContext);
Review Comment:
Seems you are trying to cache the ts for later usage -- I think this does not work. Cf comment blow.
If we really want to save a system-call here, I think we could also pass in `now` into this call though?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -845,8 +850,10 @@ public void process(final Record<Integer, Integer> record) {
// e2e latency = 23
+ time.sleep(13L);
Review Comment:
Do we need to advance time here?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskExecutor.java:
##########
@@ -181,7 +181,10 @@ private boolean processTask(final Task task, final long now, final Time time) {
log.error(String.format("Failed to process stream task %s due to the following error:", task.id()), e);
throw new StreamsException(e, task.id());
} finally {
- task.recordProcessBatchTime(time.milliseconds() - now);
+ final long end = time.milliseconds();
+ task.recordProcessBatchTime(end - now);
+ // record terminal e2e latency against the end time already read for the batch metric
+ task.maybeFlushTerminalE2ELatency(end);
Review Comment:
Claude highlighted, that this call is crucial for correctness, but it's not covered by any tests.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -817,8 +818,11 @@ public void process(final Record<Integer, Integer> record) {
final Metric terminalMax = getProcessorMetric("record-e2e-latency", "%s-max", task.id().toString(), terminalNodeName);
// e2e latency = 10
+ time.sleep(10L);
Review Comment:
Do we need to advance time here?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org