Uploaded image for project: 'Kafka'
  1. Kafka
  2. KAFKA-10062

Add a method to retrieve the current timestamp as known by the Streams app

    XMLWordPrintableJSON

Details

    • Improvement
    • Status: Resolved
    • Major
    • Resolution: Fixed
    • None
    • 3.0.0
    • streams

    Description

      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.

       

      KIP-622: https://cwiki.apache.org/confluence/display/KAFKA/KIP-622%3A+Add+currentSystemTimeMs+and+currentStreamTimeMs+to+ProcessorContext 

      Attachments

        Issue Links

          Activity

            People

              rohitdeshaws Rohit Deshpande
              psmolinski Piotr Smolinski
              Votes:
              0 Vote for this issue
              Watchers:
              10 Start watching this issue

              Dates

                Created:
                Updated:
                Resolved: