aliehsaeedii commented on code in PR #14626:
URL: https://github.com/apache/kafka/pull/14626#discussion_r1384178950
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java:
##########
@@ -253,6 +259,86 @@ public VersionedRecord<byte[]> get(final Bytes key, final
long asOfTimestamp) {
return null;
}
+ // Visible for testing
+ ValueIterator<VersionedRecord<byte[]>> get(final Bytes key, final long
fromTimestamp, final long toTimestamp, final boolean isAscending) {
+
+ Objects.requireNonNull(key, "key cannot be null");
+ validateStoreOpen();
+
+ final List<VersionedRecord<byte[]>> queryResults = new ArrayList<>();
+
+ if (toTimestamp < observedStreamTime - historyRetention) {
+ // history retention exceeded. we still check the latest value
store in case the
+ // latest record version satisfies the timestamp bound, in which
case it should
+ // still be returned (i.e., the latest record version per key
never expires).
+ final byte[] rawLatestValueAndTimestamp =
latestValueStore.get(key);
+ if (rawLatestValueAndTimestamp != null) {
+ final long latestTimestamp =
LatestValueFormatter.getTimestamp(rawLatestValueAndTimestamp);
+ if (latestTimestamp <= toTimestamp) {
+ // latest value satisfies timestamp bound
+ queryResults.add(new VersionedRecord<>(
+
LatestValueFormatter.getValue(rawLatestValueAndTimestamp),
+ latestTimestamp
+ ));
+ }
+ }
+
+ // history retention has elapsed and the latest record version (if
present) does
+ // not satisfy the timestamp bound. return null for
predictability, even if data
+ // is still present in segments.
+ if (queryResults.size() == 0) {
+ LOG.warn("Returning null for expired get.");
Review Comment:
> Why do we log a WARNING here? Returning nothing seems to be regular
execution -- we should not "worry" users about this.
I just added the warning because we have it in above `get` method.
--
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
