You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "nicktelford (via GitHub)" <gi...@apache.org> on 2026/05/18 16:56:19 UTC
[PR] KAFKA-20497: Add readOnly(IsolationLevel) to CachingWindowStore [kafka]
nicktelford opened a new pull request, #22314:
URL: https://github.com/apache/kafka/pull/22314
The cache holds uncommitted writes that must not be visible under
READ_COMMITTED, so that isolation level bypasses the cache entirely and
delegates straight to the inner store's readOnly view. READ_UNCOMMITTED
requires a merged view of both cache and store through a ReadOnlyView.
The ReadOnlyView converts Instant arguments to epoch-milliseconds before
calling the shared fetchInternal/fetchKeyRangeInternal/fetchAllInternal
helpers rather than converting at the merge boundary; the cache key
schema operates in longs throughout, so converting early avoids repeated
Instant-to-long conversions in iterator hot paths.
KAFKA-20497
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20497: Add readOnly(IsolationLevel) to CachingWindowStore [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck merged PR #22314:
URL: https://github.com/apache/kafka/pull/22314
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20497: Add readOnly(IsolationLevel) to CachingWindowStore [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on code in PR #22314:
URL: https://github.com/apache/kafka/pull/22314#discussion_r3341538761
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java:
##########
@@ -170,242 +174,200 @@ public synchronized void put(final Bytes key,
}
@Override
- public byte[] fetch(final Bytes key,
- final long timestamp) {
+ public byte[] fetch(final Bytes key, final long timestamp) {
+ return fetchPointInternal(key, timestamp, wrapped());
+ }
+
+ private byte[] fetchPointInternal(final Bytes key,
+ final long timestamp,
+ final ReadOnlyWindowStore<Bytes, byte[]> source) {
validateStoreOpen();
final Bytes bytesKey = WindowKeySchema.toStoreKeyBinary(key, timestamp, 0);
final Bytes cacheKey = cacheFunction.cacheKey(bytesKey);
if (internalContext.cache() == null) {
- return wrapped().fetch(key, timestamp);
+ return source.fetch(key, timestamp);
}
final LRUCacheEntry entry = internalContext.cache().get(cacheName, cacheKey);
if (entry == null) {
- return wrapped().fetch(key, timestamp);
+ return source.fetch(key, timestamp);
} else {
return entry.value();
}
}
+
@Override
- public synchronized WindowStoreIterator<byte[]> fetch(final Bytes key,
Review Comment:
Good catch! It's been reinstated now.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20497: Add readOnly(IsolationLevel) to CachingWindowStore [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on PR #22314:
URL: https://github.com/apache/kafka/pull/22314#issuecomment-4479919964
@bbejeck
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20497: Add readOnly(IsolationLevel) to CachingWindowStore [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on PR #22314:
URL: https://github.com/apache/kafka/pull/22314#issuecomment-4602880927
> Thanks for the PR @nicktelford overall looks good. I have a question about the moving the `synchronized` method modfier.
>
> Also the `org.apache.kafka.streams.state.internals.CachingPersistentWindowStoreTest#shouldNotThrowInvalidBackwardRangeExceptionWithNegativeFromKey` is failing
@bbejeck Fixed the synchronisation, as requested, and rebased on trunk, which should fix the tests.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20497: Add readOnly(IsolationLevel) to CachingWindowStore [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #22314:
URL: https://github.com/apache/kafka/pull/22314#issuecomment-4607380132
merged #22314 into trunk
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20497: Add readOnly(IsolationLevel) to CachingWindowStore [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22314:
URL: https://github.com/apache/kafka/pull/22314#issuecomment-4540047292
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20497: Add readOnly(IsolationLevel) to CachingWindowStore [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #22314:
URL: https://github.com/apache/kafka/pull/22314#discussion_r3337351219
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java:
##########
@@ -170,242 +174,200 @@ public synchronized void put(final Bytes key,
}
@Override
- public byte[] fetch(final Bytes key,
- final long timestamp) {
+ public byte[] fetch(final Bytes key, final long timestamp) {
+ return fetchPointInternal(key, timestamp, wrapped());
+ }
+
+ private byte[] fetchPointInternal(final Bytes key,
+ final long timestamp,
+ final ReadOnlyWindowStore<Bytes, byte[]> source) {
validateStoreOpen();
final Bytes bytesKey = WindowKeySchema.toStoreKeyBinary(key, timestamp, 0);
final Bytes cacheKey = cacheFunction.cacheKey(bytesKey);
if (internalContext.cache() == null) {
- return wrapped().fetch(key, timestamp);
+ return source.fetch(key, timestamp);
}
final LRUCacheEntry entry = internalContext.cache().get(cacheName, cacheKey);
if (entry == null) {
- return wrapped().fetch(key, timestamp);
+ return source.fetch(key, timestamp);
} else {
return entry.value();
}
}
+
@Override
- public synchronized WindowStoreIterator<byte[]> fetch(final Bytes key,
Review Comment:
I'm thinking we need to move the `synchronized` modifer back to the external fetch call as it stands now the `validateStoreOpen` and the `fetch` for the underlying iterator are not thread safe - the `wrapped().fetch` call will be evaluated before entering the method body - hence not synchronized.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org