vcrfxia commented on code in PR #13364:
URL: https://github.com/apache/kafka/pull/13364#discussion_r1152358984
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStoreSegmentValueFormatter.java:
##########
@@ -495,6 +501,41 @@ private boolean isLastIndex(final int index) {
return unpackedReversedTimestampAndValueSizes.get(index).timestamp
== minTimestamp;
}
+ private void truncateRecordsToTimestamp(final long timestamp) {
+ if (timestamp <= minTimestamp) {
+ // delete everything in this current segment by replacing it
with a degenerate segment
+ initializeWithRecord(new ValueAndValueSize(null), timestamp,
timestamp);
+ return;
+ }
+
+ final SegmentSearchResult searchResult = find(timestamp, false);
+ // all records with later timestamps should be removed
+ int fullRecordsToTruncate = searchResult.index();
+ // additionally remove the current record as well, if its
validFrom equals the
+ // timestamp to truncate to
+ if (searchResult.validFrom() == timestamp) {
+ fullRecordsToTruncate++;
+ }
+
+ if (fullRecordsToTruncate == 0) {
+ // no records to remove; update nextTimestamp and return
Review Comment:
The logic added in this PR does handle this case. (It's not "expected" that
this case will come up, but it is possible in weird edge cases such as the
examples that you gave above.) There are unit tests for it as well.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
