KAFKA-20173: Metered layer of KV-stores needs to pass Headers by mjsax · Pull Request #21684 · apache/kafka

mjsax · 2026-03-09T03:45:06Z

Updates the metered ks-stores layer to pass the context headers into
serdes. Simplifies the code with some refactoring.

mjsax · 2026-03-09T03:49:07Z

streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java

-    protected V outerValue(final byte[] value) {
-        return value != null ? serdes.valueFrom(value, new RecordHeaders()) : null;
+    protected byte[] serializeValue(final V value) {
+        return value != null ? serdes.rawValue(value, internalContext.headers()) : null;


This is the key question -- should we pass context.headers() or new RecordHeaders for non-header stores. Both solutions have advantages and disadvantages.

Using new RecordHeaders is strictly more backward compatible; the idea to pass in context.headers() feels like a "bug fix" though -- we should have always done this IMHO. That's why I opted for this solution.

Of course, we can also "fix" this "bug" by arguing: we stay 100% backward compatible and pass in new RecordHeaders and user get the fix by enabling the new headers stores...

Let me know what you think.

IMHO. I think new RecordHeaders looks more correct (from backward compatibility POV), but essentially that's okay to do context.headers() because:

If user doesn't use headers aware serializer, headers will be ignored (ether empty or not)

The idea of this ticket is to actually fix code base to propagate headers, and if we can propagate original headers instead of "mocked" one we go for it

It doesn't make any difference if users aren't using headers state stores / serializers, but helps us to consistently apply simple paradigm: if there is a access to original headers -> go for it, if not fallback to new RecordHeaders

@mjsax
Regarding backward compability, I think this risk is very low because, most serdes ignore headers - the default interface methods delegate to the non-headers versions. Using internalContext.headers() is semantically correct behavior - serdes should have access to record context.
1Q: should we check if internalContext != null?

1Q: should we check if internalContext != null?

I would say no? If internalContext would be null, it would be a bug (I believe), so we should rather expose such an issue directly, and fail fast by crashing, and not "mask" the bug by not failing?

mjsax · 2026-03-09T03:49:34Z

.../src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStore.java

    }

-    public RawAndDeserializedValue<V> getWithBinary(final K key) {
+    RawAndDeserializedValue<V> getWithBinary(final K key) {


Side cleanup

UladzislauBlok

LGTM Overall. Left few small comments
FYI:

Found 3 test failures:
FAILED ❌ MockProcessorContextAPITest > shouldStoreAndReturnStateStores()
FAILED ❌ WordCountProcessorTest > test()
FAILED ❌ WordCountTransformerTest > test()

UladzislauBlok · 2026-03-09T10:14:56Z

streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java

+    MeteredKeyValueStore(
+        final KeyValueStore<Bytes, byte[]> inner,
+        final String metricsScope,
+        final Time time,
+        final Serde<K> keySerde,
+        final Serde<V> valueSerde
+    ) {


Offtopic

Question about formatting fix: do we have consistent code style across code base? I mean literally cli formatted or ide setting
I was able to find this: https://kafka.apache.org/community/developer/#streams-api , but I don't think that's enough to keep code style consistent

No we don't have anything that would do strict enforcement...

UladzislauBlok · 2026-03-09T10:31:35Z

streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java

-    protected V outerValue(final byte[] value) {
-        return value != null ? serdes.valueFrom(value, new RecordHeaders()) : null;
+    protected byte[] serializeValue(final V value) {
+        return value != null ? serdes.rawValue(value, internalContext.headers()) : null;


IMHO. I think new RecordHeaders looks more correct (from backward compatibility POV), but essentially that's okay to do context.headers() because:

If user doesn't use headers aware serializer, headers will be ignored (ether empty or not)

The idea of this ticket is to actually fix code base to propagate headers, and if we can propagate original headers instead of "mocked" one we go for it

It doesn't make any difference if users aren't using headers state stores / serializers, but helps us to consistently apply simple paradigm: if there is a access to original headers -> go for it, if not fallback to new RecordHeaders

UladzislauBlok · 2026-03-09T10:40:20Z

...ava/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java

+        throw new UnsupportedOperationException("Position is not supported for " + getClass().getSimpleName());
    }

    protected Bytes keyBytes(final K key, final Headers headers) {


Do we still need this method, if we already have it from parent store?
see https://github.com/apache/kafka/pull/21684/changes#diff-9af87381ff50464fc0979726bd22231c747d941b3bb7a9c152ddbf430c61cf23R444

I was focusing on MeteredKeyValueStore only, as the PR is already large enough. -- Would need to revisit both MeteredTimestampedKeyValueStore and MeteredTimestampedKeyValueStoreWithHeaders as follow up to further cleanup, in a follow up PR.

UladzislauBlok · 2026-03-09T10:42:01Z

...ms/src/main/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStore.java

            Objects.requireNonNull(key, "key cannot be null");
            try {
-                final long validTo = maybeMeasureLatency(() -> inner.put(keyBytes(key), plainValueSerdes.rawValue(value), timestamp), time, putSensor);
+                final long validTo = maybeMeasureLatency(() -> inner.put(serializeKey(key), plainValueSerdes.rawValue(value), timestamp), time, putSensor);


plainValueSerdes.rawValue(value). do we need headers there?

Good catch -- similar to my other comment -- focus of this PR was MeteredKeyValueStore -- we haven't even implemented "header-versioned-store" yet. Will address in follow up PR.

UladzislauBlok · 2026-03-09T10:45:22Z

streams/src/test/java/org/apache/kafka/streams/state/StateSerdesTest.java


    @Test
    public void shouldThrowIfIncompatibleSerdeForKey() throws ClassNotFoundException {
+        @SuppressWarnings("rawtypes")


out of curiosity: why new RecordHeaders() leads to @SuppressWarnings("rawtypes")?

It doesn't -- this is additional side cleanup. We use Class below, which is the offender (and we cannot switch to Class<?> either to avoid the rawtype, because we want to test the wrong type condition...)

Updates the metered ks-stores layer to pass the context headers into serdes. Simplifies the code with some refactoring.

UladzislauBlok · 2026-03-10T19:55:44Z

...amples/src/test/java/org/apache/kafka/streams/examples/wordcount/WordCountProcessorTest.java

-        store.init(context.getStateStoreContext(), store);
+        store.init(
+            new AbstractProcessorContext<>(new TaskId(0, 0), new StreamsConfig(context.appConfigs()), (StreamsMetricsImpl) context.metrics(), null) {


I assume MockProcessorContext doesn't have a method to mock headers, and that's the reason why you re-implemented it, right?

btw, don't think this comment is still correct:

/** * Demonstrate the use of {@link MockProcessorContext} for testing the {@link Processor} in the {@link WordCountProcessorDemo}. */

Oh. That's a good point. I totally missed that this is example code... Need to do this differently.

Seems this is related to https://issues.apache.org/jira/browse/KAFKA-19983

UladzislauBlok · 2026-03-10T19:59:07Z

...ples/src/test/java/org/apache/kafka/streams/examples/wordcount/WordCountTransformerTest.java

                .withLoggingDisabled() // Changelog is not supported by MockProcessorContext.
                // Caching is disabled by default, but FYI: caching is also not supported by MockProcessorContext.
                .build();
-            store.init(context.getStateStoreContext(), store);


aliehsaeedii

LGTM, thanks.

aliehsaeedii · 2026-03-11T00:29:38Z

streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java

-    protected V outerValue(final byte[] value) {
-        return value != null ? serdes.valueFrom(value, new RecordHeaders()) : null;
+    protected byte[] serializeValue(final V value) {
+        return value != null ? serdes.rawValue(value, internalContext.headers()) : null;


@mjsax
Regarding backward compability, I think this risk is very low because, most serdes ignore headers - the default interface methods delegate to the non-headers versions. Using internalContext.headers() is semantically correct behavior - serdes should have access to record context.
1Q: should we check if internalContext != null?

mjsax added streams kip Requires or implements a KIP labels Mar 9, 2026

mjsax commented Mar 9, 2026

View reviewed changes

mjsax changed the title ~~KAFKA-20173: Metered layer of kv-stores need to pass Headers~~ KAFKA-20173: Metered layer of KV-stores needs to pass Headers Mar 9, 2026

UladzislauBlok reviewed Mar 9, 2026

View reviewed changes

mjsax added 4 commits March 9, 2026 12:14

KAFKA-20173: Metered layer of kv-stores need to pass Headers

fc054ac

Updates the metered ks-stores layer to pass the context headers into serdes. Simplifies the code with some refactoring.

spotless

e5aa8d2

fix

06008e1

fix tests

ee99c2c

mjsax force-pushed the kafka-20173-meteredkeyvaluestore-headers branch from 6f5fc81 to ee99c2c Compare March 9, 2026 19:42

checkstyle

0343c08

UladzislauBlok reviewed Mar 10, 2026

View reviewed changes

aliehsaeedii approved these changes Mar 11, 2026

View reviewed changes

review comments

0e0d030

mjsax mentioned this pull request Mar 11, 2026

KAFKA-19983: Fix MockProcessorContext doesn't work with WindowStores. #21677

Draft

Conversation

mjsax commented Mar 9, 2026 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

UladzislauBlok left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Offtopic

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

UladzislauBlok Mar 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax Mar 11, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aliehsaeedii left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

mjsax commented Mar 9, 2026 •

edited by github-actions bot

Loading

UladzislauBlok left a comment •

edited

Loading

UladzislauBlok Mar 10, 2026 •

edited

Loading

mjsax Mar 11, 2026 •

edited

Loading