You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "UladzislauBlok (via GitHub)" <gi...@apache.org> on 2026/04/05 12:01:36 UTC
[PR] KAFKA-20173: Propagate headers into serde: CachingKeyValueStore [kafka]
UladzislauBlok opened a new pull request, #21971:
URL: https://github.com/apache/kafka/pull/21971
as a title
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #21971:
URL: https://github.com/apache/kafka/pull/21971#issuecomment-5027731375
If you get a chance, please rebase this PR to `trunk` to pickup the fix for failing `TopologyDescriptionPluginExpirationIntegrationTest` so we can get a green build. Thanks.
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21971:
URL: https://github.com/apache/kafka/pull/21971#discussion_r3592189930
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java:
##########
@@ -438,6 +438,7 @@ private <PS extends Serializer<P>, P> KeyValueIterator<Bytes, byte[]> prefixScan
final ReadOnlyKeyValueStore<Bytes, byte[]> underlying) {
validateStoreOpen();
final KeyValueIterator<Bytes, byte[]> storeIterator = underlying.prefixScan(prefix, prefixKeySerializer);
+ // headers aren't needed because the prefix already arrives serialized
Review Comment:
But this is only correct _if_ the most outer store is `MeteredTimestampedKeyValueStoreWithHeaders` for which this PR changes the `prefixScan` impl. For the other metered-kv-stores this is not correct.
And I am wondering why we want to put the serialization into the metered layer now? Guess not wrong, because that's is what the metered layer does in general, but for this case, we should also update `MeteredKeyValueStore.java` accordingly? (`MeteredTimestampedKeyValueStore` does not overwrite `prefixScan()`)?
If we add this comment, it seems we would need to add it elsewhere, too, eg `RocksDBStore` and maybe also others -- ie, everywhere when `prefixKeySerializer.serialize(...)` is used?
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21971:
URL: https://github.com/apache/kafka/pull/21971#discussion_r3598508720
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java:
##########
@@ -438,6 +438,7 @@ private <PS extends Serializer<P>, P> KeyValueIterator<Bytes, byte[]> prefixScan
final ReadOnlyKeyValueStore<Bytes, byte[]> underlying) {
validateStoreOpen();
final KeyValueIterator<Bytes, byte[]> storeIterator = underlying.prefixScan(prefix, prefixKeySerializer);
+ // headers aren't needed because the prefix already arrives serialized
Review Comment:
> f you're asking from perspective of metered vs underlying: this is because underlying doesn't have an access to context, meaning we can't really get headers
I see. Thanks for clarifying.
> I can fix basic store as well, but we still need to keep the fix on `MeteredTimestampedKeyValueStoreWithHeaders`
Yes, makes sense. Let add the same fix to `MeteredKeyValueStore` -- we did similar improvement already in the past as part of the headers work for non-header store (ie, the work we did to alway pass headers into all serdes)
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on code in PR #21971:
URL: https://github.com/apache/kafka/pull/21971#discussion_r3589666964
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java:
##########
@@ -697,4 +702,58 @@ public void shouldUseHeadersFromValueToDeserializeKeyInPrefixScan() {
// The critical verification: key deserializer must have been called with HEADERS (not empty headers)
verify(keyDeserializer).deserialize(any(), eq(HEADERS), eq(KEY.getBytes()));
}
+
+ @Test
+ public void shouldReturnRightEntriesForHeaderDependentSerdeInPrefixScan() {
+ setUp();
+ reset(inner);
Review Comment:
needed because of strict stubs
--
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] [WIP] KAFKA-20412: Propagate headers into serde: CachingKeyValueStore [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok closed pull request #21971: [WIP] KAFKA-20412: Propagate headers into serde: CachingKeyValueStore
URL: https://github.com/apache/kafka/pull/21971
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on code in PR #21971:
URL: https://github.com/apache/kafka/pull/21971#discussion_r3573199077
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java:
##########
@@ -697,4 +699,21 @@ public void shouldUseHeadersFromValueToDeserializeKeyInPrefixScan() {
// The critical verification: key deserializer must have been called with HEADERS (not empty headers)
verify(keyDeserializer).deserialize(any(), eq(HEADERS), eq(KEY.getBytes()));
}
+
+ @Test
+ public void shouldUseHeadersFromContextForPrefixSerialization() {
Review Comment:
Overall I prefer high level / public api level tests with small number of mock, however in this case effectively there is no changes to other stores
I'm not sure this kind of test (in this particular case) will give us much value, as it will check that underlying stores are doing nothing
This test _may be_ could be useful from future regression checking point of view, but rn I cannot image how underlying stores could even brake current logic. What do you think?
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #21971:
URL: https://github.com/apache/kafka/pull/21971#issuecomment-5081377791
Thanks for the fix. Merged to `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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #21971:
URL: https://github.com/apache/kafka/pull/21971
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on code in PR #21971:
URL: https://github.com/apache/kafka/pull/21971#discussion_r3589670284
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java:
##########
@@ -697,4 +699,21 @@ public void shouldUseHeadersFromValueToDeserializeKeyInPrefixScan() {
// The critical verification: key deserializer must have been called with HEADERS (not empty headers)
verify(keyDeserializer).deserialize(any(), eq(HEADERS), eq(KEY.getBytes()));
}
+
+ @Test
+ public void shouldUseHeadersFromContextForPrefixSerialization() {
Review Comment:
I changed the test to rely on real store instead of mocking
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on code in PR #21971:
URL: https://github.com/apache/kafka/pull/21971#discussion_r3597595167
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java:
##########
@@ -438,6 +438,7 @@ private <PS extends Serializer<P>, P> KeyValueIterator<Bytes, byte[]> prefixScan
final ReadOnlyKeyValueStore<Bytes, byte[]> underlying) {
validateStoreOpen();
final KeyValueIterator<Bytes, byte[]> storeIterator = underlying.prefixScan(prefix, prefixKeySerializer);
+ // headers aren't needed because the prefix already arrives serialized
Review Comment:
> everywhere when prefixKeySerializer.serialize(...)
I did it everywhere, where we call it beside `RocksDBStore`. From what I understand while implementing this `RocksDBStore#prefixScan` is not really called anywhere directly, as all `wrappers` (like `metered`, `caching`, etc... call `range`... correct me if I'm wrong here)
> put the serialization into the metered
If you're asking from perspective of metered vs underlying: this is because underlying doesn't have an access to context, meaning we can't really get headers
If you're asking from perspective why `MeteredTimestampedKeyValueStoreWithHeaders` and not `MeteredKeyValueStore`, so that because:
- That's what jira ticket says (sorry for dumb argument :) )
- We know that headers are critical for headers-full stores, so we deferentially need to fix `MeteredTimestampedKeyValueStoreWithHeaders`, while headers propagation in `MeteredKeyValueStore` is nice to have
- Actually we can't just move code to higher level (fix in `MeteredKeyValueStore` and let child classes inherit the fix), as `prefixScan` is parallel implemented in both basic and headers versions
See: https://github.com/apache/kafka/blob/bd5fcec64e0a2ec057ca9f88da345926d8c58342/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java#L584
See: https://github.com/apache/kafka/blob/bd5fcec64e0a2ec057ca9f88da345926d8c58342/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java#L402
I can fix basic store as well, but we still need to keep the fix on `MeteredTimestampedKeyValueStoreWithHeaders`
WDYT?
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on code in PR #21971:
URL: https://github.com/apache/kafka/pull/21971#discussion_r3597595167
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java:
##########
@@ -438,6 +438,7 @@ private <PS extends Serializer<P>, P> KeyValueIterator<Bytes, byte[]> prefixScan
final ReadOnlyKeyValueStore<Bytes, byte[]> underlying) {
validateStoreOpen();
final KeyValueIterator<Bytes, byte[]> storeIterator = underlying.prefixScan(prefix, prefixKeySerializer);
+ // headers aren't needed because the prefix already arrives serialized
Review Comment:
> everywhere when prefixKeySerializer.serialize(...)
I did it everywhere, where we call it beside `RocksDBStore`. From what I understand while implementing this `RocksDBStore#prefixScan` is not really called anywhere directly, as all `wrappers` (like `metered`, `caching`, etc... call `range`... correct me if I'm wrong here)
> put the serialization into the metered
If you're asking from perspective of metered vs underlying: this is because underlying doesn't have an access to context, meaning we can't really get headers
If you're asking from perspective why `MeteredTimestampedKeyValueStoreWithHeaders` and not `MeteredKeyValueStore`, so that because:
- That's what jira ticket says (sorry for dumb argument :) )
- We know that headers are critical for headers-full stores, so we deferentially need to fix `MeteredTimestampedKeyValueStoreWithHeaders`, while headers propagation in `MeteredKeyValueStore` is nice to have
- Actually we can't just move code to higher level (fix in `MeteredKeyValueStore` and let child classes inherit the fix), as `prefixScan` is parallel implemented in both basic and headers versions
See: https://github.com/apache/kafka/blob/bd5fcec64e0a2ec057ca9f88da345926d8c58342/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java#L584
See: https://github.com/apache/kafka/blob/bd5fcec64e0a2ec057ca9f88da345926d8c58342/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java#L402
I can fix basic store as well, but we still need to keep the fix on `MeteredTimestampedKeyValueStoreWithHeaders`
WDYT?
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on PR #21971:
URL: https://github.com/apache/kafka/pull/21971#issuecomment-4934110529
Hey @UladzislauBlok. Sorry for the delay, but we are goign to merge all PRs of the parent ticket in the next two weeks. So we are back again and we need you here again. The PR needs a rebase, could you please do that? Then I go over it as final round and try to merge it asap. Thanks in advance:)
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on code in PR #21971:
URL: https://github.com/apache/kafka/pull/21971#discussion_r3582073236
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java:
##########
@@ -697,4 +699,21 @@ public void shouldUseHeadersFromValueToDeserializeKeyInPrefixScan() {
// The critical verification: key deserializer must have been called with HEADERS (not empty headers)
verify(keyDeserializer).deserialize(any(), eq(HEADERS), eq(KEY.getBytes()));
}
+
+ @Test
+ public void shouldUseHeadersFromContextForPrefixSerialization() {
Review Comment:
I see. I agree now, it makes sense to add this kind of test. Do we have any similar now? (asking because I assume this will require some setup of state store or topology and I'd like to re-use existing infra)
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on code in PR #21971:
URL: https://github.com/apache/kafka/pull/21971#discussion_r3614575459
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStoreTest.java:
##########
@@ -521,20 +522,26 @@ public void shouldThrowNullPointerOnReverseRangeIfToIsNull() {
assertThrows(NullPointerException.class, () -> metered.reverseRange("from", null));
}
+ @SuppressWarnings("unchecked")
@Test
public void shouldGetRecordsWithPrefixKey() {
setUp();
- final StringSerializer stringSerializer = new StringSerializer();
- when(inner.prefixScan(KEY, stringSerializer))
+ final Serializer<String> mockSerializer = mock(Serializer.class);
+ final Headers headers = new RecordHeaders();
+ when(context.headers()).thenReturn(headers);
Review Comment:
Yes, heavy mock test (comparing to more realistic test for metered store with headers)
This is done intentionally as
- They are simpler to setup
- Headers propagation is not critical for this store (it's more about being consistent thing)
- Generally same testing approach is used for other tests in this class
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #21971:
URL: https://github.com/apache/kafka/pull/21971#discussion_r3572692051
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java:
##########
@@ -548,7 +551,8 @@ public <PS extends Serializer<P>, P> KeyValueIterator<K, ValueTimestampHeaders<V
) {
Objects.requireNonNull(prefix, "prefix cannot be null");
Objects.requireNonNull(prefixKeySerializer, "prefixKeySerializer cannot be null");
- return prefixScanInternal(wrapped(), prefix, prefixKeySerializer);
+ final byte[] keyBytes = prefixKeySerializer.serialize(null, internalContext.headers(), prefix);
+ return prefixScanInternal(wrapped(), keyBytes, BYTE_ARRAY_SERIALIZER);
Review Comment:
The `readOnly(isolationLevel)` view's `prefixScan` (in `ReadOnlyHeadersView`) still forwards the raw prefix and serializer via `prefixScanInternal(underlying, prefix, prefixKeySerializer)` without serializing the prefix with headers — the same bug you're fixing here. Its `get`/`range` already use the header-aware helpers, so this path is left inconsistent. Apply the same fix there and add a test for `readOnly().prefixScan(...)`.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java:
##########
@@ -238,6 +238,7 @@ public <PS extends Serializer<P>, P> KeyValueIterator<Bytes, byte[]> prefixScan(
private <PS extends Serializer<P>, P> KeyValueIterator<Bytes, byte[]> prefixScan(final P prefix, final PS prefixKeySerializer,
final IsolationLevel isolationLevel) {
+ // Header propagation is not required here because serialization was already performed with headers in the metered store
Review Comment:
This wording is misleading: this store is also used by the plain `MeteredKeyValueStore`, where the prefix is serialized right here, not "in the metered store". Reword to something like: headers aren't needed because the prefix already arrives serialized. Same comment in `CachingKeyValueStore` and `MemoryNavigableLRUCache`.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java:
##########
@@ -697,4 +699,21 @@ public void shouldUseHeadersFromValueToDeserializeKeyInPrefixScan() {
// The critical verification: key deserializer must have been called with HEADERS (not empty headers)
verify(keyDeserializer).deserialize(any(), eq(HEADERS), eq(KEY.getBytes()));
}
+
+ @Test
+ public void shouldUseHeadersFromContextForPrefixSerialization() {
Review Comment:
This only checks the mock got the right args. There's no test driving the real metered → caching → inner chain with caching on and non-empty headers to prove prefixScan returns the correct entries. I was thinking if we should add something?
--
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-20173: Propagate headers into serde: CachingKeyValueStore [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21971:
URL: https://github.com/apache/kafka/pull/21971#discussion_r3049137368
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java:
##########
@@ -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));
Review Comment:
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. 🤔
--
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-20173: Propagate headers into serde: CachingKeyValueStore [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #21971:
URL: https://github.com/apache/kafka/pull/21971#issuecomment-4203774200
We filed a new ticket to track the `prefixScan` work: https://issues.apache.org/jira/browse/KAFKA-20412 -- If you want to tackle it end-to-end, please assign the ticket to yourself, update the KAFKA-XXX number on this PR.
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on code in PR #21971:
URL: https://github.com/apache/kafka/pull/21971#discussion_r3573300657
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java:
##########
@@ -548,7 +551,8 @@ public <PS extends Serializer<P>, P> KeyValueIterator<K, ValueTimestampHeaders<V
) {
Objects.requireNonNull(prefix, "prefix cannot be null");
Objects.requireNonNull(prefixKeySerializer, "prefixKeySerializer cannot be null");
- return prefixScanInternal(wrapped(), prefix, prefixKeySerializer);
+ final byte[] keyBytes = prefixKeySerializer.serialize(null, internalContext.headers(), prefix);
+ return prefixScanInternal(wrapped(), keyBytes, BYTE_ARRAY_SERIALIZER);
Review Comment:
Wow. This actually good catch! When I first implemented this fix, `IsolationLevel` KIP wasn't here yet, and after rebase I didn't realized it also has to be fixed :)
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #21971:
URL: https://github.com/apache/kafka/pull/21971#discussion_r3640175494
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java:
##########
@@ -868,4 +873,58 @@ public void shouldUseHeadersFromValueToDeserializeKeyInPrefixScan() {
// The critical verification: key deserializer must have been called with HEADERS (not empty headers)
verify(keyDeserializer).deserialize(any(), eq(HEADERS), eq(KEY.getBytes()));
}
+
+ @Test
+ public void shouldReturnRightEntriesForHeaderDependentSerdeInPrefixScan() {
+ setUp();
+ reset(inner);
+ when(context.headers()).thenReturn(HEADERS);
+
+ final InMemoryKeyValueStore inMemoryStore = new InMemoryKeyValueStore(STORE_NAME);
Review Comment:
Just wondering if the caching path (`CachingKeyValueStoreWithHeaders`) for a headers-dependent prefix should be tested too?!
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #21971:
URL: https://github.com/apache/kafka/pull/21971#discussion_r3575010257
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java:
##########
@@ -697,4 +699,21 @@ public void shouldUseHeadersFromValueToDeserializeKeyInPrefixScan() {
// The critical verification: key deserializer must have been called with HEADERS (not empty headers)
verify(keyDeserializer).deserialize(any(), eq(HEADERS), eq(KEY.getBytes()));
}
+
+ @Test
+ public void shouldUseHeadersFromContextForPrefixSerialization() {
Review Comment:
I agree with your point — we shouldn't add tests per underlying store, but my concern wasn't about the underlying stores; it was about the outcome. Keys are stored serialized with headers, and the bug was exactly that prefixScan serialized the prefix without them, so the range didn't match the stored keys. The mock test only asserts the plumbing (serializer called with headers, inner called with the bytes) — it never proves that a header-dependent serde produces a prefix range that actually matches the stored keys and returns the right entries.
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on PR #21971:
URL: https://github.com/apache/kafka/pull/21971#issuecomment-4948103003
@aliehsaeedii Hello. Thanks for letting em know. Rebase is done, PR is ready for review
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21971:
URL: https://github.com/apache/kafka/pull/21971#discussion_r3598508720
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java:
##########
@@ -438,6 +438,7 @@ private <PS extends Serializer<P>, P> KeyValueIterator<Bytes, byte[]> prefixScan
final ReadOnlyKeyValueStore<Bytes, byte[]> underlying) {
validateStoreOpen();
final KeyValueIterator<Bytes, byte[]> storeIterator = underlying.prefixScan(prefix, prefixKeySerializer);
+ // headers aren't needed because the prefix already arrives serialized
Review Comment:
> If you're asking from perspective of metered vs underlying: this is because underlying doesn't have an access to context, meaning we can't really get headers
I see. Thanks for clarifying.
> I can fix basic store as well, but we still need to keep the fix on `MeteredTimestampedKeyValueStoreWithHeaders`
Yes, makes sense. Let add the same fix to `MeteredKeyValueStore` -- we did similar improvement already in the past as part of the headers work for non-header store (ie, the work we did to alway pass headers into all serdes)
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on code in PR #21971:
URL: https://github.com/apache/kafka/pull/21971#discussion_r3646510650
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java:
##########
@@ -868,4 +873,58 @@ public void shouldUseHeadersFromValueToDeserializeKeyInPrefixScan() {
// The critical verification: key deserializer must have been called with HEADERS (not empty headers)
verify(keyDeserializer).deserialize(any(), eq(HEADERS), eq(KEY.getBytes()));
}
+
+ @Test
+ public void shouldReturnRightEntriesForHeaderDependentSerdeInPrefixScan() {
+ setUp();
+ reset(inner);
+ when(context.headers()).thenReturn(HEADERS);
+
+ final InMemoryKeyValueStore inMemoryStore = new InMemoryKeyValueStore(STORE_NAME);
Review Comment:
actually we can just add caching layer, wdyt?
pushed this version for review
--
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-20412: Fix prefixScan for KV-Store with headers [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on PR #21971:
URL: https://github.com/apache/kafka/pull/21971#issuecomment-4698515206
@aliehsaeedii Hello
This PR was already waiting for some time. Can you take a look pls? It's fix for header-full prefix scan
--
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