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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
