Status

Current state: Draft

Discussion thread: (to be added after posting to dev@kafka.apache.org)

JIRA:

Motivation

Kafka Streams provides periodic callbacks via:

...

As discussed in KAFKA-13678, this KIP proposes extending the existing schedule() API by adding a time-shift parameter to allow explicit control over punctuation alignment.

Public Interfaces

This KIP introduces a new overloaded method in:

...

Cancellable schedule(Duration interval,
                     PunctuationType type,
                     Duration timeShift,
                     Punctuator callback);

Parameter Semantics

• interval — existing interval duration

• type — existing punctuation type

• timeShift — alignment shift relative to epoch time

• callback — existing punctuator

timeShift Behavior

• timeShift == null
  → preserves existing behavior (relative to first event or scheduling time)

• timeShift == Duration.ZERO
  → align punctuations to epoch-aligned boundaries

• timeShift == X
  → align to epoch boundaries shifted by X

...

No protocol, configuration, or binary format changes are introduced.

Proposed Changes

1. Alignment Calculation

When timeShift is provided, the next punctuation timestamp is computed as:

...

If timeShift is null, existing scheduling logic remains unchanged.

2. STREAM_TIME Semantics

If:

interval = 10
timeShift = 0

...

Independent of first processed record.

3. WALL_CLOCK_TIME Semantics

If:

interval = 10
timeShift = 5

...

5, 15, 25, 35...

Relative to epoch time.

Streams Internal Implementation

This section describes how the change integrates with the Kafka Streams internals.

Current Scheduling Flow

Image RemovedImage Added

Relevant Classes

Located under:

streams/src/main/java/org/apache/kafka/streams/processor/internals/

...

• ProcessorContextImpl

• StreamTask

• PunctuationQueue

• PunctuationSchedule

ProcessorContextImpl

Add overloaded schedule() method accepting timeShift.

...

The existing schedule(Duration, PunctuationType, Punctuator) remains unchanged.

StreamTask

Extend internal scheduling logic to accept optional timeShift.

...

• During initial scheduling

• During rescheduling after punctuation firing

PunctuationSchedule

Add new field:

private final Long timeShiftMillis;

...

No changes to queue ordering logic are required.

PunctuationQueue

No structural modifications.

Queue continues to order schedules by nextTimestamp.

...

Architecture Diagram

Current

Image RemovedImage Added

...

With timeShift

Image RemovedImage Added

Compatibility, Deprecation, and Migration Plan

• Existing schedule(Duration, PunctuationType, Punctuator) remains unchanged.

• Default behavior is preserved.

• New behavior is opt-in.

• No migration required.

• No deprecation of existing APIs.

Test Plan

Unit Tests

Add tests covering:

• STREAM_TIME with null shift (existing behavior)

• STREAM_TIME with zero shift

• STREAM_TIME with custom shift

• WALL_CLOCK_TIME with zero shift

• Multiple punctuations with different shifts in same task

• Restart simulation ensuring deterministic alignment

Integration Tests

Add Streams integration test:

1. Start topology.

2. Produce records with varying first timestamps.

3. Restart application.

4. Verify punctuation alignment remains consistent when shift is specified.

Regression Validation

All existing punctuation tests must pass without modification.

Rejected Alternatives

1. Alignment Strategy Enum

Rejected to keep the change minimal and close to the original discussion in KAFKA-13678.

A numeric shift parameter provides greater flexibility with less API surface.

2. Separate Timer API

Rejected because periodic punctuations already provide the required abstraction.

3. Changing Default Behavior

Rejected due to backward compatibility concerns.

Conclusion

This KIP extends Kafka Streams punctuation scheduling with an optional time-shift parameter.

It enables deterministic and configurable alignment of periodic callbacks while preserving existing semantics and maintaining full backward compatibility.