[
https://issues.apache.org/jira/browse/KAFKA-7994?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16835192#comment-16835192
]
ASF GitHub Bot commented on KAFKA-7994:
---------------------------------------
ConcurrencyPractitioner commented on pull request #6694: [KAFKA-7994] Improve
Stream time accuracy for restarts and rebalances
URL: https://github.com/apache/kafka/pull/6694
The issue for this PR could be found here:
https://issues.apache.org/jira/browse/KAFKA-7994?jql=project%20%3D%20KAFKA%20AND%20status%20in%20(Open%2C%20%22In%20Progress%22%2C%20Reopened%2C%20%22Patch%20Available%22)
As noted in the JIRA description, stream time is incorrectly set to -1 after
rebalances and restarts. To help resolve this issue, a straightforward approach
is to commit the individual partition time with last message processed for each
RecordQueue. Hence, after a restart, we could set the partition time to the
last committed partition time.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Improve Stream-Time for rebalances and restarts
> -----------------------------------------------
>
> Key: KAFKA-7994
> URL: https://issues.apache.org/jira/browse/KAFKA-7994
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Reporter: Matthias J. Sax
> Priority: Major
> Attachments: possible-patch.diff
>
>
> We compute a per-partition partition-time as the maximum timestamp over all
> records processed so far. Furthermore, we use partition-time to compute
> stream-time for each task as maximum over all partition-times (for all
> corresponding task partitions). This stream-time is used to make decisions
> about processing out-of-order records or drop them if they are late (ie,
> timestamp < stream-time - grace-period).
> During rebalances and restarts, stream-time is initialized as UNKNOWN (ie,
> -1) for tasks that are newly created (or migrated). In net effect, we forget
> current stream-time for this case what may lead to non-deterministic behavior
> if we stop processing right before a late record, that would be dropped if we
> continue processing, but is not dropped after rebalance/restart. Let's look
> at an examples with a grade period of 5ms for a tumbling windowed of 5ms, and
> the following records (timestamps in parenthesis):
>
> {code:java}
> r1(0) r2(5) r3(11) r4(2){code}
> In the example, stream-time advances as 0, 5, 11, 11 and thus record `r4` is
> dropped as late because 2 < 6 = 11 - 5. However, if we stop processing or
> rebalance after processing `r3` but before processing `r4`, we would
> reinitialize stream-time as -1, and thus would process `r4` on restart/after
> rebalance. The problem is, that stream-time does advance differently from a
> global point of view: 0, 5, 11, 2.
>
> Note, this is a corner case, because if we would stop processing one record
> earlier, ie, after processing `r2` but before processing `r3`, stream-time
> would be advance correctly from a global point of view.
> A potential fix would be, to store latest observed partition-time in the
> metadata of committed offsets. Thus way, on restart/rebalance we can
> re-initialize time correctly.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
