rohitrmd commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r551055293
##########
File path:
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
##########
@@ -494,7 +494,9 @@ public void process(final Record<Integer, Integer> record) {
// e2e latency = 10
task.addRecords(partition1, singletonList(getConsumerRecord(0, 0L)));
- task.process(10L);
+ time = new MockTime(0L, 10L, 0L);
Review comment:
@mjsax The reason I had to change StreamTaskTest is because of following
issue:
StreamTask -> process(wallClockTime) method updates processor context's
system time in this method:
https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L685
AbstractProcessorContect has another systemTime field:
https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java#L48
which is updated in streamTask's process method. I removed this field from
AbstractProcessorContect as we want to fetch the time from StreamTask. But the
latency is measured
[here](https://github.com/apache/kafka/blob/b9dfc196aaee91f4fa23fe6563261c6f2d01ac33/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java#L284)
which is streamTask's time filed now which is not updated.
I couldn't find any other way to change time field in StreamTask unless I
add setTime method in it. If i don't change time in StreamTask object,
subsequent metric tests fail as stream-time is not updated.
Do you have recommendation about how i can fix it without adding setTime
method?
----------------------------------------------------------------
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:
[email protected]
