You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2026/08/10 22:37:02 UTC
(kafka) branch 4.2 updated: KAFKA-14597: record-e2e-latency-max only considers consumption latencies but not processing delays (#22452)
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch 4.2
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.2 by this push:
new 30b4e288122 KAFKA-14597: record-e2e-latency-max only considers consumption latencies but not processing delays (#22452)
30b4e288122 is described below
commit 30b4e288122b2899a1c38192b037a4a2c1b63ae2
Author: Alan Lau <72...@users.noreply.github.com>
AuthorDate: Mon Aug 10 18:22:03 2026 -0400
KAFKA-14597: record-e2e-latency-max only considers consumption latencies but not processing delays (#22452)
The record-e2e-latency metric at terminal nodes is computed incorrectly,
using a cached timestamp which is take before actual processing happens.
This PR ensures, that we are taking a new timestamp after processing
completed. To avoid per-record system calls, we take a single timestamp
after a batch of records completed. During processing, we capture
record.timestamp values (eg, min, max, etc) allowing us to compute the
latency per record at the end of a batch.
Reviewers: Lucy Liu <lu...@confluent.io>, Matthias J. Sax
<ma...@confluent.io>
---
docs/operations/monitoring.md | 2 +-
.../processor/internals/ProcessorContextImpl.java | 2 +-
.../streams/processor/internals/ReadOnlyTask.java | 5 +
.../streams/processor/internals/StreamTask.java | 96 ++++++++
.../kafka/streams/processor/internals/Task.java | 10 +
.../streams/processor/internals/TaskExecutor.java | 2 +
.../internals/tasks/DefaultTaskExecutor.java | 5 +-
.../processor/internals/StreamTaskTest.java | 263 +++++++++++++++++++++
.../processor/internals/TaskExecutorTest.java | 49 ++++
.../internals/tasks/DefaultTaskExecutorTest.java | 1 +
.../apache/kafka/streams/TopologyTestDriver.java | 4 +
11 files changed, 436 insertions(+), 3 deletions(-)
diff --git a/docs/operations/monitoring.md b/docs/operations/monitoring.md
index 0b7ceab67ef..db1c14bab18 100644
--- a/docs/operations/monitoring.md
+++ b/docs/operations/monitoring.md
@@ -4776,7 +4776,7 @@ kafka.streams:type=stream-task-metrics,thread-id=([-.\w]+),task-id=([-.\w]+)
### Processor Node Metrics
-The following metrics are only available on certain types of nodes, i.e., the process-* metrics are only available for source processor nodes, the `suppression-emit-*` metrics are only available for suppression operation nodes, `emit-final-*` metrics are only available for windowed aggregations nodes, and the `record-e2e-latency-*` metrics are only available for source processor nodes and terminal nodes (nodes without successor nodes). All the metrics have a recording level of `debug`, e [...]
+The following metrics are only available on certain types of nodes, i.e., the process-* metrics are only available for source processor nodes, the `suppression-emit-*` metrics are only available for suppression operation nodes, `emit-final-*` metrics are only available for windowed aggregations nodes, and the `record-e2e-latency-*` metrics are only available for source processor nodes and terminal nodes (nodes without successor nodes). At terminal nodes every record processed in the same [...]
<table>
<tr>
<th>
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java
index 26f724b75f4..0bc83e52103 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java
@@ -294,7 +294,7 @@ public final class ProcessorContextImpl extends AbstractProcessorContext<Object,
child.process(record);
if (child.isTerminalNode()) {
- streamTask.maybeRecordE2ELatency(record.timestamp(), currentSystemTimeMs(), child.name());
+ streamTask.maybeAddTerminalE2ELatency(record.timestamp(), child.name());
}
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java
index dd5a2c6e1d7..a2dc9a69402 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java
@@ -164,6 +164,11 @@ public class ReadOnlyTask implements Task {
throw new UnsupportedOperationException("This task is read-only");
}
+ @Override
+ public void maybeFlushTerminalE2ELatency(final long endMs) {
+ throw new UnsupportedOperationException("This task is read-only");
+ }
+
@Override
public void recordProcessTimeRatioAndBufferSize(final long allTaskProcessMs, final long now) {
throw new UnsupportedOperationException("This task is read-only");
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
index 85c322f2b5f..36610c445e3 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
@@ -107,6 +107,9 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator,
private final Sensor bufferedRecordsSensor;
private final Sensor droppedRecordsSensor;
private final Map<String, Sensor> e2eLatencySensors = new HashMap<>();
+ // per-terminal-node timestamp aggregates for the current batch/punctuation, flushed against a
+ // single wall-clock time in maybeFlushTerminalE2ELatency so we avoid a clock read per record
+ private final Map<String, TerminalE2ELatencyAccumulator> terminalE2ELatencyAccumulators = new HashMap<>();
private final RecordQueueCreator recordQueueCreator;
@@ -174,6 +177,7 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator,
terminalNodeName,
ProcessorNodeMetrics.e2ELatencySensor(threadId, taskId, terminalNodeName, streamsMetrics)
);
+ terminalE2ELatencyAccumulators.put(terminalNodeName, new TerminalE2ELatencyAccumulator());
}
for (final ProcessorNode<?, ?, ?, ?> sourceNode : topology.sources()) {
@@ -429,6 +433,10 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator,
public void flush() {
stateMgr.flushCache();
+ // flushing the caches may evict records into terminal nodes, so record their e2e latency
+ // before we leave the processing path; otherwise those samples would either be attributed
+ // to the next batch's end time or dropped entirely if the task never processes again
+ maybeFlushTerminalE2ELatency(time.milliseconds());
recordCollector.flush();
}
@@ -1022,6 +1030,10 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator,
droppedRecordsSensor.record();
}
} finally {
+ // flush e2e latency for any records the punctuator forwarded to terminal nodes; the clock
+ // must be re-read because the punctuator itself is part of the latency we are measuring,
+ // and punctuators fire on a schedule so the extra read is not on the per-record path
+ maybeFlushTerminalE2ELatency(time.milliseconds());
processorContext.setCurrentNode(null);
}
}
@@ -1305,6 +1317,90 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator,
}
}
+ /**
+ * 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;
+ e2eLatencySensor.record(maxLatency, endMs);
+ if (n >= 2) {
+ 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 {
+ minTs = Math.min(minTs, recordTimestamp);
+ maxTs = Math.max(maxTs, recordTimestamp);
+ sumOffset += recordTimestamp - baseTs;
+ }
+ count++;
+ }
+
+ void reset() {
+ count = 0L;
+ }
+ }
+
/**
* Request committing the current task's state
*/
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
index ba09700af8a..10f198fe8ed 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
@@ -187,6 +187,16 @@ public interface Task {
default void recordProcessBatchTime(final long processBatchTime) {
}
+ /**
+ * Record terminal-node e2e latency for everything buffered since the last flush against a single
+ * wall-clock time. Must be called on every path that can forward to a terminal node -- the end of
+ * a processing batch, punctuation, and flushing the caches -- with a clock read taken after that
+ * work completed, so the processing delay is part of the measured latency.
+ * No-op for tasks that do not forward to terminal nodes.
+ */
+ default void maybeFlushTerminalE2ELatency(final long endMs) {
+ }
+
default void recordProcessTimeRatioAndBufferSize(final long allTaskProcessMs, final long now) {
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutor.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutor.java
index 8294bf407d0..be3159b5aa7 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutor.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutor.java
@@ -124,6 +124,8 @@ public class TaskExecutor {
} 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);
}
return processed;
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskExecutor.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskExecutor.java
index fc7550952be..5c78fe9a6d5 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskExecutor.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskExecutor.java
@@ -181,7 +181,10 @@ public class DefaultTaskExecutor implements TaskExecutor {
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);
}
return processed;
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
index 16a9ddeddf4..76f4c539f38 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
@@ -129,6 +129,7 @@ import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.closeTo;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.hasItem;
import static org.hamcrest.Matchers.is;
@@ -809,6 +810,8 @@ public class StreamTaskTest {
// e2e latency = 10
task.addRecords(partition1, singletonList(getConsumerRecordWithOffsetAsTimestamp(0, 0L)));
task.process(10L);
+ // flush terminal e2e for the batch, as TaskExecutor.processTask does
+ task.maybeFlushTerminalE2ELatency(10L);
assertThat(sourceAvg.metricValue(), equalTo(10.0));
assertThat(sourceMin.metricValue(), equalTo(10.0));
@@ -823,6 +826,7 @@ public class StreamTaskTest {
// e2e latency = 15
task.addRecords(partition1, singletonList(getConsumerRecordWithOffsetAsTimestamp(1, 0L)));
task.process(15L);
+ task.maybeFlushTerminalE2ELatency(15L);
assertThat(sourceAvg.metricValue(), equalTo(12.5));
assertThat(sourceMin.metricValue(), equalTo(10.0));
@@ -837,6 +841,7 @@ public class StreamTaskTest {
// e2e latency = 23
task.addRecords(partition1, singletonList(getConsumerRecordWithOffsetAsTimestamp(2, 0L)));
task.process(23L);
+ task.maybeFlushTerminalE2ELatency(23L);
assertThat(sourceAvg.metricValue(), equalTo(16.0));
assertThat(sourceMin.metricValue(), equalTo(10.0));
@@ -851,6 +856,7 @@ public class StreamTaskTest {
// e2e latency = 5
task.addRecords(partition1, singletonList(getConsumerRecordWithOffsetAsTimestamp(3, 0L)));
task.process(5L);
+ task.maybeFlushTerminalE2ELatency(5L);
assertThat(sourceAvg.metricValue(), equalTo(13.25));
assertThat(sourceMin.metricValue(), equalTo(5.0));
@@ -862,6 +868,225 @@ public class StreamTaskTest {
assertThat(terminalMax.metricValue(), equalTo(23.0));
}
+ @Test
+ public void shouldIncludeProcessingDelayInTerminalNodeE2ELatency() {
+ when(stateManager.taskId()).thenReturn(taskId);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+ time = new MockTime(0L, 0L, 0L);
+ metrics = new Metrics(new MetricConfig().recordLevel(Sensor.RecordingLevel.INFO), time);
+
+ final long processingDelay = 5L;
+
+ // A source node that simulates processing work by advancing wall-clock time before forwarding
+ final MockSourceNode<Integer, Integer> delayingSourceNode = new MockSourceNode<>(intDeserializer, intDeserializer) {
+ InternalProcessorContext<Integer, Integer> context;
+
+ @Override
+ public void init(final InternalProcessorContext<Integer, Integer> context) {
+ this.context = context;
+ super.init(context);
+ }
+
+ @Override
+ public void process(final Record<Integer, Integer> record) {
+ time.sleep(processingDelay);
+ context.forward(record);
+ }
+ };
+
+ task = createStatelessTaskWithForwardingTopology(delayingSourceNode);
+ task.initializeIfNeeded();
+ task.completeRestoration(noOpResetter -> { });
+
+ final String sourceNodeName = delayingSourceNode.name();
+ final String terminalNodeName = processorStreamTime.name();
+
+ final Metric sourceMax = getProcessorMetric("record-e2e-latency", "%s-max", task.id().toString(), sourceNodeName);
+ final Metric terminalMax = getProcessorMetric("record-e2e-latency", "%s-max", task.id().toString(), terminalNodeName);
+
+ // record (timestamp 0) is observed by the source node at wall-clock time 10
+ time.sleep(10L);
+ task.addRecords(partition1, singletonList(getConsumerRecordWithOffsetAsTimestamp(0, 0L)));
+ task.process(time.milliseconds());
+ // the source advanced the clock while processing, so the batch-end flush time includes the delay
+ task.maybeFlushTerminalE2ELatency(time.milliseconds());
+
+ // source node measures only consumption latency (record timestamp -> source observed it)
+ assertThat(sourceMax.metricValue(), equalTo(10.0));
+ // terminal node measures full end-to-end latency, including the processing delay
+ assertThat(terminalMax.metricValue(), equalTo(10.0 + processingDelay));
+ }
+
+ @Test
+ public void shouldReconstructTerminalE2ELatencyForBatchWithSkewedTimestamps() {
+ when(stateManager.taskId()).thenReturn(taskId);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+ time = new MockTime(0L, 0L, 0L);
+ metrics = new Metrics(new MetricConfig().recordLevel(Sensor.RecordingLevel.INFO), time);
+
+ // Source node that forwards every record to the terminal node.
+ final MockSourceNode<Integer, Integer> forwardingSourceNode = new MockSourceNode<>(intDeserializer, intDeserializer) {
+ InternalProcessorContext<Integer, Integer> context;
+
+ @Override
+ public void init(final InternalProcessorContext<Integer, Integer> context) {
+ this.context = context;
+ super.init(context);
+ }
+
+ @Override
+ public void process(final Record<Integer, Integer> record) {
+ context.forward(record);
+ }
+ };
+
+ task = createStatelessTaskWithForwardingTopology(forwardingSourceNode);
+ task.initializeIfNeeded();
+ task.completeRestoration(noOpResetter -> { });
+
+ final String terminalNodeName = processorStreamTime.name();
+ final Metric terminalAvg = getProcessorMetric("record-e2e-latency", "%s-avg", task.id().toString(), terminalNodeName);
+ final Metric terminalMin = getProcessorMetric("record-e2e-latency", "%s-min", task.id().toString(), terminalNodeName);
+ final Metric terminalMax = getProcessorMetric("record-e2e-latency", "%s-max", task.id().toString(), terminalNodeName);
+
+ // five records with skewed timestamps (= offsets) reach the terminal node in one batch
+ final long[] timestamps = {100L, 450L, 480L, 490L, 500L};
+ for (int i = 0; i < timestamps.length; i++) {
+ task.addRecords(partition1, singletonList(getConsumerRecordWithOffsetAsTimestamp(i, timestamps[i])));
+ task.process(1000L);
+ }
+
+ // batch completes at 1000; true latencies [900,550,520,510,500] -> max 900, min 500, avg 596
+ task.maybeFlushTerminalE2ELatency(1000L);
+
+ assertThat(terminalMax.metricValue(), equalTo(900.0));
+ assertThat(terminalMin.metricValue(), equalTo(500.0));
+ assertThat((double) terminalAvg.metricValue(), closeTo(596.0, 1e-9));
+ }
+
+ @Test
+ public void shouldReconstructTerminalE2ELatencyForBatchOfTwoRecords() {
+ when(stateManager.taskId()).thenReturn(taskId);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+ time = new MockTime(0L, 0L, 0L);
+ metrics = new Metrics(new MetricConfig().recordLevel(Sensor.RecordingLevel.INFO), time);
+
+ final MockSourceNode<Integer, Integer> forwardingSourceNode = new MockSourceNode<>(intDeserializer, intDeserializer) {
+ InternalProcessorContext<Integer, Integer> context;
+
+ @Override
+ public void init(final InternalProcessorContext<Integer, Integer> context) {
+ this.context = context;
+ super.init(context);
+ }
+
+ @Override
+ public void process(final Record<Integer, Integer> record) {
+ context.forward(record);
+ }
+ };
+
+ task = createStatelessTaskWithForwardingTopology(forwardingSourceNode);
+ task.initializeIfNeeded();
+ task.completeRestoration(noOpResetter -> { });
+
+ final String terminalNodeName = processorStreamTime.name();
+ final Metric terminalAvg = getProcessorMetric("record-e2e-latency", "%s-avg", task.id().toString(), terminalNodeName);
+ final Metric terminalMin = getProcessorMetric("record-e2e-latency", "%s-min", task.id().toString(), terminalNodeName);
+ final Metric terminalMax = getProcessorMetric("record-e2e-latency", "%s-max", task.id().toString(), terminalNodeName);
+
+ // exactly two records reach the terminal node, so both are emitted exactly and no interior
+ // filler sample is needed to make the count-weighted average exact
+ final long[] timestamps = {100L, 400L};
+ for (int i = 0; i < timestamps.length; i++) {
+ task.addRecords(partition1, singletonList(getConsumerRecordWithOffsetAsTimestamp(i, timestamps[i])));
+ task.process(1000L);
+ }
+
+ // batch completes at 1000; true latencies [900,600] -> max 900, min 600, avg 750
+ task.maybeFlushTerminalE2ELatency(1000L);
+
+ assertThat(terminalMax.metricValue(), equalTo(900.0));
+ assertThat(terminalMin.metricValue(), equalTo(600.0));
+ assertThat((double) terminalAvg.metricValue(), closeTo(750.0, 1e-9));
+ }
+
+ @Test
+ public void shouldIncludePunctuatorDelayInTerminalNodeE2ELatency() {
+ when(stateManager.taskId()).thenReturn(taskId);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+ time = new MockTime(0L, 0L, 0L);
+ metrics = new Metrics(new MetricConfig().recordLevel(Sensor.RecordingLevel.INFO), time);
+
+ final long punctuatorDelay = 7L;
+
+ // A node whose punctuator simulates work by advancing wall-clock time before forwarding
+ final AtomicReference<InternalProcessorContext<Integer, Integer>> nodeContext = new AtomicReference<>();
+ final MockProcessorNode<Integer, Integer, Integer, Integer> punctuatingNode = new MockProcessorNode<>(10L) {
+ @Override
+ public void init(final InternalProcessorContext<Integer, Integer> context) {
+ super.init(context);
+ nodeContext.set(context);
+ }
+ };
+
+ task = createStatelessTaskWithPunctuatingTopology(punctuatingNode);
+ task.initializeIfNeeded();
+ task.completeRestoration(noOpResetter -> { });
+
+ final Metric terminalMax = getProcessorMetric("record-e2e-latency", "%s-max", task.id().toString(), processorStreamTime.name());
+
+ // punctuation fires at wall-clock time 20 and forwards a record with timestamp 0
+ time.sleep(20L);
+ task.punctuate(punctuatingNode, time.milliseconds(), PunctuationType.WALL_CLOCK_TIME, timestamp -> {
+ time.sleep(punctuatorDelay);
+ nodeContext.get().forward(new Record<>(1, 1, 0L));
+ });
+
+ // the latency must include the time the punctuator itself spent before forwarding
+ assertThat(terminalMax.metricValue(), equalTo(20.0 + punctuatorDelay));
+ }
+
+ @Test
+ public void shouldFlushPendingTerminalE2ELatencyOnTaskFlush() {
+ when(stateManager.taskId()).thenReturn(taskId);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+ time = new MockTime(0L, 0L, 0L);
+ metrics = new Metrics(new MetricConfig().recordLevel(Sensor.RecordingLevel.INFO), time);
+
+ final MockSourceNode<Integer, Integer> forwardingSourceNode = new MockSourceNode<>(intDeserializer, intDeserializer) {
+ InternalProcessorContext<Integer, Integer> context;
+
+ @Override
+ public void init(final InternalProcessorContext<Integer, Integer> context) {
+ this.context = context;
+ super.init(context);
+ }
+
+ @Override
+ public void process(final Record<Integer, Integer> record) {
+ context.forward(record);
+ }
+ };
+
+ task = createStatelessTaskWithForwardingTopology(forwardingSourceNode);
+ task.initializeIfNeeded();
+ task.completeRestoration(noOpResetter -> { });
+
+ final Metric terminalMax = getProcessorMetric("record-e2e-latency", "%s-max", task.id().toString(), processorStreamTime.name());
+
+ // a record reaches the terminal node but the batch-end flush has not run yet
+ time.sleep(10L);
+ task.addRecords(partition1, singletonList(getConsumerRecordWithOffsetAsTimestamp(0, 0L)));
+ task.process(time.milliseconds());
+
+ // flushing the caches can emit to terminal nodes, so flush() must drain what is buffered
+ time.sleep(5L);
+ task.flush();
+
+ assertThat(terminalMax.metricValue(), equalTo(15.0));
+ }
+
@Test
public void shouldRecordRestoredRecords() {
when(stateManager.taskId()).thenReturn(taskId);
@@ -3534,6 +3759,44 @@ public class StreamTaskTest {
);
}
+ private StreamTask createStatelessTaskWithPunctuatingTopology(final ProcessorNode<Integer, Integer, Integer, Integer> punctuatingNode) {
+ // wire up the children before building the topology so that only the leaf is registered as terminal
+ source1.addChild(punctuatingNode);
+ punctuatingNode.addChild(processorStreamTime);
+
+ final ProcessorTopology topology = withSources(
+ asList(source1, punctuatingNode, processorStreamTime),
+ singletonMap(topic1, source1)
+ );
+
+ final StreamsConfig config = createConfig();
+
+ final InternalProcessorContext<?, ?> context = new ProcessorContextImpl(
+ taskId,
+ config,
+ stateManager,
+ streamsMetrics,
+ null
+ );
+
+ return new StreamTask(
+ taskId,
+ singleton(partition1),
+ topology,
+ consumer,
+ new TopologyConfig(null, config, new Properties()).getTaskConfig(),
+ new StreamsMetricsImpl(metrics, "test", time),
+ stateDirectory,
+ cache,
+ time,
+ stateManager,
+ recordCollector,
+ context,
+ logContext,
+ false
+ );
+ }
+
private void createTimeoutTask(final String eosConfig) {
final ProcessorTopology topology = withSources(
singletonList(timeoutSource),
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskExecutorTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskExecutorTest.java
index 568998931c9..6ae4450bb18 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskExecutorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskExecutorTest.java
@@ -18,12 +18,19 @@ package org.apache.kafka.streams.processor.internals;
import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.streams.errors.StreamsException;
import org.junit.jupiter.api.Test;
import java.util.Collections;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.kafka.streams.internals.StreamsConfigUtils.ProcessingMode.EXACTLY_ONCE_V2;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -59,4 +66,46 @@ public class TaskExecutorTest {
verify(producer).commitTransaction(Collections.emptyMap(), groupMetadata);
}
+
+ @Test
+ public void shouldFlushTerminalE2ELatencyAgainstTheBatchEndTime() {
+ final Tasks tasks = mock(Tasks.class);
+ final TaskManager taskManager = mock(TaskManager.class);
+ final TaskExecutionMetadata metadata = mock(TaskExecutionMetadata.class);
+ final StreamTask task = mock(StreamTask.class);
+ final MockTime time = new MockTime(0L, 0L, 0L);
+
+ when(tasks.activeInitializedTasks()).thenReturn(Collections.singleton(task));
+ when(metadata.canProcessTask(eq(task), anyLong())).thenReturn(true);
+ // every step of the batch takes 5ms, and the batch ends once process() returns false
+ final AtomicInteger processCalls = new AtomicInteger();
+ when(task.process(anyLong())).thenAnswer(invocation -> {
+ time.sleep(5L);
+ return processCalls.getAndIncrement() < 2;
+ });
+
+ final TaskExecutor taskExecutor = new TaskExecutor(tasks, taskManager, metadata, new LogContext());
+
+ assertEquals(2, taskExecutor.process(10, time));
+ // three calls to process() means the batch ends at 15, not at the time it began
+ verify(task).maybeFlushTerminalE2ELatency(15L);
+ }
+
+ @Test
+ public void shouldFlushTerminalE2ELatencyWhenProcessingThrows() {
+ final Tasks tasks = mock(Tasks.class);
+ final TaskManager taskManager = mock(TaskManager.class);
+ final TaskExecutionMetadata metadata = mock(TaskExecutionMetadata.class);
+ final StreamTask task = mock(StreamTask.class);
+ final MockTime time = new MockTime(0L, 0L, 0L);
+
+ when(tasks.activeInitializedTasks()).thenReturn(Collections.singleton(task));
+ when(metadata.canProcessTask(eq(task), anyLong())).thenReturn(true);
+ when(task.process(anyLong())).thenThrow(new RuntimeException("KABOOM!"));
+
+ final TaskExecutor taskExecutor = new TaskExecutor(tasks, taskManager, metadata, new LogContext());
+
+ assertThrows(StreamsException.class, () -> taskExecutor.process(10, time));
+ verify(task).maybeFlushTerminalE2ELatency(anyLong());
+ }
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskExecutorTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskExecutorTest.java
index d43670429b1..284bacb184c 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskExecutorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskExecutorTest.java
@@ -143,6 +143,7 @@ public class DefaultTaskExecutorTest {
verify(task, timeout(VERIFICATION_TIMEOUT).atLeast(2)).process(anyLong());
verify(task, timeout(VERIFICATION_TIMEOUT).atLeastOnce()).recordProcessBatchTime(anyLong());
+ verify(task, timeout(VERIFICATION_TIMEOUT).atLeastOnce()).maybeFlushTerminalE2ELatency(anyLong());
}
diff --git a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
index 5ef01bc849b..ace50d66947 100644
--- a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
+++ b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
@@ -588,6 +588,10 @@ public class TopologyTestDriver implements Closeable {
while (task.hasRecordsQueued() && task.isProcessable(mockWallClockTime.milliseconds())) {
// Process the record ...
task.process(mockWallClockTime.milliseconds());
+ // flush terminal e2e latency for the record just processed, mirroring the end-of-batch
+ // flush in TaskExecutor; the mock clock only advances via advanceWallClockTime, so this
+ // second read returns the same value the record was processed with
+ task.maybeFlushTerminalE2ELatency(mockWallClockTime.milliseconds());
task.maybePunctuateStreamTime();
commit(task.prepareCommit(true));
task.postCommit(true);