You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kafka.apache.org by Alieh Saeedi via dev <de...@kafka.apache.org> on 2026/02/18 11:30:53 UTC
[DISCUSS] KIP-1285: DSL Opt-in Support for Header-Aware State Stores
Hi all,
I’d like to start a discussion on a new KIP, KIP-1285: DSL Opt-in Support
for Header-Aware State Stores.
KIP-1271 introduced headers-aware state stores in Kafka Streams. KIP-1285
proposes adding corresponding DSL-level opt-in support, so that DSL
applications can take advantage of these headers-aware stores while keeping
existing applications and topologies unchanged by default.
You can find the KIP here:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-1285%3A+DSL+Opt-in+Support+for+Header-Aware+State+Stores
Feedback would be greatly appreciated.
Best,
Alieh
Re: [DISCUSS] KIP-1285: DSL Opt-in Support for Header-Aware State Stores
Posted by Bill Bejeck <bb...@gmail.com>.
> Not really. In most cases DEFAULT already maps to a timestamped store.
OK, thanks for the clarification Alieh.
On Mon, Feb 23, 2026 at 6:51 PM Matthias J. Sax <mj...@apache.org> wrote:
> Thanks for updating the KIP. LGTM.
>
> On 2/23/26 11:46 AM, Alieh Saeedi via dev wrote:
> > Hey Bill,
> >
> >> I'm assuming we are going to add some additional types to the
> > DslStoreFormat enum to include timestamped types?
> >
> > Not really. In most cases DEFAULT already maps to a timestamped store.
> For
> > session stores, though, DEFAULT means the normal session store because we
> > don’t have a timestamped variant there. Even today (before this KIP), the
> > default is timestamped and there’s no way to globally configure the DSL
> to
> > use non-timestamped stores instead.
> >
> > Thanks,
> > Alieh
> >
> > On Mon, Feb 23, 2026 at 5:39 PM Bill Bejeck <bb...@gmail.com> wrote:
> >
> >> Hi Alieh,
> >>
> >> Thanks for the KIP!
> >>
> >> I agree with everything stated so far.
> >> In the KIP, we are deprecating the `isTimestamped` field and the
> >> associated getter, but the `DslStoreFormat` enum only specifies
> `DEFAULT`
> >> and `HEADERS`.
> >> I'm assuming we are going to add some additional types to the
> >> DslStoreFormat enum to include timestamped types?
> >>
> >> Otherwise this looks good to me.
> >>
> >> Thanks,
> >> Bill
> >>
> >> On Mon, Feb 23, 2026 at 8:48 AM Alieh Saeedi via dev <
> dev@kafka.apache.org>
> >> wrote:
> >>
> >>> Thanks Matthias.
> >>>
> >>> Good points! I updated the KIP accordingly.
> >>>
> >>> Bests
> >>> Alieh
> >>>
> >>> On Mon, Feb 23, 2026 at 11:15 AM Lucas Brutschy via dev <
> >>> dev@kafka.apache.org> wrote:
> >>>
> >>>> Thanks for the updates Alieh!
> >>>>
> >>>> On Sun, Feb 22, 2026 at 8:52 PM Matthias J. Sax <mj...@apache.org>
> >>> wrote:
> >>>>>
> >>>>> Thanks Alieh.
> >>>>>
> >>>>> I like Lucas' idea about "format" vs "type".
> >>>>>
> >>>>>
> >>>>> For the downgrade path, KIP-1271 says:
> >>>>>
> >>>>>> Downgrades are not supported; the only way to achieve one is by
> >>>> deleting the local store and rebuilding it from the changelog.
> >>>>>
> >>>>> While KIP-1285 only say:
> >>>>>
> >>>>>> The same as in KIP-1271, downgrades are not supported.
> >>>>>
> >>>>> Might be good to align both?
> >>>>>
> >>>>>
> >>>>>
> >>>>> One last question: Should we call out that
> >>>>> `Materialized.as(KeyValueBytesStoreSupplier)` (same for window and
> >>>>> session case), will also allow to enable a headers store on a
> >>> per-store
> >>>>> basis, similar to the plain-kv / ts-case? Ie, if one uses
> >>>>>
> >>>>>> Materialized.as(
> >>>>>> Stores.persistentTimestampedKeyValueStoreWithHeaders(...)
> >>>>>> )
> >>>>>
> >>>>> The corresponding DSL operator would also use a header store,
> >>>>> overwriting whatever is configured via `StreamsConfig`.
> >>>>>
> >>>>>
> >>>>> While it's unfortunately not well documented, it is currently
> possible
> >>>>> to pass in both types of supplier, and the DSL would "bridge" between
> >>>>> both store types. Thus, it's currently possible, to disable ts-stores
> >>> in
> >>>>> the DSL via
> >>>>>
> >>>>>> Materialized.as(
> >>>>>> Stores.persistentKeyValueStore(...)
> >>>>>> )
> >>>>>
> >>>>> The determining factor is, if the passed-in supplier implements
> >>>>> `TimestampedBytesStore` or not and we added a similar marker
> interface
> >>>>> `HeaderBytesStore` with KIP-1271, to we can do the same thing for
> >>> header
> >>>>> stores.
> >>>>>
> >>>>> This should not imply any more code changes I believe, but should
> work
> >>>>> this way automatically base on KIP-1285 proposal. I think it's mainly
> >>> an
> >>>>> documentation question.
> >>>>>
> >>>>>
> >>>>>
> >>>>> -Matthias
> >>>>>
> >>>>>
> >>>>>
> >>>>>
> >>>>> On 2/22/26 6:16 AM, Alieh Saeedi via dev wrote:
> >>>>>> Thanks Lucas for the feedback.
> >>>>>>
> >>>>>> LB1) dsl.store.type makes sense to me, but I’m not attached to it. I
> >>>> can
> >>>>>> change it (and already have changed it) to dsl.store.format if we
> >>>> prefer
> >>>>>> that naming.
> >>>>>>
> >>>>>> LB2) Since Kafka Streams does not have a timestamped session store,
> >>> we
> >>>>>> don’t have any DslSessionParams constructor with isTimestamped.
> >>> What
> >>>> we
> >>>>>> are deprecating in the KIP is the older constructor that does not
> >>> take
> >>>> a
> >>>>>> DslStoreType (now DslStoreFormat). To keep things cleaner, we only
> >>>> keep a
> >>>>>> single constructor that also includes DslStoreFormat. For session
> >>>> stores,
> >>>>>> DEFAULT corresponds to the regular session store, and HEADERS
> >>>> corresponds
> >>>>>> to SessionStoreWithHeaders. This detail is already mentioned in the
> >>>> KIP.
> >>>>>>
> >>>>>> LB3) Because this KIP does not introduce any new state store
> >>> formats,
> >>>> the
> >>>>>> upgrade and downgrade paths are identical to those described in
> >>>> KIP-1271.
> >>>>>> I’ve added this note to the KIP as well.
> >>>>>>
> >>>>>> Thanks,
> >>>>>> Alieh
> >>>>>>
> >>>>>> On Fri, Feb 20, 2026 at 9:08 PM Lucas Brutschy <
> >>> lbrutschy@confluent.io
> >>>>>
> >>>>>> wrote:
> >>>>>>
> >>>>>>> Hi Alieh,
> >>>>>>>
> >>>>>>> Thanks for the KIP. A few thoughts below.
> >>>>>>>
> >>>>>>> LB1: The config name dsl.store.type feels a bit ambiguous since we
> >>>>>>> already have dsl.store.suppliers.class (which controls RocksDB vs.
> >>>>>>> in-memory). dsl.store.type could be confused with that distinction.
> >>>>>>> Would something like dsl.store.format be clearer, since this is
> >>> really
> >>>>>>> about the value format (timestamped vs. timestamped+headers)?
> >>>>>>>
> >>>>>>> LB2: Regarding DslSessionParams: the current code entirely lacks an
> >>>>>>> isTimestamped field or constructor parameter. The KIP shows
> >>>>>>> deprecating a constructor with isTimestamped on DslSessionParams,
> >>> but
> >>>>>>> that constructor does not exist today. Could you clarify whether
> >>> the
> >>>>>>> KIP intends to add and immediately deprecate the constructor, or
> >>>>>>> whether the old DslSessionParams constructor should simply remain
> >>>>>>> as-is and a new overload with DslStoreType should be added
> >>> alongside
> >>>>>>> it?
> >>>>>>>
> >>>>>>> LB3: Regarding compatibility, the KIP mentions that switching from
> >>>>>>> DEFAULT to HEADERS triggers KIP-1271 upgrade paths. It would be
> >>> good
> >>>>>>> to clarify whether switching back from HEADERS to DEFAULT is also
> >>>>>>> supported (i.e., is it a two-way migration or one-way only?). If
> >>> it is
> >>>>>>> one-way, that must be documented explicitly.
> >>>>>>>
> >>>>>>> Thanks,
> >>>>>>> Lucas
> >>>>>>>
> >>>>>>> On Fri, Feb 20, 2026 at 10:30 AM Alieh Saeedi via dev
> >>>>>>> <de...@kafka.apache.org> wrote:
> >>>>>>>>
> >>>>>>>> Hey Matthias,
> >>>>>>>> Thanks for the suggestions. I agree with all of them. They all
> >>> make
> >>>> sense
> >>>>>>>> to me. I updated the KIP accordingly.
> >>>>>>>>
> >>>>>>>> Bests,
> >>>>>>>> Alieh
> >>>>>>>>
> >>>>>>>> On Thu, Feb 19, 2026 at 8:33 PM Matthias J. Sax <mjsax@apache.org
> >>>>
> >>>>>>> wrote:
> >>>>>>>>
> >>>>>>>>> Thanks for the KIP. Great follow up to KIP-1271.
> >>>>>>>>>
> >>>>>>>>>
> >>>>>>>>>
> >>>>>>>>> Nit: The `DSL_STORE_TYPE_DOC` should be `private` and I also
> >>> think
> >>>> we
> >>>>>>>>> can improve the wording -- but that's more of am impl detail, not
> >>>>>>> really
> >>>>>>>>> critical for the KIP itself.
> >>>>>>>>>
> >>>>>>>>> How about:
> >>>>>>>>>
> >>>>>>>>> "Controls the state store type used by the DSL store supplier
> >>> (see
> >>>>>>>>> config 'dsl.store.suppliers.class'). 'default' uses timestamped
> >>>> stores.
> >>>>>>>>> 'headers' uses the headers-aware stores."
> >>>>>>>>>
> >>>>>>>>>
> >>>>>>>>>
> >>>>>>>>> MJS1: Should we deprecate the existing `DslKeyValueParams(final
> >>>> String
> >>>>>>>>> name, final boolean isTimestamped)` constructor, and make the
> >>> newly
> >>>>>>>>> introduced "type" a mandatory parameter? (Cf MJS2 below)
> >>>>>>>>>
> >>>>>>>>>
> >>>>>>>>>
> >>>>>>>>> MJS2: Should we add new enum `DslStoreType` with DEFAULT and
> >>> HEADERS
> >>>>>>>>> value, and change the constructor
> >>>>>>>>>
> >>>>>>>>> from
> >>>>>>>>> - DslKeyValueParams(final String name, final boolean
> >>>> isTimestamped,
> >>>>>>>>> final boolean isHeadersEnabled)
> >>>>>>>>>
> >>>>>>>>> to
> >>>>>>>>> - DslKeyValueParams(final String name, final DslStoreType
> >>>> storeType)
> >>>>>>>>>
> >>>>>>>>> This would more align to the newly added String config (and the
> >>> KIP
> >>>>>>>>> calls out that we did not want to add a boolean config, what
> >>> makes
> >>>>>>> sense
> >>>>>>>>> to me), and would subsume the existing `isTimestamped` parameter
> >>> on
> >>>> the
> >>>>>>>>> existing constructor which we could remove (Cf MJS1)
> >>>>>>>>>
> >>>>>>>>> [We have a similar thing for `group.protocol` config.]
> >>>>>>>>>
> >>>>>>>>>
> >>>>>>>>> If yes, wondering if we should also deprecate `isTimestamped()`
> >>> and
> >>>> not
> >>>>>>>>> add `headersEnabled()` but instead add `storeType()` getter?
> >>>>>>>>>
> >>>>>>>>> Similar for window/session Dsl*Parameter classes.
> >>>>>>>>>
> >>>>>>>>>
> >>>>>>>>>
> >>>>>>>>> -Matthias
> >>>>>>>>>
> >>>>>>>>>
> >>>>>>>>> On 2/18/26 3:30 AM, Alieh Saeedi via dev wrote:
> >>>>>>>>>> Hi all,
> >>>>>>>>>>
> >>>>>>>>>> I’d like to start a discussion on a new KIP, KIP-1285: DSL
> >>> Opt-in
> >>>>>>> Support
> >>>>>>>>>> for Header-Aware State Stores.
> >>>>>>>>>>
> >>>>>>>>>> KIP-1271 introduced headers-aware state stores in Kafka Streams.
> >>>>>>> KIP-1285
> >>>>>>>>>> proposes adding corresponding DSL-level opt-in support, so that
> >>> DSL
> >>>>>>>>>> applications can take advantage of these headers-aware stores
> >>> while
> >>>>>>>>> keeping
> >>>>>>>>>> existing applications and topologies unchanged by default.
> >>>>>>>>>>
> >>>>>>>>>> You can find the KIP here:
> >>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>>
> >>>>>>>
> >>>>
> >>>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-1285%3A+DSL+Opt-in+Support+for+Header-Aware+State+Stores
> >>>>>>>>>>
> >>>>>>>>>> Feedback would be greatly appreciated.
> >>>>>>>>>>
> >>>>>>>>>> Best,
> >>>>>>>>>> Alieh
> >>>>>>>>>>
> >>>>>>>>>
> >>>>>>>>>
> >>>>>>>
> >>>>>>
> >>>>>
> >>>>
> >>>
> >>
> >
>
>
Re: [DISCUSS] KIP-1285: DSL Opt-in Support for Header-Aware State Stores
Posted by "Matthias J. Sax" <mj...@apache.org>.
Thanks for updating the KIP. LGTM.
On 2/23/26 11:46 AM, Alieh Saeedi via dev wrote:
> Hey Bill,
>
>> I'm assuming we are going to add some additional types to the
> DslStoreFormat enum to include timestamped types?
>
> Not really. In most cases DEFAULT already maps to a timestamped store. For
> session stores, though, DEFAULT means the normal session store because we
> don’t have a timestamped variant there. Even today (before this KIP), the
> default is timestamped and there’s no way to globally configure the DSL to
> use non-timestamped stores instead.
>
> Thanks,
> Alieh
>
> On Mon, Feb 23, 2026 at 5:39 PM Bill Bejeck <bb...@gmail.com> wrote:
>
>> Hi Alieh,
>>
>> Thanks for the KIP!
>>
>> I agree with everything stated so far.
>> In the KIP, we are deprecating the `isTimestamped` field and the
>> associated getter, but the `DslStoreFormat` enum only specifies `DEFAULT`
>> and `HEADERS`.
>> I'm assuming we are going to add some additional types to the
>> DslStoreFormat enum to include timestamped types?
>>
>> Otherwise this looks good to me.
>>
>> Thanks,
>> Bill
>>
>> On Mon, Feb 23, 2026 at 8:48 AM Alieh Saeedi via dev <de...@kafka.apache.org>
>> wrote:
>>
>>> Thanks Matthias.
>>>
>>> Good points! I updated the KIP accordingly.
>>>
>>> Bests
>>> Alieh
>>>
>>> On Mon, Feb 23, 2026 at 11:15 AM Lucas Brutschy via dev <
>>> dev@kafka.apache.org> wrote:
>>>
>>>> Thanks for the updates Alieh!
>>>>
>>>> On Sun, Feb 22, 2026 at 8:52 PM Matthias J. Sax <mj...@apache.org>
>>> wrote:
>>>>>
>>>>> Thanks Alieh.
>>>>>
>>>>> I like Lucas' idea about "format" vs "type".
>>>>>
>>>>>
>>>>> For the downgrade path, KIP-1271 says:
>>>>>
>>>>>> Downgrades are not supported; the only way to achieve one is by
>>>> deleting the local store and rebuilding it from the changelog.
>>>>>
>>>>> While KIP-1285 only say:
>>>>>
>>>>>> The same as in KIP-1271, downgrades are not supported.
>>>>>
>>>>> Might be good to align both?
>>>>>
>>>>>
>>>>>
>>>>> One last question: Should we call out that
>>>>> `Materialized.as(KeyValueBytesStoreSupplier)` (same for window and
>>>>> session case), will also allow to enable a headers store on a
>>> per-store
>>>>> basis, similar to the plain-kv / ts-case? Ie, if one uses
>>>>>
>>>>>> Materialized.as(
>>>>>> Stores.persistentTimestampedKeyValueStoreWithHeaders(...)
>>>>>> )
>>>>>
>>>>> The corresponding DSL operator would also use a header store,
>>>>> overwriting whatever is configured via `StreamsConfig`.
>>>>>
>>>>>
>>>>> While it's unfortunately not well documented, it is currently possible
>>>>> to pass in both types of supplier, and the DSL would "bridge" between
>>>>> both store types. Thus, it's currently possible, to disable ts-stores
>>> in
>>>>> the DSL via
>>>>>
>>>>>> Materialized.as(
>>>>>> Stores.persistentKeyValueStore(...)
>>>>>> )
>>>>>
>>>>> The determining factor is, if the passed-in supplier implements
>>>>> `TimestampedBytesStore` or not and we added a similar marker interface
>>>>> `HeaderBytesStore` with KIP-1271, to we can do the same thing for
>>> header
>>>>> stores.
>>>>>
>>>>> This should not imply any more code changes I believe, but should work
>>>>> this way automatically base on KIP-1285 proposal. I think it's mainly
>>> an
>>>>> documentation question.
>>>>>
>>>>>
>>>>>
>>>>> -Matthias
>>>>>
>>>>>
>>>>>
>>>>>
>>>>> On 2/22/26 6:16 AM, Alieh Saeedi via dev wrote:
>>>>>> Thanks Lucas for the feedback.
>>>>>>
>>>>>> LB1) dsl.store.type makes sense to me, but I’m not attached to it. I
>>>> can
>>>>>> change it (and already have changed it) to dsl.store.format if we
>>>> prefer
>>>>>> that naming.
>>>>>>
>>>>>> LB2) Since Kafka Streams does not have a timestamped session store,
>>> we
>>>>>> don’t have any DslSessionParams constructor with isTimestamped.
>>> What
>>>> we
>>>>>> are deprecating in the KIP is the older constructor that does not
>>> take
>>>> a
>>>>>> DslStoreType (now DslStoreFormat). To keep things cleaner, we only
>>>> keep a
>>>>>> single constructor that also includes DslStoreFormat. For session
>>>> stores,
>>>>>> DEFAULT corresponds to the regular session store, and HEADERS
>>>> corresponds
>>>>>> to SessionStoreWithHeaders. This detail is already mentioned in the
>>>> KIP.
>>>>>>
>>>>>> LB3) Because this KIP does not introduce any new state store
>>> formats,
>>>> the
>>>>>> upgrade and downgrade paths are identical to those described in
>>>> KIP-1271.
>>>>>> I’ve added this note to the KIP as well.
>>>>>>
>>>>>> Thanks,
>>>>>> Alieh
>>>>>>
>>>>>> On Fri, Feb 20, 2026 at 9:08 PM Lucas Brutschy <
>>> lbrutschy@confluent.io
>>>>>
>>>>>> wrote:
>>>>>>
>>>>>>> Hi Alieh,
>>>>>>>
>>>>>>> Thanks for the KIP. A few thoughts below.
>>>>>>>
>>>>>>> LB1: The config name dsl.store.type feels a bit ambiguous since we
>>>>>>> already have dsl.store.suppliers.class (which controls RocksDB vs.
>>>>>>> in-memory). dsl.store.type could be confused with that distinction.
>>>>>>> Would something like dsl.store.format be clearer, since this is
>>> really
>>>>>>> about the value format (timestamped vs. timestamped+headers)?
>>>>>>>
>>>>>>> LB2: Regarding DslSessionParams: the current code entirely lacks an
>>>>>>> isTimestamped field or constructor parameter. The KIP shows
>>>>>>> deprecating a constructor with isTimestamped on DslSessionParams,
>>> but
>>>>>>> that constructor does not exist today. Could you clarify whether
>>> the
>>>>>>> KIP intends to add and immediately deprecate the constructor, or
>>>>>>> whether the old DslSessionParams constructor should simply remain
>>>>>>> as-is and a new overload with DslStoreType should be added
>>> alongside
>>>>>>> it?
>>>>>>>
>>>>>>> LB3: Regarding compatibility, the KIP mentions that switching from
>>>>>>> DEFAULT to HEADERS triggers KIP-1271 upgrade paths. It would be
>>> good
>>>>>>> to clarify whether switching back from HEADERS to DEFAULT is also
>>>>>>> supported (i.e., is it a two-way migration or one-way only?). If
>>> it is
>>>>>>> one-way, that must be documented explicitly.
>>>>>>>
>>>>>>> Thanks,
>>>>>>> Lucas
>>>>>>>
>>>>>>> On Fri, Feb 20, 2026 at 10:30 AM Alieh Saeedi via dev
>>>>>>> <de...@kafka.apache.org> wrote:
>>>>>>>>
>>>>>>>> Hey Matthias,
>>>>>>>> Thanks for the suggestions. I agree with all of them. They all
>>> make
>>>> sense
>>>>>>>> to me. I updated the KIP accordingly.
>>>>>>>>
>>>>>>>> Bests,
>>>>>>>> Alieh
>>>>>>>>
>>>>>>>> On Thu, Feb 19, 2026 at 8:33 PM Matthias J. Sax <mjsax@apache.org
>>>>
>>>>>>> wrote:
>>>>>>>>
>>>>>>>>> Thanks for the KIP. Great follow up to KIP-1271.
>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> Nit: The `DSL_STORE_TYPE_DOC` should be `private` and I also
>>> think
>>>> we
>>>>>>>>> can improve the wording -- but that's more of am impl detail, not
>>>>>>> really
>>>>>>>>> critical for the KIP itself.
>>>>>>>>>
>>>>>>>>> How about:
>>>>>>>>>
>>>>>>>>> "Controls the state store type used by the DSL store supplier
>>> (see
>>>>>>>>> config 'dsl.store.suppliers.class'). 'default' uses timestamped
>>>> stores.
>>>>>>>>> 'headers' uses the headers-aware stores."
>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> MJS1: Should we deprecate the existing `DslKeyValueParams(final
>>>> String
>>>>>>>>> name, final boolean isTimestamped)` constructor, and make the
>>> newly
>>>>>>>>> introduced "type" a mandatory parameter? (Cf MJS2 below)
>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> MJS2: Should we add new enum `DslStoreType` with DEFAULT and
>>> HEADERS
>>>>>>>>> value, and change the constructor
>>>>>>>>>
>>>>>>>>> from
>>>>>>>>> - DslKeyValueParams(final String name, final boolean
>>>> isTimestamped,
>>>>>>>>> final boolean isHeadersEnabled)
>>>>>>>>>
>>>>>>>>> to
>>>>>>>>> - DslKeyValueParams(final String name, final DslStoreType
>>>> storeType)
>>>>>>>>>
>>>>>>>>> This would more align to the newly added String config (and the
>>> KIP
>>>>>>>>> calls out that we did not want to add a boolean config, what
>>> makes
>>>>>>> sense
>>>>>>>>> to me), and would subsume the existing `isTimestamped` parameter
>>> on
>>>> the
>>>>>>>>> existing constructor which we could remove (Cf MJS1)
>>>>>>>>>
>>>>>>>>> [We have a similar thing for `group.protocol` config.]
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> If yes, wondering if we should also deprecate `isTimestamped()`
>>> and
>>>> not
>>>>>>>>> add `headersEnabled()` but instead add `storeType()` getter?
>>>>>>>>>
>>>>>>>>> Similar for window/session Dsl*Parameter classes.
>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> -Matthias
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> On 2/18/26 3:30 AM, Alieh Saeedi via dev wrote:
>>>>>>>>>> Hi all,
>>>>>>>>>>
>>>>>>>>>> I’d like to start a discussion on a new KIP, KIP-1285: DSL
>>> Opt-in
>>>>>>> Support
>>>>>>>>>> for Header-Aware State Stores.
>>>>>>>>>>
>>>>>>>>>> KIP-1271 introduced headers-aware state stores in Kafka Streams.
>>>>>>> KIP-1285
>>>>>>>>>> proposes adding corresponding DSL-level opt-in support, so that
>>> DSL
>>>>>>>>>> applications can take advantage of these headers-aware stores
>>> while
>>>>>>>>> keeping
>>>>>>>>>> existing applications and topologies unchanged by default.
>>>>>>>>>>
>>>>>>>>>> You can find the KIP here:
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>
>>>>>>>
>>>>
>>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-1285%3A+DSL+Opt-in+Support+for+Header-Aware+State+Stores
>>>>>>>>>>
>>>>>>>>>> Feedback would be greatly appreciated.
>>>>>>>>>>
>>>>>>>>>> Best,
>>>>>>>>>> Alieh
>>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>>
>
Re: [DISCUSS] KIP-1285: DSL Opt-in Support for Header-Aware State Stores
Posted by Alieh Saeedi via dev <de...@kafka.apache.org>.
Hey Bill,
>I'm assuming we are going to add some additional types to the
DslStoreFormat enum to include timestamped types?
Not really. In most cases DEFAULT already maps to a timestamped store. For
session stores, though, DEFAULT means the normal session store because we
don’t have a timestamped variant there. Even today (before this KIP), the
default is timestamped and there’s no way to globally configure the DSL to
use non-timestamped stores instead.
Thanks,
Alieh
On Mon, Feb 23, 2026 at 5:39 PM Bill Bejeck <bb...@gmail.com> wrote:
> Hi Alieh,
>
> Thanks for the KIP!
>
> I agree with everything stated so far.
> In the KIP, we are deprecating the `isTimestamped` field and the
> associated getter, but the `DslStoreFormat` enum only specifies `DEFAULT`
> and `HEADERS`.
> I'm assuming we are going to add some additional types to the
> DslStoreFormat enum to include timestamped types?
>
> Otherwise this looks good to me.
>
> Thanks,
> Bill
>
> On Mon, Feb 23, 2026 at 8:48 AM Alieh Saeedi via dev <de...@kafka.apache.org>
> wrote:
>
>> Thanks Matthias.
>>
>> Good points! I updated the KIP accordingly.
>>
>> Bests
>> Alieh
>>
>> On Mon, Feb 23, 2026 at 11:15 AM Lucas Brutschy via dev <
>> dev@kafka.apache.org> wrote:
>>
>> > Thanks for the updates Alieh!
>> >
>> > On Sun, Feb 22, 2026 at 8:52 PM Matthias J. Sax <mj...@apache.org>
>> wrote:
>> > >
>> > > Thanks Alieh.
>> > >
>> > > I like Lucas' idea about "format" vs "type".
>> > >
>> > >
>> > > For the downgrade path, KIP-1271 says:
>> > >
>> > > > Downgrades are not supported; the only way to achieve one is by
>> > deleting the local store and rebuilding it from the changelog.
>> > >
>> > > While KIP-1285 only say:
>> > >
>> > > > The same as in KIP-1271, downgrades are not supported.
>> > >
>> > > Might be good to align both?
>> > >
>> > >
>> > >
>> > > One last question: Should we call out that
>> > > `Materialized.as(KeyValueBytesStoreSupplier)` (same for window and
>> > > session case), will also allow to enable a headers store on a
>> per-store
>> > > basis, similar to the plain-kv / ts-case? Ie, if one uses
>> > >
>> > > > Materialized.as(
>> > > > Stores.persistentTimestampedKeyValueStoreWithHeaders(...)
>> > > > )
>> > >
>> > > The corresponding DSL operator would also use a header store,
>> > > overwriting whatever is configured via `StreamsConfig`.
>> > >
>> > >
>> > > While it's unfortunately not well documented, it is currently possible
>> > > to pass in both types of supplier, and the DSL would "bridge" between
>> > > both store types. Thus, it's currently possible, to disable ts-stores
>> in
>> > > the DSL via
>> > >
>> > > > Materialized.as(
>> > > > Stores.persistentKeyValueStore(...)
>> > > > )
>> > >
>> > > The determining factor is, if the passed-in supplier implements
>> > > `TimestampedBytesStore` or not and we added a similar marker interface
>> > > `HeaderBytesStore` with KIP-1271, to we can do the same thing for
>> header
>> > > stores.
>> > >
>> > > This should not imply any more code changes I believe, but should work
>> > > this way automatically base on KIP-1285 proposal. I think it's mainly
>> an
>> > > documentation question.
>> > >
>> > >
>> > >
>> > > -Matthias
>> > >
>> > >
>> > >
>> > >
>> > > On 2/22/26 6:16 AM, Alieh Saeedi via dev wrote:
>> > > > Thanks Lucas for the feedback.
>> > > >
>> > > > LB1) dsl.store.type makes sense to me, but I’m not attached to it. I
>> > can
>> > > > change it (and already have changed it) to dsl.store.format if we
>> > prefer
>> > > > that naming.
>> > > >
>> > > > LB2) Since Kafka Streams does not have a timestamped session store,
>> we
>> > > > don’t have any DslSessionParams constructor with isTimestamped.
>> What
>> > we
>> > > > are deprecating in the KIP is the older constructor that does not
>> take
>> > a
>> > > > DslStoreType (now DslStoreFormat). To keep things cleaner, we only
>> > keep a
>> > > > single constructor that also includes DslStoreFormat. For session
>> > stores,
>> > > > DEFAULT corresponds to the regular session store, and HEADERS
>> > corresponds
>> > > > to SessionStoreWithHeaders. This detail is already mentioned in the
>> > KIP.
>> > > >
>> > > > LB3) Because this KIP does not introduce any new state store
>> formats,
>> > the
>> > > > upgrade and downgrade paths are identical to those described in
>> > KIP-1271.
>> > > > I’ve added this note to the KIP as well.
>> > > >
>> > > > Thanks,
>> > > > Alieh
>> > > >
>> > > > On Fri, Feb 20, 2026 at 9:08 PM Lucas Brutschy <
>> lbrutschy@confluent.io
>> > >
>> > > > wrote:
>> > > >
>> > > >> Hi Alieh,
>> > > >>
>> > > >> Thanks for the KIP. A few thoughts below.
>> > > >>
>> > > >> LB1: The config name dsl.store.type feels a bit ambiguous since we
>> > > >> already have dsl.store.suppliers.class (which controls RocksDB vs.
>> > > >> in-memory). dsl.store.type could be confused with that distinction.
>> > > >> Would something like dsl.store.format be clearer, since this is
>> really
>> > > >> about the value format (timestamped vs. timestamped+headers)?
>> > > >>
>> > > >> LB2: Regarding DslSessionParams: the current code entirely lacks an
>> > > >> isTimestamped field or constructor parameter. The KIP shows
>> > > >> deprecating a constructor with isTimestamped on DslSessionParams,
>> but
>> > > >> that constructor does not exist today. Could you clarify whether
>> the
>> > > >> KIP intends to add and immediately deprecate the constructor, or
>> > > >> whether the old DslSessionParams constructor should simply remain
>> > > >> as-is and a new overload with DslStoreType should be added
>> alongside
>> > > >> it?
>> > > >>
>> > > >> LB3: Regarding compatibility, the KIP mentions that switching from
>> > > >> DEFAULT to HEADERS triggers KIP-1271 upgrade paths. It would be
>> good
>> > > >> to clarify whether switching back from HEADERS to DEFAULT is also
>> > > >> supported (i.e., is it a two-way migration or one-way only?). If
>> it is
>> > > >> one-way, that must be documented explicitly.
>> > > >>
>> > > >> Thanks,
>> > > >> Lucas
>> > > >>
>> > > >> On Fri, Feb 20, 2026 at 10:30 AM Alieh Saeedi via dev
>> > > >> <de...@kafka.apache.org> wrote:
>> > > >>>
>> > > >>> Hey Matthias,
>> > > >>> Thanks for the suggestions. I agree with all of them. They all
>> make
>> > sense
>> > > >>> to me. I updated the KIP accordingly.
>> > > >>>
>> > > >>> Bests,
>> > > >>> Alieh
>> > > >>>
>> > > >>> On Thu, Feb 19, 2026 at 8:33 PM Matthias J. Sax <mjsax@apache.org
>> >
>> > > >> wrote:
>> > > >>>
>> > > >>>> Thanks for the KIP. Great follow up to KIP-1271.
>> > > >>>>
>> > > >>>>
>> > > >>>>
>> > > >>>> Nit: The `DSL_STORE_TYPE_DOC` should be `private` and I also
>> think
>> > we
>> > > >>>> can improve the wording -- but that's more of am impl detail, not
>> > > >> really
>> > > >>>> critical for the KIP itself.
>> > > >>>>
>> > > >>>> How about:
>> > > >>>>
>> > > >>>> "Controls the state store type used by the DSL store supplier
>> (see
>> > > >>>> config 'dsl.store.suppliers.class'). 'default' uses timestamped
>> > stores.
>> > > >>>> 'headers' uses the headers-aware stores."
>> > > >>>>
>> > > >>>>
>> > > >>>>
>> > > >>>> MJS1: Should we deprecate the existing `DslKeyValueParams(final
>> > String
>> > > >>>> name, final boolean isTimestamped)` constructor, and make the
>> newly
>> > > >>>> introduced "type" a mandatory parameter? (Cf MJS2 below)
>> > > >>>>
>> > > >>>>
>> > > >>>>
>> > > >>>> MJS2: Should we add new enum `DslStoreType` with DEFAULT and
>> HEADERS
>> > > >>>> value, and change the constructor
>> > > >>>>
>> > > >>>> from
>> > > >>>> - DslKeyValueParams(final String name, final boolean
>> > isTimestamped,
>> > > >>>> final boolean isHeadersEnabled)
>> > > >>>>
>> > > >>>> to
>> > > >>>> - DslKeyValueParams(final String name, final DslStoreType
>> > storeType)
>> > > >>>>
>> > > >>>> This would more align to the newly added String config (and the
>> KIP
>> > > >>>> calls out that we did not want to add a boolean config, what
>> makes
>> > > >> sense
>> > > >>>> to me), and would subsume the existing `isTimestamped` parameter
>> on
>> > the
>> > > >>>> existing constructor which we could remove (Cf MJS1)
>> > > >>>>
>> > > >>>> [We have a similar thing for `group.protocol` config.]
>> > > >>>>
>> > > >>>>
>> > > >>>> If yes, wondering if we should also deprecate `isTimestamped()`
>> and
>> > not
>> > > >>>> add `headersEnabled()` but instead add `storeType()` getter?
>> > > >>>>
>> > > >>>> Similar for window/session Dsl*Parameter classes.
>> > > >>>>
>> > > >>>>
>> > > >>>>
>> > > >>>> -Matthias
>> > > >>>>
>> > > >>>>
>> > > >>>> On 2/18/26 3:30 AM, Alieh Saeedi via dev wrote:
>> > > >>>>> Hi all,
>> > > >>>>>
>> > > >>>>> I’d like to start a discussion on a new KIP, KIP-1285: DSL
>> Opt-in
>> > > >> Support
>> > > >>>>> for Header-Aware State Stores.
>> > > >>>>>
>> > > >>>>> KIP-1271 introduced headers-aware state stores in Kafka Streams.
>> > > >> KIP-1285
>> > > >>>>> proposes adding corresponding DSL-level opt-in support, so that
>> DSL
>> > > >>>>> applications can take advantage of these headers-aware stores
>> while
>> > > >>>> keeping
>> > > >>>>> existing applications and topologies unchanged by default.
>> > > >>>>>
>> > > >>>>> You can find the KIP here:
>> > > >>>>>
>> > > >>>>>
>> > > >>>>
>> > > >>
>> >
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-1285%3A+DSL+Opt-in+Support+for+Header-Aware+State+Stores
>> > > >>>>>
>> > > >>>>> Feedback would be greatly appreciated.
>> > > >>>>>
>> > > >>>>> Best,
>> > > >>>>> Alieh
>> > > >>>>>
>> > > >>>>
>> > > >>>>
>> > > >>
>> > > >
>> > >
>> >
>>
>
Re: [DISCUSS] KIP-1285: DSL Opt-in Support for Header-Aware State Stores
Posted by Bill Bejeck <bb...@gmail.com>.
Hi Alieh,
Thanks for the KIP!
I agree with everything stated so far.
In the KIP, we are deprecating the `isTimestamped` field and the associated
getter, but the `DslStoreFormat` enum only specifies `DEFAULT` and
`HEADERS`.
I'm assuming we are going to add some additional types to the
DslStoreFormat enum to include timestamped types?
Otherwise this looks good to me.
Thanks,
Bill
On Mon, Feb 23, 2026 at 8:48 AM Alieh Saeedi via dev <de...@kafka.apache.org>
wrote:
> Thanks Matthias.
>
> Good points! I updated the KIP accordingly.
>
> Bests
> Alieh
>
> On Mon, Feb 23, 2026 at 11:15 AM Lucas Brutschy via dev <
> dev@kafka.apache.org> wrote:
>
> > Thanks for the updates Alieh!
> >
> > On Sun, Feb 22, 2026 at 8:52 PM Matthias J. Sax <mj...@apache.org>
> wrote:
> > >
> > > Thanks Alieh.
> > >
> > > I like Lucas' idea about "format" vs "type".
> > >
> > >
> > > For the downgrade path, KIP-1271 says:
> > >
> > > > Downgrades are not supported; the only way to achieve one is by
> > deleting the local store and rebuilding it from the changelog.
> > >
> > > While KIP-1285 only say:
> > >
> > > > The same as in KIP-1271, downgrades are not supported.
> > >
> > > Might be good to align both?
> > >
> > >
> > >
> > > One last question: Should we call out that
> > > `Materialized.as(KeyValueBytesStoreSupplier)` (same for window and
> > > session case), will also allow to enable a headers store on a per-store
> > > basis, similar to the plain-kv / ts-case? Ie, if one uses
> > >
> > > > Materialized.as(
> > > > Stores.persistentTimestampedKeyValueStoreWithHeaders(...)
> > > > )
> > >
> > > The corresponding DSL operator would also use a header store,
> > > overwriting whatever is configured via `StreamsConfig`.
> > >
> > >
> > > While it's unfortunately not well documented, it is currently possible
> > > to pass in both types of supplier, and the DSL would "bridge" between
> > > both store types. Thus, it's currently possible, to disable ts-stores
> in
> > > the DSL via
> > >
> > > > Materialized.as(
> > > > Stores.persistentKeyValueStore(...)
> > > > )
> > >
> > > The determining factor is, if the passed-in supplier implements
> > > `TimestampedBytesStore` or not and we added a similar marker interface
> > > `HeaderBytesStore` with KIP-1271, to we can do the same thing for
> header
> > > stores.
> > >
> > > This should not imply any more code changes I believe, but should work
> > > this way automatically base on KIP-1285 proposal. I think it's mainly
> an
> > > documentation question.
> > >
> > >
> > >
> > > -Matthias
> > >
> > >
> > >
> > >
> > > On 2/22/26 6:16 AM, Alieh Saeedi via dev wrote:
> > > > Thanks Lucas for the feedback.
> > > >
> > > > LB1) dsl.store.type makes sense to me, but I’m not attached to it. I
> > can
> > > > change it (and already have changed it) to dsl.store.format if we
> > prefer
> > > > that naming.
> > > >
> > > > LB2) Since Kafka Streams does not have a timestamped session store,
> we
> > > > don’t have any DslSessionParams constructor with isTimestamped. What
> > we
> > > > are deprecating in the KIP is the older constructor that does not
> take
> > a
> > > > DslStoreType (now DslStoreFormat). To keep things cleaner, we only
> > keep a
> > > > single constructor that also includes DslStoreFormat. For session
> > stores,
> > > > DEFAULT corresponds to the regular session store, and HEADERS
> > corresponds
> > > > to SessionStoreWithHeaders. This detail is already mentioned in the
> > KIP.
> > > >
> > > > LB3) Because this KIP does not introduce any new state store formats,
> > the
> > > > upgrade and downgrade paths are identical to those described in
> > KIP-1271.
> > > > I’ve added this note to the KIP as well.
> > > >
> > > > Thanks,
> > > > Alieh
> > > >
> > > > On Fri, Feb 20, 2026 at 9:08 PM Lucas Brutschy <
> lbrutschy@confluent.io
> > >
> > > > wrote:
> > > >
> > > >> Hi Alieh,
> > > >>
> > > >> Thanks for the KIP. A few thoughts below.
> > > >>
> > > >> LB1: The config name dsl.store.type feels a bit ambiguous since we
> > > >> already have dsl.store.suppliers.class (which controls RocksDB vs.
> > > >> in-memory). dsl.store.type could be confused with that distinction.
> > > >> Would something like dsl.store.format be clearer, since this is
> really
> > > >> about the value format (timestamped vs. timestamped+headers)?
> > > >>
> > > >> LB2: Regarding DslSessionParams: the current code entirely lacks an
> > > >> isTimestamped field or constructor parameter. The KIP shows
> > > >> deprecating a constructor with isTimestamped on DslSessionParams,
> but
> > > >> that constructor does not exist today. Could you clarify whether the
> > > >> KIP intends to add and immediately deprecate the constructor, or
> > > >> whether the old DslSessionParams constructor should simply remain
> > > >> as-is and a new overload with DslStoreType should be added alongside
> > > >> it?
> > > >>
> > > >> LB3: Regarding compatibility, the KIP mentions that switching from
> > > >> DEFAULT to HEADERS triggers KIP-1271 upgrade paths. It would be good
> > > >> to clarify whether switching back from HEADERS to DEFAULT is also
> > > >> supported (i.e., is it a two-way migration or one-way only?). If it
> is
> > > >> one-way, that must be documented explicitly.
> > > >>
> > > >> Thanks,
> > > >> Lucas
> > > >>
> > > >> On Fri, Feb 20, 2026 at 10:30 AM Alieh Saeedi via dev
> > > >> <de...@kafka.apache.org> wrote:
> > > >>>
> > > >>> Hey Matthias,
> > > >>> Thanks for the suggestions. I agree with all of them. They all make
> > sense
> > > >>> to me. I updated the KIP accordingly.
> > > >>>
> > > >>> Bests,
> > > >>> Alieh
> > > >>>
> > > >>> On Thu, Feb 19, 2026 at 8:33 PM Matthias J. Sax <mj...@apache.org>
> > > >> wrote:
> > > >>>
> > > >>>> Thanks for the KIP. Great follow up to KIP-1271.
> > > >>>>
> > > >>>>
> > > >>>>
> > > >>>> Nit: The `DSL_STORE_TYPE_DOC` should be `private` and I also think
> > we
> > > >>>> can improve the wording -- but that's more of am impl detail, not
> > > >> really
> > > >>>> critical for the KIP itself.
> > > >>>>
> > > >>>> How about:
> > > >>>>
> > > >>>> "Controls the state store type used by the DSL store supplier (see
> > > >>>> config 'dsl.store.suppliers.class'). 'default' uses timestamped
> > stores.
> > > >>>> 'headers' uses the headers-aware stores."
> > > >>>>
> > > >>>>
> > > >>>>
> > > >>>> MJS1: Should we deprecate the existing `DslKeyValueParams(final
> > String
> > > >>>> name, final boolean isTimestamped)` constructor, and make the
> newly
> > > >>>> introduced "type" a mandatory parameter? (Cf MJS2 below)
> > > >>>>
> > > >>>>
> > > >>>>
> > > >>>> MJS2: Should we add new enum `DslStoreType` with DEFAULT and
> HEADERS
> > > >>>> value, and change the constructor
> > > >>>>
> > > >>>> from
> > > >>>> - DslKeyValueParams(final String name, final boolean
> > isTimestamped,
> > > >>>> final boolean isHeadersEnabled)
> > > >>>>
> > > >>>> to
> > > >>>> - DslKeyValueParams(final String name, final DslStoreType
> > storeType)
> > > >>>>
> > > >>>> This would more align to the newly added String config (and the
> KIP
> > > >>>> calls out that we did not want to add a boolean config, what makes
> > > >> sense
> > > >>>> to me), and would subsume the existing `isTimestamped` parameter
> on
> > the
> > > >>>> existing constructor which we could remove (Cf MJS1)
> > > >>>>
> > > >>>> [We have a similar thing for `group.protocol` config.]
> > > >>>>
> > > >>>>
> > > >>>> If yes, wondering if we should also deprecate `isTimestamped()`
> and
> > not
> > > >>>> add `headersEnabled()` but instead add `storeType()` getter?
> > > >>>>
> > > >>>> Similar for window/session Dsl*Parameter classes.
> > > >>>>
> > > >>>>
> > > >>>>
> > > >>>> -Matthias
> > > >>>>
> > > >>>>
> > > >>>> On 2/18/26 3:30 AM, Alieh Saeedi via dev wrote:
> > > >>>>> Hi all,
> > > >>>>>
> > > >>>>> I’d like to start a discussion on a new KIP, KIP-1285: DSL Opt-in
> > > >> Support
> > > >>>>> for Header-Aware State Stores.
> > > >>>>>
> > > >>>>> KIP-1271 introduced headers-aware state stores in Kafka Streams.
> > > >> KIP-1285
> > > >>>>> proposes adding corresponding DSL-level opt-in support, so that
> DSL
> > > >>>>> applications can take advantage of these headers-aware stores
> while
> > > >>>> keeping
> > > >>>>> existing applications and topologies unchanged by default.
> > > >>>>>
> > > >>>>> You can find the KIP here:
> > > >>>>>
> > > >>>>>
> > > >>>>
> > > >>
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-1285%3A+DSL+Opt-in+Support+for+Header-Aware+State+Stores
> > > >>>>>
> > > >>>>> Feedback would be greatly appreciated.
> > > >>>>>
> > > >>>>> Best,
> > > >>>>> Alieh
> > > >>>>>
> > > >>>>
> > > >>>>
> > > >>
> > > >
> > >
> >
>
Re: [DISCUSS] KIP-1285: DSL Opt-in Support for Header-Aware State Stores
Posted by Alieh Saeedi via dev <de...@kafka.apache.org>.
Thanks Matthias.
Good points! I updated the KIP accordingly.
Bests
Alieh
On Mon, Feb 23, 2026 at 11:15 AM Lucas Brutschy via dev <
dev@kafka.apache.org> wrote:
> Thanks for the updates Alieh!
>
> On Sun, Feb 22, 2026 at 8:52 PM Matthias J. Sax <mj...@apache.org> wrote:
> >
> > Thanks Alieh.
> >
> > I like Lucas' idea about "format" vs "type".
> >
> >
> > For the downgrade path, KIP-1271 says:
> >
> > > Downgrades are not supported; the only way to achieve one is by
> deleting the local store and rebuilding it from the changelog.
> >
> > While KIP-1285 only say:
> >
> > > The same as in KIP-1271, downgrades are not supported.
> >
> > Might be good to align both?
> >
> >
> >
> > One last question: Should we call out that
> > `Materialized.as(KeyValueBytesStoreSupplier)` (same for window and
> > session case), will also allow to enable a headers store on a per-store
> > basis, similar to the plain-kv / ts-case? Ie, if one uses
> >
> > > Materialized.as(
> > > Stores.persistentTimestampedKeyValueStoreWithHeaders(...)
> > > )
> >
> > The corresponding DSL operator would also use a header store,
> > overwriting whatever is configured via `StreamsConfig`.
> >
> >
> > While it's unfortunately not well documented, it is currently possible
> > to pass in both types of supplier, and the DSL would "bridge" between
> > both store types. Thus, it's currently possible, to disable ts-stores in
> > the DSL via
> >
> > > Materialized.as(
> > > Stores.persistentKeyValueStore(...)
> > > )
> >
> > The determining factor is, if the passed-in supplier implements
> > `TimestampedBytesStore` or not and we added a similar marker interface
> > `HeaderBytesStore` with KIP-1271, to we can do the same thing for header
> > stores.
> >
> > This should not imply any more code changes I believe, but should work
> > this way automatically base on KIP-1285 proposal. I think it's mainly an
> > documentation question.
> >
> >
> >
> > -Matthias
> >
> >
> >
> >
> > On 2/22/26 6:16 AM, Alieh Saeedi via dev wrote:
> > > Thanks Lucas for the feedback.
> > >
> > > LB1) dsl.store.type makes sense to me, but I’m not attached to it. I
> can
> > > change it (and already have changed it) to dsl.store.format if we
> prefer
> > > that naming.
> > >
> > > LB2) Since Kafka Streams does not have a timestamped session store, we
> > > don’t have any DslSessionParams constructor with isTimestamped. What
> we
> > > are deprecating in the KIP is the older constructor that does not take
> a
> > > DslStoreType (now DslStoreFormat). To keep things cleaner, we only
> keep a
> > > single constructor that also includes DslStoreFormat. For session
> stores,
> > > DEFAULT corresponds to the regular session store, and HEADERS
> corresponds
> > > to SessionStoreWithHeaders. This detail is already mentioned in the
> KIP.
> > >
> > > LB3) Because this KIP does not introduce any new state store formats,
> the
> > > upgrade and downgrade paths are identical to those described in
> KIP-1271.
> > > I’ve added this note to the KIP as well.
> > >
> > > Thanks,
> > > Alieh
> > >
> > > On Fri, Feb 20, 2026 at 9:08 PM Lucas Brutschy <lbrutschy@confluent.io
> >
> > > wrote:
> > >
> > >> Hi Alieh,
> > >>
> > >> Thanks for the KIP. A few thoughts below.
> > >>
> > >> LB1: The config name dsl.store.type feels a bit ambiguous since we
> > >> already have dsl.store.suppliers.class (which controls RocksDB vs.
> > >> in-memory). dsl.store.type could be confused with that distinction.
> > >> Would something like dsl.store.format be clearer, since this is really
> > >> about the value format (timestamped vs. timestamped+headers)?
> > >>
> > >> LB2: Regarding DslSessionParams: the current code entirely lacks an
> > >> isTimestamped field or constructor parameter. The KIP shows
> > >> deprecating a constructor with isTimestamped on DslSessionParams, but
> > >> that constructor does not exist today. Could you clarify whether the
> > >> KIP intends to add and immediately deprecate the constructor, or
> > >> whether the old DslSessionParams constructor should simply remain
> > >> as-is and a new overload with DslStoreType should be added alongside
> > >> it?
> > >>
> > >> LB3: Regarding compatibility, the KIP mentions that switching from
> > >> DEFAULT to HEADERS triggers KIP-1271 upgrade paths. It would be good
> > >> to clarify whether switching back from HEADERS to DEFAULT is also
> > >> supported (i.e., is it a two-way migration or one-way only?). If it is
> > >> one-way, that must be documented explicitly.
> > >>
> > >> Thanks,
> > >> Lucas
> > >>
> > >> On Fri, Feb 20, 2026 at 10:30 AM Alieh Saeedi via dev
> > >> <de...@kafka.apache.org> wrote:
> > >>>
> > >>> Hey Matthias,
> > >>> Thanks for the suggestions. I agree with all of them. They all make
> sense
> > >>> to me. I updated the KIP accordingly.
> > >>>
> > >>> Bests,
> > >>> Alieh
> > >>>
> > >>> On Thu, Feb 19, 2026 at 8:33 PM Matthias J. Sax <mj...@apache.org>
> > >> wrote:
> > >>>
> > >>>> Thanks for the KIP. Great follow up to KIP-1271.
> > >>>>
> > >>>>
> > >>>>
> > >>>> Nit: The `DSL_STORE_TYPE_DOC` should be `private` and I also think
> we
> > >>>> can improve the wording -- but that's more of am impl detail, not
> > >> really
> > >>>> critical for the KIP itself.
> > >>>>
> > >>>> How about:
> > >>>>
> > >>>> "Controls the state store type used by the DSL store supplier (see
> > >>>> config 'dsl.store.suppliers.class'). 'default' uses timestamped
> stores.
> > >>>> 'headers' uses the headers-aware stores."
> > >>>>
> > >>>>
> > >>>>
> > >>>> MJS1: Should we deprecate the existing `DslKeyValueParams(final
> String
> > >>>> name, final boolean isTimestamped)` constructor, and make the newly
> > >>>> introduced "type" a mandatory parameter? (Cf MJS2 below)
> > >>>>
> > >>>>
> > >>>>
> > >>>> MJS2: Should we add new enum `DslStoreType` with DEFAULT and HEADERS
> > >>>> value, and change the constructor
> > >>>>
> > >>>> from
> > >>>> - DslKeyValueParams(final String name, final boolean
> isTimestamped,
> > >>>> final boolean isHeadersEnabled)
> > >>>>
> > >>>> to
> > >>>> - DslKeyValueParams(final String name, final DslStoreType
> storeType)
> > >>>>
> > >>>> This would more align to the newly added String config (and the KIP
> > >>>> calls out that we did not want to add a boolean config, what makes
> > >> sense
> > >>>> to me), and would subsume the existing `isTimestamped` parameter on
> the
> > >>>> existing constructor which we could remove (Cf MJS1)
> > >>>>
> > >>>> [We have a similar thing for `group.protocol` config.]
> > >>>>
> > >>>>
> > >>>> If yes, wondering if we should also deprecate `isTimestamped()` and
> not
> > >>>> add `headersEnabled()` but instead add `storeType()` getter?
> > >>>>
> > >>>> Similar for window/session Dsl*Parameter classes.
> > >>>>
> > >>>>
> > >>>>
> > >>>> -Matthias
> > >>>>
> > >>>>
> > >>>> On 2/18/26 3:30 AM, Alieh Saeedi via dev wrote:
> > >>>>> Hi all,
> > >>>>>
> > >>>>> I’d like to start a discussion on a new KIP, KIP-1285: DSL Opt-in
> > >> Support
> > >>>>> for Header-Aware State Stores.
> > >>>>>
> > >>>>> KIP-1271 introduced headers-aware state stores in Kafka Streams.
> > >> KIP-1285
> > >>>>> proposes adding corresponding DSL-level opt-in support, so that DSL
> > >>>>> applications can take advantage of these headers-aware stores while
> > >>>> keeping
> > >>>>> existing applications and topologies unchanged by default.
> > >>>>>
> > >>>>> You can find the KIP here:
> > >>>>>
> > >>>>>
> > >>>>
> > >>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-1285%3A+DSL+Opt-in+Support+for+Header-Aware+State+Stores
> > >>>>>
> > >>>>> Feedback would be greatly appreciated.
> > >>>>>
> > >>>>> Best,
> > >>>>> Alieh
> > >>>>>
> > >>>>
> > >>>>
> > >>
> > >
> >
>
Re: [DISCUSS] KIP-1285: DSL Opt-in Support for Header-Aware State Stores
Posted by Lucas Brutschy via dev <de...@kafka.apache.org>.
Thanks for the updates Alieh!
On Sun, Feb 22, 2026 at 8:52 PM Matthias J. Sax <mj...@apache.org> wrote:
>
> Thanks Alieh.
>
> I like Lucas' idea about "format" vs "type".
>
>
> For the downgrade path, KIP-1271 says:
>
> > Downgrades are not supported; the only way to achieve one is by deleting the local store and rebuilding it from the changelog.
>
> While KIP-1285 only say:
>
> > The same as in KIP-1271, downgrades are not supported.
>
> Might be good to align both?
>
>
>
> One last question: Should we call out that
> `Materialized.as(KeyValueBytesStoreSupplier)` (same for window and
> session case), will also allow to enable a headers store on a per-store
> basis, similar to the plain-kv / ts-case? Ie, if one uses
>
> > Materialized.as(
> > Stores.persistentTimestampedKeyValueStoreWithHeaders(...)
> > )
>
> The corresponding DSL operator would also use a header store,
> overwriting whatever is configured via `StreamsConfig`.
>
>
> While it's unfortunately not well documented, it is currently possible
> to pass in both types of supplier, and the DSL would "bridge" between
> both store types. Thus, it's currently possible, to disable ts-stores in
> the DSL via
>
> > Materialized.as(
> > Stores.persistentKeyValueStore(...)
> > )
>
> The determining factor is, if the passed-in supplier implements
> `TimestampedBytesStore` or not and we added a similar marker interface
> `HeaderBytesStore` with KIP-1271, to we can do the same thing for header
> stores.
>
> This should not imply any more code changes I believe, but should work
> this way automatically base on KIP-1285 proposal. I think it's mainly an
> documentation question.
>
>
>
> -Matthias
>
>
>
>
> On 2/22/26 6:16 AM, Alieh Saeedi via dev wrote:
> > Thanks Lucas for the feedback.
> >
> > LB1) dsl.store.type makes sense to me, but I’m not attached to it. I can
> > change it (and already have changed it) to dsl.store.format if we prefer
> > that naming.
> >
> > LB2) Since Kafka Streams does not have a timestamped session store, we
> > don’t have any DslSessionParams constructor with isTimestamped. What we
> > are deprecating in the KIP is the older constructor that does not take a
> > DslStoreType (now DslStoreFormat). To keep things cleaner, we only keep a
> > single constructor that also includes DslStoreFormat. For session stores,
> > DEFAULT corresponds to the regular session store, and HEADERS corresponds
> > to SessionStoreWithHeaders. This detail is already mentioned in the KIP.
> >
> > LB3) Because this KIP does not introduce any new state store formats, the
> > upgrade and downgrade paths are identical to those described in KIP-1271.
> > I’ve added this note to the KIP as well.
> >
> > Thanks,
> > Alieh
> >
> > On Fri, Feb 20, 2026 at 9:08 PM Lucas Brutschy <lb...@confluent.io>
> > wrote:
> >
> >> Hi Alieh,
> >>
> >> Thanks for the KIP. A few thoughts below.
> >>
> >> LB1: The config name dsl.store.type feels a bit ambiguous since we
> >> already have dsl.store.suppliers.class (which controls RocksDB vs.
> >> in-memory). dsl.store.type could be confused with that distinction.
> >> Would something like dsl.store.format be clearer, since this is really
> >> about the value format (timestamped vs. timestamped+headers)?
> >>
> >> LB2: Regarding DslSessionParams: the current code entirely lacks an
> >> isTimestamped field or constructor parameter. The KIP shows
> >> deprecating a constructor with isTimestamped on DslSessionParams, but
> >> that constructor does not exist today. Could you clarify whether the
> >> KIP intends to add and immediately deprecate the constructor, or
> >> whether the old DslSessionParams constructor should simply remain
> >> as-is and a new overload with DslStoreType should be added alongside
> >> it?
> >>
> >> LB3: Regarding compatibility, the KIP mentions that switching from
> >> DEFAULT to HEADERS triggers KIP-1271 upgrade paths. It would be good
> >> to clarify whether switching back from HEADERS to DEFAULT is also
> >> supported (i.e., is it a two-way migration or one-way only?). If it is
> >> one-way, that must be documented explicitly.
> >>
> >> Thanks,
> >> Lucas
> >>
> >> On Fri, Feb 20, 2026 at 10:30 AM Alieh Saeedi via dev
> >> <de...@kafka.apache.org> wrote:
> >>>
> >>> Hey Matthias,
> >>> Thanks for the suggestions. I agree with all of them. They all make sense
> >>> to me. I updated the KIP accordingly.
> >>>
> >>> Bests,
> >>> Alieh
> >>>
> >>> On Thu, Feb 19, 2026 at 8:33 PM Matthias J. Sax <mj...@apache.org>
> >> wrote:
> >>>
> >>>> Thanks for the KIP. Great follow up to KIP-1271.
> >>>>
> >>>>
> >>>>
> >>>> Nit: The `DSL_STORE_TYPE_DOC` should be `private` and I also think we
> >>>> can improve the wording -- but that's more of am impl detail, not
> >> really
> >>>> critical for the KIP itself.
> >>>>
> >>>> How about:
> >>>>
> >>>> "Controls the state store type used by the DSL store supplier (see
> >>>> config 'dsl.store.suppliers.class'). 'default' uses timestamped stores.
> >>>> 'headers' uses the headers-aware stores."
> >>>>
> >>>>
> >>>>
> >>>> MJS1: Should we deprecate the existing `DslKeyValueParams(final String
> >>>> name, final boolean isTimestamped)` constructor, and make the newly
> >>>> introduced "type" a mandatory parameter? (Cf MJS2 below)
> >>>>
> >>>>
> >>>>
> >>>> MJS2: Should we add new enum `DslStoreType` with DEFAULT and HEADERS
> >>>> value, and change the constructor
> >>>>
> >>>> from
> >>>> - DslKeyValueParams(final String name, final boolean isTimestamped,
> >>>> final boolean isHeadersEnabled)
> >>>>
> >>>> to
> >>>> - DslKeyValueParams(final String name, final DslStoreType storeType)
> >>>>
> >>>> This would more align to the newly added String config (and the KIP
> >>>> calls out that we did not want to add a boolean config, what makes
> >> sense
> >>>> to me), and would subsume the existing `isTimestamped` parameter on the
> >>>> existing constructor which we could remove (Cf MJS1)
> >>>>
> >>>> [We have a similar thing for `group.protocol` config.]
> >>>>
> >>>>
> >>>> If yes, wondering if we should also deprecate `isTimestamped()` and not
> >>>> add `headersEnabled()` but instead add `storeType()` getter?
> >>>>
> >>>> Similar for window/session Dsl*Parameter classes.
> >>>>
> >>>>
> >>>>
> >>>> -Matthias
> >>>>
> >>>>
> >>>> On 2/18/26 3:30 AM, Alieh Saeedi via dev wrote:
> >>>>> Hi all,
> >>>>>
> >>>>> I’d like to start a discussion on a new KIP, KIP-1285: DSL Opt-in
> >> Support
> >>>>> for Header-Aware State Stores.
> >>>>>
> >>>>> KIP-1271 introduced headers-aware state stores in Kafka Streams.
> >> KIP-1285
> >>>>> proposes adding corresponding DSL-level opt-in support, so that DSL
> >>>>> applications can take advantage of these headers-aware stores while
> >>>> keeping
> >>>>> existing applications and topologies unchanged by default.
> >>>>>
> >>>>> You can find the KIP here:
> >>>>>
> >>>>>
> >>>>
> >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-1285%3A+DSL+Opt-in+Support+for+Header-Aware+State+Stores
> >>>>>
> >>>>> Feedback would be greatly appreciated.
> >>>>>
> >>>>> Best,
> >>>>> Alieh
> >>>>>
> >>>>
> >>>>
> >>
> >
>
Re: [DISCUSS] KIP-1285: DSL Opt-in Support for Header-Aware State Stores
Posted by "Matthias J. Sax" <mj...@apache.org>.
Thanks Alieh.
I like Lucas' idea about "format" vs "type".
For the downgrade path, KIP-1271 says:
> Downgrades are not supported; the only way to achieve one is by deleting the local store and rebuilding it from the changelog.
While KIP-1285 only say:
> The same as in KIP-1271, downgrades are not supported.
Might be good to align both?
One last question: Should we call out that
`Materialized.as(KeyValueBytesStoreSupplier)` (same for window and
session case), will also allow to enable a headers store on a per-store
basis, similar to the plain-kv / ts-case? Ie, if one uses
> Materialized.as(
> Stores.persistentTimestampedKeyValueStoreWithHeaders(...)
> )
The corresponding DSL operator would also use a header store,
overwriting whatever is configured via `StreamsConfig`.
While it's unfortunately not well documented, it is currently possible
to pass in both types of supplier, and the DSL would "bridge" between
both store types. Thus, it's currently possible, to disable ts-stores in
the DSL via
> Materialized.as(
> Stores.persistentKeyValueStore(...)
> )
The determining factor is, if the passed-in supplier implements
`TimestampedBytesStore` or not and we added a similar marker interface
`HeaderBytesStore` with KIP-1271, to we can do the same thing for header
stores.
This should not imply any more code changes I believe, but should work
this way automatically base on KIP-1285 proposal. I think it's mainly an
documentation question.
-Matthias
On 2/22/26 6:16 AM, Alieh Saeedi via dev wrote:
> Thanks Lucas for the feedback.
>
> LB1) dsl.store.type makes sense to me, but I’m not attached to it. I can
> change it (and already have changed it) to dsl.store.format if we prefer
> that naming.
>
> LB2) Since Kafka Streams does not have a timestamped session store, we
> don’t have any DslSessionParams constructor with isTimestamped. What we
> are deprecating in the KIP is the older constructor that does not take a
> DslStoreType (now DslStoreFormat). To keep things cleaner, we only keep a
> single constructor that also includes DslStoreFormat. For session stores,
> DEFAULT corresponds to the regular session store, and HEADERS corresponds
> to SessionStoreWithHeaders. This detail is already mentioned in the KIP.
>
> LB3) Because this KIP does not introduce any new state store formats, the
> upgrade and downgrade paths are identical to those described in KIP-1271.
> I’ve added this note to the KIP as well.
>
> Thanks,
> Alieh
>
> On Fri, Feb 20, 2026 at 9:08 PM Lucas Brutschy <lb...@confluent.io>
> wrote:
>
>> Hi Alieh,
>>
>> Thanks for the KIP. A few thoughts below.
>>
>> LB1: The config name dsl.store.type feels a bit ambiguous since we
>> already have dsl.store.suppliers.class (which controls RocksDB vs.
>> in-memory). dsl.store.type could be confused with that distinction.
>> Would something like dsl.store.format be clearer, since this is really
>> about the value format (timestamped vs. timestamped+headers)?
>>
>> LB2: Regarding DslSessionParams: the current code entirely lacks an
>> isTimestamped field or constructor parameter. The KIP shows
>> deprecating a constructor with isTimestamped on DslSessionParams, but
>> that constructor does not exist today. Could you clarify whether the
>> KIP intends to add and immediately deprecate the constructor, or
>> whether the old DslSessionParams constructor should simply remain
>> as-is and a new overload with DslStoreType should be added alongside
>> it?
>>
>> LB3: Regarding compatibility, the KIP mentions that switching from
>> DEFAULT to HEADERS triggers KIP-1271 upgrade paths. It would be good
>> to clarify whether switching back from HEADERS to DEFAULT is also
>> supported (i.e., is it a two-way migration or one-way only?). If it is
>> one-way, that must be documented explicitly.
>>
>> Thanks,
>> Lucas
>>
>> On Fri, Feb 20, 2026 at 10:30 AM Alieh Saeedi via dev
>> <de...@kafka.apache.org> wrote:
>>>
>>> Hey Matthias,
>>> Thanks for the suggestions. I agree with all of them. They all make sense
>>> to me. I updated the KIP accordingly.
>>>
>>> Bests,
>>> Alieh
>>>
>>> On Thu, Feb 19, 2026 at 8:33 PM Matthias J. Sax <mj...@apache.org>
>> wrote:
>>>
>>>> Thanks for the KIP. Great follow up to KIP-1271.
>>>>
>>>>
>>>>
>>>> Nit: The `DSL_STORE_TYPE_DOC` should be `private` and I also think we
>>>> can improve the wording -- but that's more of am impl detail, not
>> really
>>>> critical for the KIP itself.
>>>>
>>>> How about:
>>>>
>>>> "Controls the state store type used by the DSL store supplier (see
>>>> config 'dsl.store.suppliers.class'). 'default' uses timestamped stores.
>>>> 'headers' uses the headers-aware stores."
>>>>
>>>>
>>>>
>>>> MJS1: Should we deprecate the existing `DslKeyValueParams(final String
>>>> name, final boolean isTimestamped)` constructor, and make the newly
>>>> introduced "type" a mandatory parameter? (Cf MJS2 below)
>>>>
>>>>
>>>>
>>>> MJS2: Should we add new enum `DslStoreType` with DEFAULT and HEADERS
>>>> value, and change the constructor
>>>>
>>>> from
>>>> - DslKeyValueParams(final String name, final boolean isTimestamped,
>>>> final boolean isHeadersEnabled)
>>>>
>>>> to
>>>> - DslKeyValueParams(final String name, final DslStoreType storeType)
>>>>
>>>> This would more align to the newly added String config (and the KIP
>>>> calls out that we did not want to add a boolean config, what makes
>> sense
>>>> to me), and would subsume the existing `isTimestamped` parameter on the
>>>> existing constructor which we could remove (Cf MJS1)
>>>>
>>>> [We have a similar thing for `group.protocol` config.]
>>>>
>>>>
>>>> If yes, wondering if we should also deprecate `isTimestamped()` and not
>>>> add `headersEnabled()` but instead add `storeType()` getter?
>>>>
>>>> Similar for window/session Dsl*Parameter classes.
>>>>
>>>>
>>>>
>>>> -Matthias
>>>>
>>>>
>>>> On 2/18/26 3:30 AM, Alieh Saeedi via dev wrote:
>>>>> Hi all,
>>>>>
>>>>> I’d like to start a discussion on a new KIP, KIP-1285: DSL Opt-in
>> Support
>>>>> for Header-Aware State Stores.
>>>>>
>>>>> KIP-1271 introduced headers-aware state stores in Kafka Streams.
>> KIP-1285
>>>>> proposes adding corresponding DSL-level opt-in support, so that DSL
>>>>> applications can take advantage of these headers-aware stores while
>>>> keeping
>>>>> existing applications and topologies unchanged by default.
>>>>>
>>>>> You can find the KIP here:
>>>>>
>>>>>
>>>>
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-1285%3A+DSL+Opt-in+Support+for+Header-Aware+State+Stores
>>>>>
>>>>> Feedback would be greatly appreciated.
>>>>>
>>>>> Best,
>>>>> Alieh
>>>>>
>>>>
>>>>
>>
>
Re: [DISCUSS] KIP-1285: DSL Opt-in Support for Header-Aware State Stores
Posted by Alieh Saeedi via dev <de...@kafka.apache.org>.
Thanks Lucas for the feedback.
LB1) dsl.store.type makes sense to me, but I’m not attached to it. I can
change it (and already have changed it) to dsl.store.format if we prefer
that naming.
LB2) Since Kafka Streams does not have a timestamped session store, we
don’t have any DslSessionParams constructor with isTimestamped. What we
are deprecating in the KIP is the older constructor that does not take a
DslStoreType (now DslStoreFormat). To keep things cleaner, we only keep a
single constructor that also includes DslStoreFormat. For session stores,
DEFAULT corresponds to the regular session store, and HEADERS corresponds
to SessionStoreWithHeaders. This detail is already mentioned in the KIP.
LB3) Because this KIP does not introduce any new state store formats, the
upgrade and downgrade paths are identical to those described in KIP-1271.
I’ve added this note to the KIP as well.
Thanks,
Alieh
On Fri, Feb 20, 2026 at 9:08 PM Lucas Brutschy <lb...@confluent.io>
wrote:
> Hi Alieh,
>
> Thanks for the KIP. A few thoughts below.
>
> LB1: The config name dsl.store.type feels a bit ambiguous since we
> already have dsl.store.suppliers.class (which controls RocksDB vs.
> in-memory). dsl.store.type could be confused with that distinction.
> Would something like dsl.store.format be clearer, since this is really
> about the value format (timestamped vs. timestamped+headers)?
>
> LB2: Regarding DslSessionParams: the current code entirely lacks an
> isTimestamped field or constructor parameter. The KIP shows
> deprecating a constructor with isTimestamped on DslSessionParams, but
> that constructor does not exist today. Could you clarify whether the
> KIP intends to add and immediately deprecate the constructor, or
> whether the old DslSessionParams constructor should simply remain
> as-is and a new overload with DslStoreType should be added alongside
> it?
>
> LB3: Regarding compatibility, the KIP mentions that switching from
> DEFAULT to HEADERS triggers KIP-1271 upgrade paths. It would be good
> to clarify whether switching back from HEADERS to DEFAULT is also
> supported (i.e., is it a two-way migration or one-way only?). If it is
> one-way, that must be documented explicitly.
>
> Thanks,
> Lucas
>
> On Fri, Feb 20, 2026 at 10:30 AM Alieh Saeedi via dev
> <de...@kafka.apache.org> wrote:
> >
> > Hey Matthias,
> > Thanks for the suggestions. I agree with all of them. They all make sense
> > to me. I updated the KIP accordingly.
> >
> > Bests,
> > Alieh
> >
> > On Thu, Feb 19, 2026 at 8:33 PM Matthias J. Sax <mj...@apache.org>
> wrote:
> >
> > > Thanks for the KIP. Great follow up to KIP-1271.
> > >
> > >
> > >
> > > Nit: The `DSL_STORE_TYPE_DOC` should be `private` and I also think we
> > > can improve the wording -- but that's more of am impl detail, not
> really
> > > critical for the KIP itself.
> > >
> > > How about:
> > >
> > > "Controls the state store type used by the DSL store supplier (see
> > > config 'dsl.store.suppliers.class'). 'default' uses timestamped stores.
> > > 'headers' uses the headers-aware stores."
> > >
> > >
> > >
> > > MJS1: Should we deprecate the existing `DslKeyValueParams(final String
> > > name, final boolean isTimestamped)` constructor, and make the newly
> > > introduced "type" a mandatory parameter? (Cf MJS2 below)
> > >
> > >
> > >
> > > MJS2: Should we add new enum `DslStoreType` with DEFAULT and HEADERS
> > > value, and change the constructor
> > >
> > > from
> > > - DslKeyValueParams(final String name, final boolean isTimestamped,
> > > final boolean isHeadersEnabled)
> > >
> > > to
> > > - DslKeyValueParams(final String name, final DslStoreType storeType)
> > >
> > > This would more align to the newly added String config (and the KIP
> > > calls out that we did not want to add a boolean config, what makes
> sense
> > > to me), and would subsume the existing `isTimestamped` parameter on the
> > > existing constructor which we could remove (Cf MJS1)
> > >
> > > [We have a similar thing for `group.protocol` config.]
> > >
> > >
> > > If yes, wondering if we should also deprecate `isTimestamped()` and not
> > > add `headersEnabled()` but instead add `storeType()` getter?
> > >
> > > Similar for window/session Dsl*Parameter classes.
> > >
> > >
> > >
> > > -Matthias
> > >
> > >
> > > On 2/18/26 3:30 AM, Alieh Saeedi via dev wrote:
> > > > Hi all,
> > > >
> > > > I’d like to start a discussion on a new KIP, KIP-1285: DSL Opt-in
> Support
> > > > for Header-Aware State Stores.
> > > >
> > > > KIP-1271 introduced headers-aware state stores in Kafka Streams.
> KIP-1285
> > > > proposes adding corresponding DSL-level opt-in support, so that DSL
> > > > applications can take advantage of these headers-aware stores while
> > > keeping
> > > > existing applications and topologies unchanged by default.
> > > >
> > > > You can find the KIP here:
> > > >
> > > >
> > >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-1285%3A+DSL+Opt-in+Support+for+Header-Aware+State+Stores
> > > >
> > > > Feedback would be greatly appreciated.
> > > >
> > > > Best,
> > > > Alieh
> > > >
> > >
> > >
>
Re: [DISCUSS] KIP-1285: DSL Opt-in Support for Header-Aware State Stores
Posted by Lucas Brutschy via dev <de...@kafka.apache.org>.
Hi Alieh,
Thanks for the KIP. A few thoughts below.
LB1: The config name dsl.store.type feels a bit ambiguous since we
already have dsl.store.suppliers.class (which controls RocksDB vs.
in-memory). dsl.store.type could be confused with that distinction.
Would something like dsl.store.format be clearer, since this is really
about the value format (timestamped vs. timestamped+headers)?
LB2: Regarding DslSessionParams: the current code entirely lacks an
isTimestamped field or constructor parameter. The KIP shows
deprecating a constructor with isTimestamped on DslSessionParams, but
that constructor does not exist today. Could you clarify whether the
KIP intends to add and immediately deprecate the constructor, or
whether the old DslSessionParams constructor should simply remain
as-is and a new overload with DslStoreType should be added alongside
it?
LB3: Regarding compatibility, the KIP mentions that switching from
DEFAULT to HEADERS triggers KIP-1271 upgrade paths. It would be good
to clarify whether switching back from HEADERS to DEFAULT is also
supported (i.e., is it a two-way migration or one-way only?). If it is
one-way, that must be documented explicitly.
Thanks,
Lucas
On Fri, Feb 20, 2026 at 10:30 AM Alieh Saeedi via dev
<de...@kafka.apache.org> wrote:
>
> Hey Matthias,
> Thanks for the suggestions. I agree with all of them. They all make sense
> to me. I updated the KIP accordingly.
>
> Bests,
> Alieh
>
> On Thu, Feb 19, 2026 at 8:33 PM Matthias J. Sax <mj...@apache.org> wrote:
>
> > Thanks for the KIP. Great follow up to KIP-1271.
> >
> >
> >
> > Nit: The `DSL_STORE_TYPE_DOC` should be `private` and I also think we
> > can improve the wording -- but that's more of am impl detail, not really
> > critical for the KIP itself.
> >
> > How about:
> >
> > "Controls the state store type used by the DSL store supplier (see
> > config 'dsl.store.suppliers.class'). 'default' uses timestamped stores.
> > 'headers' uses the headers-aware stores."
> >
> >
> >
> > MJS1: Should we deprecate the existing `DslKeyValueParams(final String
> > name, final boolean isTimestamped)` constructor, and make the newly
> > introduced "type" a mandatory parameter? (Cf MJS2 below)
> >
> >
> >
> > MJS2: Should we add new enum `DslStoreType` with DEFAULT and HEADERS
> > value, and change the constructor
> >
> > from
> > - DslKeyValueParams(final String name, final boolean isTimestamped,
> > final boolean isHeadersEnabled)
> >
> > to
> > - DslKeyValueParams(final String name, final DslStoreType storeType)
> >
> > This would more align to the newly added String config (and the KIP
> > calls out that we did not want to add a boolean config, what makes sense
> > to me), and would subsume the existing `isTimestamped` parameter on the
> > existing constructor which we could remove (Cf MJS1)
> >
> > [We have a similar thing for `group.protocol` config.]
> >
> >
> > If yes, wondering if we should also deprecate `isTimestamped()` and not
> > add `headersEnabled()` but instead add `storeType()` getter?
> >
> > Similar for window/session Dsl*Parameter classes.
> >
> >
> >
> > -Matthias
> >
> >
> > On 2/18/26 3:30 AM, Alieh Saeedi via dev wrote:
> > > Hi all,
> > >
> > > I’d like to start a discussion on a new KIP, KIP-1285: DSL Opt-in Support
> > > for Header-Aware State Stores.
> > >
> > > KIP-1271 introduced headers-aware state stores in Kafka Streams. KIP-1285
> > > proposes adding corresponding DSL-level opt-in support, so that DSL
> > > applications can take advantage of these headers-aware stores while
> > keeping
> > > existing applications and topologies unchanged by default.
> > >
> > > You can find the KIP here:
> > >
> > >
> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-1285%3A+DSL+Opt-in+Support+for+Header-Aware+State+Stores
> > >
> > > Feedback would be greatly appreciated.
> > >
> > > Best,
> > > Alieh
> > >
> >
> >
Re: [DISCUSS] KIP-1285: DSL Opt-in Support for Header-Aware State Stores
Posted by Alieh Saeedi via dev <de...@kafka.apache.org>.
Hey Matthias,
Thanks for the suggestions. I agree with all of them. They all make sense
to me. I updated the KIP accordingly.
Bests,
Alieh
On Thu, Feb 19, 2026 at 8:33 PM Matthias J. Sax <mj...@apache.org> wrote:
> Thanks for the KIP. Great follow up to KIP-1271.
>
>
>
> Nit: The `DSL_STORE_TYPE_DOC` should be `private` and I also think we
> can improve the wording -- but that's more of am impl detail, not really
> critical for the KIP itself.
>
> How about:
>
> "Controls the state store type used by the DSL store supplier (see
> config 'dsl.store.suppliers.class'). 'default' uses timestamped stores.
> 'headers' uses the headers-aware stores."
>
>
>
> MJS1: Should we deprecate the existing `DslKeyValueParams(final String
> name, final boolean isTimestamped)` constructor, and make the newly
> introduced "type" a mandatory parameter? (Cf MJS2 below)
>
>
>
> MJS2: Should we add new enum `DslStoreType` with DEFAULT and HEADERS
> value, and change the constructor
>
> from
> - DslKeyValueParams(final String name, final boolean isTimestamped,
> final boolean isHeadersEnabled)
>
> to
> - DslKeyValueParams(final String name, final DslStoreType storeType)
>
> This would more align to the newly added String config (and the KIP
> calls out that we did not want to add a boolean config, what makes sense
> to me), and would subsume the existing `isTimestamped` parameter on the
> existing constructor which we could remove (Cf MJS1)
>
> [We have a similar thing for `group.protocol` config.]
>
>
> If yes, wondering if we should also deprecate `isTimestamped()` and not
> add `headersEnabled()` but instead add `storeType()` getter?
>
> Similar for window/session Dsl*Parameter classes.
>
>
>
> -Matthias
>
>
> On 2/18/26 3:30 AM, Alieh Saeedi via dev wrote:
> > Hi all,
> >
> > I’d like to start a discussion on a new KIP, KIP-1285: DSL Opt-in Support
> > for Header-Aware State Stores.
> >
> > KIP-1271 introduced headers-aware state stores in Kafka Streams. KIP-1285
> > proposes adding corresponding DSL-level opt-in support, so that DSL
> > applications can take advantage of these headers-aware stores while
> keeping
> > existing applications and topologies unchanged by default.
> >
> > You can find the KIP here:
> >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-1285%3A+DSL+Opt-in+Support+for+Header-Aware+State+Stores
> >
> > Feedback would be greatly appreciated.
> >
> > Best,
> > Alieh
> >
>
>
Re: [DISCUSS] KIP-1285: DSL Opt-in Support for Header-Aware State Stores
Posted by "Matthias J. Sax" <mj...@apache.org>.
Thanks for the KIP. Great follow up to KIP-1271.
Nit: The `DSL_STORE_TYPE_DOC` should be `private` and I also think we
can improve the wording -- but that's more of am impl detail, not really
critical for the KIP itself.
How about:
"Controls the state store type used by the DSL store supplier (see
config 'dsl.store.suppliers.class'). 'default' uses timestamped stores.
'headers' uses the headers-aware stores."
MJS1: Should we deprecate the existing `DslKeyValueParams(final String
name, final boolean isTimestamped)` constructor, and make the newly
introduced "type" a mandatory parameter? (Cf MJS2 below)
MJS2: Should we add new enum `DslStoreType` with DEFAULT and HEADERS
value, and change the constructor
from
- DslKeyValueParams(final String name, final boolean isTimestamped,
final boolean isHeadersEnabled)
to
- DslKeyValueParams(final String name, final DslStoreType storeType)
This would more align to the newly added String config (and the KIP
calls out that we did not want to add a boolean config, what makes sense
to me), and would subsume the existing `isTimestamped` parameter on the
existing constructor which we could remove (Cf MJS1)
[We have a similar thing for `group.protocol` config.]
If yes, wondering if we should also deprecate `isTimestamped()` and not
add `headersEnabled()` but instead add `storeType()` getter?
Similar for window/session Dsl*Parameter classes.
-Matthias
On 2/18/26 3:30 AM, Alieh Saeedi via dev wrote:
> Hi all,
>
> I’d like to start a discussion on a new KIP, KIP-1285: DSL Opt-in Support
> for Header-Aware State Stores.
>
> KIP-1271 introduced headers-aware state stores in Kafka Streams. KIP-1285
> proposes adding corresponding DSL-level opt-in support, so that DSL
> applications can take advantage of these headers-aware stores while keeping
> existing applications and topologies unchanged by default.
>
> You can find the KIP here:
>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-1285%3A+DSL+Opt-in+Support+for+Header-Aware+State+Stores
>
> Feedback would be greatly appreciated.
>
> Best,
> Alieh
>