You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "sebastienviale (via GitHub)" <gi...@apache.org> on 2024/08/30 08:37:40 UTC
[PR] KAFKA-16448 Add timestamp to error handler context [kafka]
sebastienviale opened a new pull request, #17054:
URL: https://github.com/apache/kafka/pull/17054
This PR is part of [KAFKA-16448](https://issues.apache.org/jira/browse/KAFKA-16448) which aims to bring a ProcessingExceptionHandler to Kafka Streams in order to deal with exceptions that occur during processing.
This PR add the timestamp to the ErrorHandlerContext interface
Jira: https://issues.apache.org/jira/browse/KAFKA-16448.
Contributors
@Dabz
@sebastienviale
@loicgreffier
--
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-16448 Add timestamp to error handler context [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17054:
URL: https://github.com/apache/kafka/pull/17054#discussion_r1741194282
##########
streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContext.java:
##########
@@ -102,4 +106,23 @@ public interface ErrorHandlerContext {
* @return the task ID
*/
TaskId taskId();
+
+ /**
+ * Return the current timestamp.
+ *
+ * <p> If it is triggered while processing a record streamed from the source processor,
+ * timestamp is defined as the timestamp of the current input record; the timestamp is extracted from
+ * {@link org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord} by {@link TimestampExtractor}.
+ * Note, that an upstream {@link org.apache.kafka.streams.processor.api.Processor} might have set a new timestamp by calling
+ * {@link ProcessorContext#forward(Object, Object, To) forward(..., To.all().withTimestamp(...))}.
+ * In particular, some Kafka Streams DSL operators set result record timestamps explicitly,
+ * to guarantee deterministic results.
+ *
+ * <p> If it is triggered while processing a record generated not from the source processor (for example,
+ * if this method is invoked from the punctuate call), timestamp is defined as the current
+ * task's stream time, which is defined as the largest timestamp of any record processed by the task.
Review Comment:
Did you copy this from `RecordContext#timestamp` JavaDocs? They might also need an update covering the wall-clock time case?
Overall, it seems there is no difference between old and new PAPI though?
--
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-16448 Add timestamp to error handler context [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #17054:
URL: https://github.com/apache/kafka/pull/17054#discussion_r1741646378
##########
streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContext.java:
##########
@@ -102,4 +106,23 @@ public interface ErrorHandlerContext {
* @return the task ID
*/
TaskId taskId();
+
+ /**
+ * Return the current timestamp.
+ *
+ * <p> If it is triggered while processing a record streamed from the source processor,
+ * timestamp is defined as the timestamp of the current input record; the timestamp is extracted from
+ * {@link org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord} by {@link TimestampExtractor}.
+ * Note, that an upstream {@link org.apache.kafka.streams.processor.api.Processor} might have set a new timestamp by calling
+ * {@link ProcessorContext#forward(Object, Object, To) forward(..., To.all().withTimestamp(...))}.
+ * In particular, some Kafka Streams DSL operators set result record timestamps explicitly,
+ * to guarantee deterministic results.
+ *
+ * <p> If it is triggered while processing a record generated not from the source processor (for example,
+ * if this method is invoked from the punctuate call), timestamp is defined as the current
+ * task's stream time, which is defined as the largest timestamp of any record processed by the task.
Review Comment:
> I did some more digging... Seems, we actually need to distinguish between `stream time` vs `wall-clock time` punctuations.
>
> For `stream time` punctuation, we set "stream time" as "record context timestamp" and thus `context.timestamp()` will return "stream time". For `wall-clock time` punctuations, we set "system time" as "record context timestamp" and thus `context.timestamp()` will return "system time".
>
> Did I get this correctly?
On what I saw, you are right, I have updated the java doc by splitting different cases of puntuation type
--
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-16448 Add timestamp to error handler context [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #17054:
URL: https://github.com/apache/kafka/pull/17054#issuecomment-2327596627
Build failed:
```
Task :streams:checkstyleTest
--
| 3985 | 01:49:01 AM | [ant:checkstyle] [ERROR] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-17054/streams/src/test/java/org/apache/kafka/streams/integration/ProcessingExceptionHandlerIntegrationTest.java:64:19: 'static' modifier out of order with the JLS suggestions. [ModifierOrder]
```
--
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-16448 Add timestamp to error handler context [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17054:
URL: https://github.com/apache/kafka/pull/17054#discussion_r1742802585
##########
streams/src/test/java/org/apache/kafka/streams/integration/ProcessingExceptionHandlerIntegrationTest.java:
##########
@@ -61,6 +61,8 @@
public class ProcessingExceptionHandlerIntegrationTest {
private final String threadId = Thread.currentThread().getName();
+ private final static Instant TIMESTAMP = Instant.now();
Review Comment:
```suggestion
private static final Instant TIMESTAMP = Instant.now();
```
--
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-16448 Add timestamp to error handler context [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #17054:
URL: https://github.com/apache/kafka/pull/17054#issuecomment-2332164390
Thanks for the follow up @sebastienviale!
Merged to `trunk` and cherry-picked to `3.9` branch. \cc @cmccabe
--
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-16448 Add timestamp to error handler context [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17054:
URL: https://github.com/apache/kafka/pull/17054#discussion_r1741158880
##########
streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContext.java:
##########
@@ -102,4 +106,23 @@ public interface ErrorHandlerContext {
* @return the task ID
*/
TaskId taskId();
+
+ /**
+ * Return the current timestamp.
+ *
+ * <p> If it is triggered while processing a record streamed from the source processor,
+ * timestamp is defined as the timestamp of the current input record; the timestamp is extracted from
+ * {@link org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord} by {@link TimestampExtractor}.
+ * Note, that an upstream {@link org.apache.kafka.streams.processor.api.Processor} might have set a new timestamp by calling
+ * {@link ProcessorContext#forward(Object, Object, To) forward(..., To.all().withTimestamp(...))}.
+ * In particular, some Kafka Streams DSL operators set result record timestamps explicitly,
+ * to guarantee deterministic results.
+ *
+ * <p> If it is triggered while processing a record generated not from the source processor (for example,
+ * if this method is invoked from the punctuate call), timestamp is defined as the current
+ * task's stream time, which is defined as the largest timestamp of any record processed by the task.
Review Comment:
(Did you verify the behavior? I would hope that TTD doe the right/same thing as the actual runtime for this case.)
Also, I don't see an update?
Last, given that `ErrorHandlerContext` in also used for the deserialization case, I am wondering if we should say something about this, too? For this case, the `TimestampExtractor` was not called yet, and thus we pass in `rawRecord.timestamp()`.
--
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-16448 Add timestamp to error handler context [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #17054:
URL: https://github.com/apache/kafka/pull/17054#discussion_r1741644801
##########
streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContext.java:
##########
@@ -102,4 +106,23 @@ public interface ErrorHandlerContext {
* @return the task ID
*/
TaskId taskId();
+
+ /**
+ * Return the current timestamp.
+ *
+ * <p> If it is triggered while processing a record streamed from the source processor,
+ * timestamp is defined as the timestamp of the current input record; the timestamp is extracted from
+ * {@link org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord} by {@link TimestampExtractor}.
+ * Note, that an upstream {@link org.apache.kafka.streams.processor.api.Processor} might have set a new timestamp by calling
+ * {@link ProcessorContext#forward(Object, Object, To) forward(..., To.all().withTimestamp(...))}.
+ * In particular, some Kafka Streams DSL operators set result record timestamps explicitly,
+ * to guarantee deterministic results.
+ *
+ * <p> If it is triggered while processing a record generated not from the source processor (for example,
+ * if this method is invoked from the punctuate call), timestamp is defined as the current
+ * task's stream time, which is defined as the largest timestamp of any record processed by the task.
Review Comment:
> (Did you verify the behavior? I would hope that TTD doe the right/same thing as the actual runtime for this case.)
>
> Last, given that `ErrorHandlerContext` in also used for the deserialization case, I am wondering if we should say something about this, too? For this case, the `TimestampExtractor` was not called yet, and thus we pass in `rawRecord.timestamp()`.
I added a comment when it comes from a deserialization failure
--
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-16448 Add timestamp to error handler context [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17054:
URL: https://github.com/apache/kafka/pull/17054#discussion_r1739583131
##########
streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContext.java:
##########
@@ -102,4 +106,23 @@ public interface ErrorHandlerContext {
* @return the task ID
*/
TaskId taskId();
+
+ /**
+ * Return the current timestamp.
+ *
+ * <p> If it is triggered while processing a record streamed from the source processor,
+ * timestamp is defined as the timestamp of the current input record; the timestamp is extracted from
+ * {@link org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord} by {@link TimestampExtractor}.
+ * Note, that an upstream {@link org.apache.kafka.streams.processor.api.Processor} might have set a new timestamp by calling
+ * {@link ProcessorContext#forward(Object, Object, To) forward(..., To.all().withTimestamp(...))}.
+ * In particular, some Kafka Streams DSL operators set result record timestamps explicitly,
+ * to guarantee deterministic results.
+ *
+ * <p> If it is triggered while processing a record generated not from the source processor (for example,
+ * if this method is invoked from the punctuate call), timestamp is defined as the current
+ * task's stream time, which is defined as the largest timestamp of any record processed by the task.
Review Comment:
Is this correct for all cases? It's for sure correct for the old PAPI using `processor.ProcessorContext`. But I am not sure about the new PAPI which uses `processor.api.ProcessorContext` for which `forward(Record)` take a `Record` which will have a timestamp?
Can we verify this?
--
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-16448 Add timestamp to error handler context [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17054:
URL: https://github.com/apache/kafka/pull/17054#discussion_r1741158880
##########
streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContext.java:
##########
@@ -102,4 +106,23 @@ public interface ErrorHandlerContext {
* @return the task ID
*/
TaskId taskId();
+
+ /**
+ * Return the current timestamp.
+ *
+ * <p> If it is triggered while processing a record streamed from the source processor,
+ * timestamp is defined as the timestamp of the current input record; the timestamp is extracted from
+ * {@link org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord} by {@link TimestampExtractor}.
+ * Note, that an upstream {@link org.apache.kafka.streams.processor.api.Processor} might have set a new timestamp by calling
+ * {@link ProcessorContext#forward(Object, Object, To) forward(..., To.all().withTimestamp(...))}.
+ * In particular, some Kafka Streams DSL operators set result record timestamps explicitly,
+ * to guarantee deterministic results.
+ *
+ * <p> If it is triggered while processing a record generated not from the source processor (for example,
+ * if this method is invoked from the punctuate call), timestamp is defined as the current
+ * task's stream time, which is defined as the largest timestamp of any record processed by the task.
Review Comment:
(Did you verify the behavior? I would hope that TTD doe the right/same thing as the actual runtime for this case.)
Last, given that `ErrorHandlerContext` in also used for the deserialization case, I am wondering if we should say something about this, too? For this case, the `TimestampExtractor` was not called yet, and thus we pass in `rawRecord.timestamp()`.
--
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-16448 Add timestamp to error handler context [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17054:
URL: https://github.com/apache/kafka/pull/17054#discussion_r1742802877
##########
streams/src/test/java/org/apache/kafka/streams/integration/ProcessingExceptionHandlerIntegrationTest.java:
##########
@@ -61,6 +61,8 @@
public class ProcessingExceptionHandlerIntegrationTest {
private final String threadId = Thread.currentThread().getName();
+ private final static Instant TIMESTAMP = Instant.now();
Review Comment:
@sebastienviale -- fixing checkstyle error.
--
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-16448 Add timestamp to error handler context [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #17054:
URL: https://github.com/apache/kafka/pull/17054#discussion_r1743130757
##########
streams/src/test/java/org/apache/kafka/streams/integration/ProcessingExceptionHandlerIntegrationTest.java:
##########
@@ -61,6 +61,8 @@
public class ProcessingExceptionHandlerIntegrationTest {
private final String threadId = Thread.currentThread().getName();
+ private final static Instant TIMESTAMP = Instant.now();
Review Comment:
@mjsax thanks, I forgot to check before pushing
--
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-16448 Add timestamp to error handler context [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17054:
URL: https://github.com/apache/kafka/pull/17054#discussion_r1741158880
##########
streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContext.java:
##########
@@ -102,4 +106,23 @@ public interface ErrorHandlerContext {
* @return the task ID
*/
TaskId taskId();
+
+ /**
+ * Return the current timestamp.
+ *
+ * <p> If it is triggered while processing a record streamed from the source processor,
+ * timestamp is defined as the timestamp of the current input record; the timestamp is extracted from
+ * {@link org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord} by {@link TimestampExtractor}.
+ * Note, that an upstream {@link org.apache.kafka.streams.processor.api.Processor} might have set a new timestamp by calling
+ * {@link ProcessorContext#forward(Object, Object, To) forward(..., To.all().withTimestamp(...))}.
+ * In particular, some Kafka Streams DSL operators set result record timestamps explicitly,
+ * to guarantee deterministic results.
+ *
+ * <p> If it is triggered while processing a record generated not from the source processor (for example,
+ * if this method is invoked from the punctuate call), timestamp is defined as the current
+ * task's stream time, which is defined as the largest timestamp of any record processed by the task.
Review Comment:
(Did you verify the behavior?)
Also, I don't see an update?
Last, given that `ErrorHandlerContext` in also used for the deserialization case, I am wondering if we should say something about this, too? For this case, the `TimestampExtractor` was not called yet, and thus we pass in `rawRecord.timestamp()`.
--
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-16448 Add timestamp to error handler context [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #17054:
URL: https://github.com/apache/kafka/pull/17054#discussion_r1740661104
##########
streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContext.java:
##########
@@ -102,4 +106,23 @@ public interface ErrorHandlerContext {
* @return the task ID
*/
TaskId taskId();
+
+ /**
+ * Return the current timestamp.
+ *
+ * <p> If it is triggered while processing a record streamed from the source processor,
+ * timestamp is defined as the timestamp of the current input record; the timestamp is extracted from
+ * {@link org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord} by {@link TimestampExtractor}.
+ * Note, that an upstream {@link org.apache.kafka.streams.processor.api.Processor} might have set a new timestamp by calling
+ * {@link ProcessorContext#forward(Object, Object, To) forward(..., To.all().withTimestamp(...))}.
+ * In particular, some Kafka Streams DSL operators set result record timestamps explicitly,
+ * to guarantee deterministic results.
+ *
+ * <p> If it is triggered while processing a record generated not from the source processor (for example,
+ * if this method is invoked from the punctuate call), timestamp is defined as the current
+ * task's stream time, which is defined as the largest timestamp of any record processed by the task.
Review Comment:
I have updated th comment with the new processor.api.ProcessorContext
--
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-16448 Add timestamp to error handler context [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #17054:
URL: https://github.com/apache/kafka/pull/17054
--
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-16448 Add timestamp to error handler context [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #17054:
URL: https://github.com/apache/kafka/pull/17054#discussion_r1741651812
##########
streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContext.java:
##########
@@ -102,4 +106,23 @@ public interface ErrorHandlerContext {
* @return the task ID
*/
TaskId taskId();
+
+ /**
+ * Return the current timestamp.
+ *
+ * <p> If it is triggered while processing a record streamed from the source processor,
+ * timestamp is defined as the timestamp of the current input record; the timestamp is extracted from
+ * {@link org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord} by {@link TimestampExtractor}.
+ * Note, that an upstream {@link org.apache.kafka.streams.processor.api.Processor} might have set a new timestamp by calling
+ * {@link ProcessorContext#forward(Object, Object, To) forward(..., To.all().withTimestamp(...))}.
+ * In particular, some Kafka Streams DSL operators set result record timestamps explicitly,
+ * to guarantee deterministic results.
+ *
+ * <p> If it is triggered while processing a record generated not from the source processor (for example,
+ * if this method is invoked from the punctuate call), timestamp is defined as the current
+ * task's stream time, which is defined as the largest timestamp of any record processed by the task.
Review Comment:
> Did you copy this from `RecordContext#timestamp` JavaDocs? They might also need an update covering the wall-clock time case?
>
> Overall, it seems there is no difference between old and new PAPI though?
Indeed, I have copied the java doc.
If comments on the ErrorHandlerContext are good for you, do I have to change it with the new API and add the wall clock time part ?
--
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-16448 Add timestamp to error handler context [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17054:
URL: https://github.com/apache/kafka/pull/17054#discussion_r1741186976
##########
streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContext.java:
##########
@@ -102,4 +106,23 @@ public interface ErrorHandlerContext {
* @return the task ID
*/
TaskId taskId();
+
+ /**
+ * Return the current timestamp.
+ *
+ * <p> If it is triggered while processing a record streamed from the source processor,
+ * timestamp is defined as the timestamp of the current input record; the timestamp is extracted from
+ * {@link org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord} by {@link TimestampExtractor}.
+ * Note, that an upstream {@link org.apache.kafka.streams.processor.api.Processor} might have set a new timestamp by calling
+ * {@link ProcessorContext#forward(Object, Object, To) forward(..., To.all().withTimestamp(...))}.
+ * In particular, some Kafka Streams DSL operators set result record timestamps explicitly,
+ * to guarantee deterministic results.
+ *
+ * <p> If it is triggered while processing a record generated not from the source processor (for example,
+ * if this method is invoked from the punctuate call), timestamp is defined as the current
+ * task's stream time, which is defined as the largest timestamp of any record processed by the task.
Review Comment:
I did some more digging... Seems, we actually need to distinguish between `stream time` vs `wall-clock time` punctuations.
For `stream time` punctuation, we set "stream time" as "record context timestamp" and thus `context.timestamp()` will return "stream time". For `wall-clock time` punctuations, we set "system time" as "record context timestamp" and thus `context.timestamp()` will return "system time".
Did I get this correctly?
--
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