[
https://issues.apache.org/jira/browse/KAFKA-10062?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17136432#comment-17136432
]
Piotr Smolinski commented on KAFKA-10062:
-----------------------------------------
The motivation is to use the same source of the wall-clock time in execution
environment and in the tests.
Imagine you have a case when you have to execute an action when message arrives
based on the current time and then potentially some time later. Right now when
the message arrives you have access to the message time via timestamp() method
from the ProcessorContext:
[https://kafka.apache.org/25/javadoc/index.html?org/apache/kafka/streams/processor/ProcessorContext.html]
Next you can schedule a punctuator and tell which time should be used,
wall-clock or stream. The problem is that these three times have different
semantics. timestamp() just tells you what is message timestamp. Stream time is
whatever the system thinks is the time based on timestamp(), but in theory it
should only advance. For example if your timestamps in the topic are mixed (set
as producer time), the time may jump forth and back. Wall-clock time is also
ever-advancing time; well, NTP may put it back due to drift, but we should
expect that this time also only advances.
Now, if you want to implement a solution that uses wall-clock and does actions
based on it, currently you may follow the short path and just read
System.currentTimeMillis(). The problem is that in tests you have no control on
it. In punctuate() call the time comes from the time implementation.
[https://github.com/apache/kafka/blob/2.5.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L75]
ProcessorContextImpl has already the reference path to this object, but it is
private:
[https://github.com/apache/kafka/blob/2.5.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java#L50]
Therefore if you have a reference to ProcessorContext, which is unwrapped
ProcessorContextImpl, you have the necessary path:
context->ProcessorContextImpl#task->StreamTask#time
The problem is that these fields are private.
That's about system time.
StreamTask has two punctuation queues. One for system and one for stream time.
The stream time comes from PartitionGroup:
[https://github.com/apache/kafka/blob/2.5.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L841]
The time is advanced before processing a new record that has time higher than
the currently known streamTime:
[https://github.com/apache/kafka/blob/2.5.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java#L136]
Therefore the time available as ProcessorContext#timestamp may be earlier than
the streamTime if the records are not processed in the time order.
To get the streamTime follow the path:
context->ProcessorContextImpl#task->StreamTask#streamTime()
The StreamTask#streamTime() is package private.
My expectation is that we have methods to access these times as public API in
ProcessorContext and implementation in ProcessorContextImpl:
public long currentSystemTimeMs() {
return task.systemTime();
}
public long currentStreamTimeMs() {
return task.streamTime();
}
In GlobalProcessorContext the methods may stay unsupported, esp. because
GlobalProcessorContext does not support schedule.
> Add a method to retrieve the current timestamp as known by the Streams app
> --------------------------------------------------------------------------
>
> Key: KAFKA-10062
> URL: https://issues.apache.org/jira/browse/KAFKA-10062
> Project: Kafka
> Issue Type: Improvement
> Components: streams
> Reporter: Piotr Smolinski
> Assignee: William Bottrell
> Priority: Major
> Labels: needs-kip, newbie
>
> Please add to the ProcessorContext a method to retrieve current timestamp
> compatible with Punctuator#punctate(long) method.
> Proposal in ProcessorContext:
> long getTimestamp(PunctuationType type);
> The method should return time value as known by the Punctuator scheduler with
> the respective PunctuationType.
> The use-case is tracking of a process with timeout-based escalation.
> A transformer receives process events and in case of missing an event execute
> an action (emit message) after given escalation timeout (several stages). The
> initial message may already arrive with reference timestamp in the past and
> may trigger different action upon arrival depending on how far in the past it
> is.
> If the timeout should be computed against some further time only, Punctuator
> is perfectly sufficient. The problem is that I have to evaluate the current
> time-related state once the message arrives.
> I am using wall-clock time. Normally accessing System.currentTimeMillis() is
> sufficient, but it breaks in unit testing with TopologyTestDriver, where the
> app wall clock time is different from the system-wide one.
> To access the mentioned clock I am using reflection to access
> ProcessorContextImpl#task and then StreamTask#time.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
