Table of Contents |
---|
This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.
Status
Current state: "Under Discussion"Accepted
Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]
JIRA: KAFKA-8953
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Motivation
Kafka Streams ships couple of different timestamp extractors, one named `UsePreviousTimeOnInvalidTimestamp`.
Given the latest improvements with regard to time tracking, it seems appropriate to rename this class to `UsePartitionTimeOnInvalidTimestamp`, as we now have fixed definition of partition time
...
.
Public Interfaces
Interface TimeStampExtractor won't have any effects UsePreviousTimeOnInvalidTimeStamp will be deprecated if class name is changed from UsePreviousTimeOnInvalidTimeStamp to UsePartitionOnInvalidTimeStamp won't change the functionality to return the highest extracted valid timestamp of the current records partition.
Proposed Changes
Previously, in Kafka 2.3 we had public long onInvalidTimestamp(ConsumerRecord<Object,Object> record, long recordTimestamp, long previousTimestamp) and public long extract(ConsumerRecord<Object,Object> record, long previousTimestamp) in UsePreviousTimeOnInvalidTimestamp
but now extract has been removed from 'UsePreviousTimeOnInvalidTimestamp' and also the argument long previousTimestamp has been changed to long partitionTime so it seems reasonable for us to change the class name to change as main functionality of this class is to return the record partition.
Also, there will be additional minor changes to be made if community agrees on the mentioned proposal which are listed below:
...
.
Code Block | ||
---|---|---|
| ||
public class UsePreviousTimeOnInvalidTimestamp extends ExtractRecordMetadataTimestamp {
...
}
|
Current method signature for onInvalidTimestamp
Code Block | ||
---|---|---|
| ||
public long onInvalidTimestamp(final ConsumerRecord<Object, Object> record, final long recordTimestamp, final long partitionTime)
throws StreamsException {
...
} |
Proposed Changes
Change 'UsePreviousTimeOnInvalidTimeStamp' to 'UsePartitionTimeOnInvalidTimestamp'
Code Block | ||
---|---|---|
| ||
public class UsePreviousTimeOnInvalidTimestamp extends ExtractRecordMetadataTimestamp {
...
}
|
Code Block | ||
---|---|---|
| ||
public class UsePartitionTimeOnInvalidTimestamp extends ExtractRecordMetadataTimestamp {
...
}
|
...
Compatibility, Deprecation, and Migration Plan
UsePreviousTimeOnInvalidTimestamp won't be present instead 'UsePartitionOnInvalidTimeStamp' will be there in next major version without change in the functionality but it's clear from the name that this will be used for partition time but UsePreviousTimeOnInvalidTimeStamp will need to be deprecated first.
Rejected Alternatives
None.