Skip to content

[WIP] KAFKA-20412: Propagate headers into serde: CachingKeyValueStore #21971

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
UladzislauBlok wants to merge 1 commit into
base: trunk
Choose a base branch
from
+1 −1
Open

[WIP] KAFKA-20412: Propagate headers into serde: CachingKeyValueStore #21971

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -426,7 +426,7 @@ public KeyValueIterator<Bytes, byte[]> all() {
public <PS extends Serializer<P>, P> KeyValueIterator<Bytes, byte[]> prefixScan(final P prefix, final PS prefixKeySerializer) {
validateStoreOpen();
final KeyValueIterator<Bytes, byte[]> storeIterator = wrapped().prefixScan(prefix, prefixKeySerializer);
final Bytes from = Bytes.wrap(prefixKeySerializer.serialize(null, prefix));
final Bytes from = Bytes.wrap(prefixKeySerializer.serialize(null, internalContext.headers(), prefix));
Copy link
Copy Markdown
Member

@mjsax mjsax Apr 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was thinking about this a little bit, and this change sounds generally ok to me.

But I would assume, that there is also other code (ie, other classes implementing prefixScan method that might need some updates, too?

Btw: we also just added CachingKeyValueStoreWithHeaders for which we might want to do this differently, ie, similar to what we did for put(key, null) -- ie, the change would not go into CachingKeyValueStoreWithHeaders but rather MeteredTimestampKeyValueStoreWithHeaders to modify the RecordContext set on the internalContext object. 🤔

final Bytes to = ByteUtils.increment(from);
final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = internalContext.cache().range(cacheName, from, to, false);
return new MergedSortedCacheKeyValueBytesStoreIterator(cacheIterator, storeIterator, true);
Expand Down
Loading