MINOR: Fix ProcessorContext JavaDocs#8603
Conversation
|
Call for review @vvcephei |
|
Java 8 and 14 passed. |
| * <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 smallest among all its input stream partition timestamps. | ||
| * task's stream time, which is defined as the largest among all its input stream partition timestamps. |
There was a problem hiding this comment.
I just took another look at the definition of streamTime, and it actually looks like it might be computed wrongly.
The way it works is that the "stream time" for a task is computed most of the time in org.apache.kafka.streams.processor.internals.PartitionGroup#nextRecord, i.e., it's the max timestamp of any record polled from the PartitionGroup.
However, when we commit, we commit the "partition time" for each TopicPartition, which is set when we move a record into the head position for that queue. During restoration, we read these committed timestamps for each TopicPartition, and we (incorrectly) set the "stream time" to be the maximum over the "partition time" of each partition in the PartitionGroup (aka Task).
This is incorrect in two ways:
- it should be the minimum, not the maximum (since we would choose the record with the minimum timestamp to process next)
- the timestamp of the head enqueued record (partition time) is not the timestamp of the last dequeued record (stream time).
I'll file a Jira ticket capturing all this. In the mean time, I'd suggest that we just update the docs to reflect the correct definition of "stream time": which is defined as the largest timestamp of any record processed by the task. Then, we can fix the code to make this true all the time. Currently, it's only true in steady state, not immediately after restoration.
There was a problem hiding this comment.
Sound like a bug :)
But it seems to be a one line fix that I can piggy-back on this PR. We advance the "partition time" too early. If we advance it when the return the record for processing, all should be fixed? Partition time should not be the head-record timestamp but the past processed record of the partition.
There was a problem hiding this comment.
Ah, perfect! This is much simpler than what I was thinking. Thanks.
|
@vvcephei Updates this. |
|
Java 8: |
|
Retest this please. |
* 'trunk' of github.com:apache/kafka: MINOR: add option to rebuild source for system tests (apache#6656) KAFKA-9850 Move KStream#repartition operator validation during Topolo… (apache#8550) MINOR: Add a duplicate() method to Message classes (apache#8556) KAFKA-9966: add internal assignment listener to stabilize eos-beta upgrade test (apache#8648) MINOR: Replace null with an actual value for timestamp field in InsertField SMT unit tests (apache#8649) MINOR: Fix ProcessorContext JavaDocs and stream-time computation (apache#8603) MINOR: improve tests for TopologyTestDriver (apache#8631) KAFKA-9821: consolidate Streams rebalance triggering mechanisms (apache#8596) KAFKA-9669; Loosen validation of inner offsets for older message formats (apache#8647) KAFKA-8770: KIP-557: Drop idempotent KTable source updates (apache#8254) MINOR: Remove allow concurrent test (apache#8641)
…che#8603) Reviewer: John Roesler <john@confluent.io>
2 participants
We changed how "stream time" is computed in
2.3release. This should be cherry-picked to older branches.