You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "AlexCSalinas (via GitHub)" <gi...@apache.org> on 2026/02/05 18:27:15 UTC
[PR] KAFKA-20099: Clarify how stream-time works in the docs [kafka]
AlexCSalinas opened a new pull request, #21414:
URL: https://github.com/apache/kafka/pull/21414
## Summary
Clarifies how stream time works in the Kafka Streams docs:
- Stream time is implemented **per task** ("task time") based on input records for that task; there is no global stream time.
- Task time is preserved across rebalances and application restarts.
- Some DSL operators that use grace periods maintain their own in-memory stream time ("operator time"), which is not preserved across rebalances/restarts and is re-established from the first input record after restart.
- Documents that this distinction affects how grace periods are applied.
## References
- KAFKA-20099
- KAFKA-9368 (related)
## License
This contribution is my original work and I license it under the project's open source license.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20099: Clarify how stream-time works in the docs [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21414:
URL: https://github.com/apache/kafka/pull/21414#discussion_r2785860467
##########
docs/streams/core-concepts.md:
##########
@@ -75,6 +75,8 @@ The choice between event-time and ingestion-time is actually done through the co
Kafka Streams assigns a **timestamp** to every data record via the `TimestampExtractor` interface. These per-record timestamps describe the progress of a stream with regards to time and are leveraged by time-dependent operations such as window operations. As a result, this time will only advance when a new record arrives at the processor. We call this data-driven time the **stream time** of the application to differentiate with the **wall-clock time** when this application is actually executing. Concrete implementations of the `TimestampExtractor` interface will then provide different semantics to the stream time definition. For example retrieving or computing timestamps based on the actual contents of data records such as an embedded timestamp field to provide event time semantics, and returning the current wall-clock time thereby yield processing time semantics to stream time. Developers can thus enforce different notions of time depending on their business needs.
+Kafka Streams implement stream time at two levels. At runtime, stream time is maintained per task based on input records for that task (called "task time"). There is no global stream time, each task tracks its own stream time independently. `Task time` is preserved across rebalances and restarts. Some DSL operators that use grace periods maintain their own in-memory stream time ("operator time"), which is **not** preserved across rebalances or restarts and is re-established from the first input record after restart. This distinction affects how grace periods are applied, as they use operator time rather than task time.
Review Comment:
```suggestion
Kafka Streams implement stream-time at two levels. At runtime, stream-time is maintained per task based on input records for that task (called "task-time"). There is no global stream-time and each task tracks its own stream-time independently. Task-time is preserved across rebalances and restarts. Some DSL operators that use grace periods maintain their own in-memory stream-time ("operator time"), which is **not** preserved across rebalances or restarts and is re-established from the first input record after restart. This distinction affects how grace periods are applied, as they use operator time rather than task time.
```
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20099: Clarify how stream-time works in the docs [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21414:
URL: https://github.com/apache/kafka/pull/21414#issuecomment-4427405665
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20099: Clarify how stream-time works in the docs [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21414:
URL: https://github.com/apache/kafka/pull/21414#discussion_r2785851444
##########
docs/streams/developer-guide/processor-api.md:
##########
@@ -53,7 +53,7 @@ The `Processor` interface takes four generic parameters: `KIn, VIn, KOut, VOut`.
Both the `Processor#process()` and the `ProcessorContext#forward()` methods handle records in the form of the `Record<K, V>` data class. This class gives you access to the main components of a Kafka record: the key, value, timestamp and headers. When forwarding records, you can use the constructor to create a new `Record` from scratch, or you can use the convenience builder methods to replace one of the `Record`'s properties and copy over the rest. For example, `inputRecord.withValue(newValue)` would copy the key, timestamp, and headers from `inputRecord` while setting the output record's value to `newValue`. Note that this does not mutate `inputRecord`, but instead creates a shallow copy. Beware that this is only a shallow copy, so if you plan to mutate the key, value, or headers elsewhere in the program, you will want to create a deep copy of those fields yourself.
-In addition to handling incoming records via `Processor#process()`, you have the option to schedule periodic invocation (called "punctuation") in your processor's `init()` method by calling `ProcessorContext#schedule()` and passing it a `Punctuator`. The `PunctuationType` determines what notion of time is used for the punctuation scheduling: either [stream-time](../core-concepts.html#streams_time) or wall-clock-time (by default, stream-time is configured to represent event-time via `TimestampExtractor`). When stream-time is used, `punctuate()` is triggered purely by data because stream-time is determined (and advanced forward) by the timestamps derived from the input data. When there is no new input data arriving, stream-time is not advanced and thus `punctuate()` is not called.
+In addition to handling incoming records via `Processor#process()`, you have the option to schedule periodic invocation (called "punctuation") in your processor's `init()` method by calling `ProcessorContext#schedule()` and passing it a `Punctuator`. The `PunctuationType` determines what notion of time is used for the punctuation scheduling: either [stream-time](../core-concepts.html#streams_time) or wall-clock-time (by default, stream-time is configured to represent event-time via `TimestampExtractor`). When stream-time is used, `punctuate()` is triggered purely by data because stream-time is determined (and advanced forward) by the timestamps derived from the input data. When there is no new input data arriving, stream-time is not advanced and thus `punctuate()` is not called. The stream time used for punctuation is task-level stream time, which is preserved across rebalances and restarts (see [Core Concepts](../core-concepts.html#time) for details on task-level vs operator-level
stream time).
Review Comment:
```suggestion
In addition to handling incoming records via `Processor#process()`, you have the option to schedule periodic invocation (called "punctuation") in your processor's `init()` method by calling `ProcessorContext#schedule()` and passing it a `Punctuator`. The `PunctuationType` determines what notion of time is used for the punctuation scheduling: either [stream-time](../core-concepts.html#streams_time) or wall-clock-time (by default, stream-time is configured to represent event-time via `TimestampExtractor`). When stream-time is used, `punctuate()` is triggered purely by data because stream-time is determined (and advanced forward) by the timestamps derived from the input data. When there is no new input data arriving, stream-time is not advanced and thus `punctuate()` is not called. The stream-time used for punctuations is task-level stream-time, which is preserved across rebalances and restarts (see [Core Concepts](../core-concepts.html#time) for details on task-level vs operator-lev
el stream-time).
```
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20099: Clarify how stream-time works in the docs [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21414:
URL: https://github.com/apache/kafka/pull/21414#discussion_r2785857071
##########
docs/streams/developer-guide/processor-api.md:
##########
@@ -309,7 +309,7 @@ You can create a persistent, versioned state store by passing a [VersionedBytesS
Each versioned store has an associated, fixed-duration _history retention_ parameter which specifies long old record versions should be kept for. In particular, a versioned store guarantees to return accurate results for timestamped retrieval operations where the timestamp being queried is within history retention of the current observed stream time.
-History retention also doubles as its _grace period_ , which determines how far back in time out-of-order writes to the store will be accepted. A versioned store will not accept writes (inserts, updates, or deletions) if the timestamp associated with the write is older than the current observed stream time by more than the grace period. Stream time in this context is tracked per-partition, rather than per-key, which means it's important that grace period (i.e., history retention) be set high enough to accommodate a record with one key arriving out-of-order relative to a record for another key.
+History retention also doubles as its _grace period_ , which determines how far back in time out-of-order writes to the store will be accepted. A versioned store will not accept writes (inserts, updates, or deletions) if the timestamp associated with the write is older than the current observed stream time by more than the grace period. Stream time in this context refers to task-level stream time (tracked per-partition, rather than per-key), which is preserved across restarts, unlike operator-level stream time used by DSL window operators. It's important that grace period be set high enough to accommodate a record with one key arriving out-of-order relative to a record for another key.
Review Comment:
For versioned kv-stores, it's also operator level stream-time, not task-level.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20099: Clarify how stream-time works in the docs [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21414:
URL: https://github.com/apache/kafka/pull/21414#issuecomment-4677346631
This PR has been closed since it has not had any activity in 120 days. If you feel like this
was a mistake, or you would like to continue working on it, please feel free to re-open the
PR and ask for a review.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20099: Clarify how stream-time works in the docs [kafka]
Posted by "AlexCSalinas (via GitHub)" <gi...@apache.org>.
AlexCSalinas commented on PR #21414:
URL: https://github.com/apache/kafka/pull/21414#issuecomment-3868285286
@mjsax Would you check out these doc changes if you get a chance?
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20099: Clarify how stream-time works in the docs [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #21414: KAFKA-20099: Clarify how stream-time works in the docs
URL: https://github.com/apache/kafka/pull/21414
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org