Status

Current state: "Under Discussion"

Discussion thread: here

JIRA: here

Released: ...

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Currently punctuate is triggered by the advance of the task's timestamp, which is the minimum of the timestamps of all input partitions. By default this means the event-time from the messages but a custom TimestampExtractor can be provided to use system-time instead of event-time. However, in that case the triggering of punctuate is still driven by the arrival of messages to all partitions and not by the advance of the system-time itself. The effect is that if any one of the input partitions has messages arriving irregularly, punctuate will be also be called at irregular intervals and in the extreme case not called at all if any one of the input partitions doesn't receive any messages.

Public Interfaces

See below

Terminology

Proposed Changes

The proposal is to deprecate the current punctuate() method:

@Deprecated
void punctuate(long timestamp); // current

Add a new Punctuator functional interface:

interface Punctuator {
    void punctuate(long timestamp);
}

On ProcessorContext deprecate the current schedule method and add a new overload taking the Punctuator added:

@Deprecated
void schedule(long interval); //current, stream-time semantics

void schedule(long interval, PunctuationType type, Punctuator callback); //new
// We could allow this to be called once for each value of PunctuationType to mix approaches.

Where PunctuationType is

enum PunctuationType {
  STREAM_TIME,
  SYSTEM_TIME,
}

Other alternatives under discussion:

(A) Change the semantics of punctuate() to be purely "system time driven", instead of "part time driven, and part data-driven". That is, the punctuate function triggering will no longer be dependent whether there are new data arriving, and hence not on the timestamps of the arriving data either. Instead it will be triggered only by system wall-clock time.

As for users, the programming pattern would be:

1. If you need to add a pure time-driven computation logic, use punctuate().

2. If you need to add a data-driven computation logic, you should always use process(), and in process() users can choose to trigger some specific logic only every some time units, but still when a new data has arrived and hence being processed. With this a punctuation with semantics close to current ones can be achieved but giving user control over the details, as follows:

       long lastPunctuationTime = 0;
   
       long interval = <some-number>; //millis
   
       @Override
       public void process(K key, V value){
   
           while (ctx.timestamp() >= lastPunctuationTime + interval){
   
               punctuate(ctx.timestamp()); //trigger punctuate or any other method at current record timestamp or lastPunctuationTime + interval, if the user prefers
   
               lastPunctuationTime += interval; // or do lastPunctuationTime = ctx.timestamp() if the user prefers
           }
   
           // do some other business logic here
   
       }

Drawbacks:

• The above approach changes the semantics of the punctuate method and therefore is not backward-compatible.
• It is not clear if doing data-driven periodic operations from the process() method without the intricate calculations of minimum timestamp per input partition is sufficient to cater for all use cases that may be attainable using present day stream-time based punctuate 

(B) An alternative could be to leave current semantics as the defaults for the punctuate method but allow a configuration switch between event and system time.

Drawback:

• It's reasonable to assume different semantics be needed in different parts of a topology and configuration is global to a KafkStreams instance, therefore this seems to be too limiting.

(C) Another alternative would be to leave current semantics as-is and add another callback method to the Processor interface that can be scheduled similarly to punctuate() but would always be called at fixed, wall clock based intervals

Drawback:

• This is similar to what's being proposed, however, the functional interface approach offers more flexibility in that the same lambda/method reference can be passed as a parameter to ProcessorContext.schedule() as a callback for both PunctuationTypes.

(D) Yet another alternative would be to leave current semantics as-is but allow users to provide a function determining the timestamp of the stream task. In a similar way to how the TimestampExtractor allows users to decide what the current timestamp is for a given message (event-time, system-time or other), this feature would allow users to decide what the current timestamp is for a given stream task irrespective of the arrival of messages to all of the input partitions. This approach brings more flexibility at the expense of added complexity.

Drawback:

• The scope of this KIP is to re-define punctuate semantics only, without alterations to the notions of stream-time itself, which the alternative would require.

(E) Finally, the hybrid approach (this is convenient for the use cases in Punctuate Use Cases but difficult to reason about):

/**
* Schedule punctuate at stream-time intervals with a system-time upper bound. 
* For pure system-time based punctuation streamTimeInterval can be set to -1 == infinite
* and systemTimeUpperBound to the desired interval
*/
schedule(Punctuator callback, long streamTimeInterval, long systemTimeUpperBound); 

/**
* Schedule punctuate at stream-time intervals without a system-time upper bound,
* i.e. pure stream-time based punctuation
*/
schedule(Punctuator callback, long streamTimeInterval);

Punctuation is triggered when either:
- the stream time advances past the (stream time of the previous punctuation) + streamTimeInterval;
- or (iff systemTimeUpperBound is set) when the system time advances past the (system time of the previous punctuation) + systemTimeUpperBound

In either case:
- we trigger punctuate passing as the argument the stream time at which the current punctuation was meant to happen
- next punctuate is scheduled at (stream time at which the current punctuation was meant to happen) + streamTimeInterval

Compatibility, Deprecation, and Migration Plan

• What impact (if any) will there be on existing users?
• If we are changing behavior how will we phase out the older behavior?
• If we need special migration tools, describe them here.
• When will we remove the existing behavior?

Test Plan

Describe in few sentences how the KIP will be tested. We are mostly interested in system tests (since unit-tests are specific to implementation details). How will we know that the implementation works as expected? How will we know nothing broke?

Rejected Alternatives

If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.

schedule(Punctuator callback, long streamTimeInterval, long systemTimeUpperBound); // schedules punctuate at stream-time intervals with a system-time upper bound - systemTimeUpperBound must be no less than streamTimeInterval

schedule(Punctuator callback, long streamTimeInterval); // schedules punctuate at stream-time intervals without a system-time upper bound - this is equivalent to current stream-time based punctuate

Punctuation is triggered when either:
- the stream time advances past the (stream time of the previous punctuation) + streamTimeInterval;
- or (iff systemTimeUpperBound is set) when the system time advances past the (system time of the previous punctuation) + systemTimeUpperBound