Hey,
the problem you describe it not totally fundamental, but there is no
workaround in KS currently... So right now, only EOS can fix it.
In the end, what you would need is a way to first write to the output
topic (and flush the write), before you update the state store.
Using the PAPI you could conceptually do `context.forward()` before you
call `store.put(...)`, but the KS runtime hides too many low level
controls that it would guarantee that `context.forward(...)` resulted in
successful write into the sink downstream, before it returns and
`state.put(...)` is executed... There is a missing `producer.flush()`
step in-between that you cannot perform.
Does this make sense?
Down the line though, I think that TX state-stores can fix it (KIP-892).
I believe that we can (and should) use this, even for ALOS mode,
allowing us to roll back state of error, ie, all write to the store are
"pending" until we do a commit, and for this case we first flush all
pending producer writes, before we "commit the store changes", and the
KS runtime can do the required flush step.
Thus, while TX state-store itself do not give you EOS, they should
address the issue at hand, and thus also unblock KIP-557 down the line.
Hope this helps,
-Matthias
On 2/25/25 4:22 PM, Steven Schlansker wrote:
> Hi kafka-users,
>
> We are implementing a Kafka Streams app that computes various
> streaming statistics over a corpus of data stored in Kafka topics.
> While some aggregates update often, others like 'min', 'max', or
> histogram buckets could have relatively few distinct updates relative
> to the input data.
>
> With our first implementation, we discovered that the majority of our
> time during catch-up processing is computing the same data redundantly
> over and over, at various stages of our pipeline. We then found that
> KIP-557, Emit-on-change support
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-557%3A+Add+emit+on+change+support+for+Kafka+Streams
> is reverted and doesn't seem to be coming back in the short term.
>
> So, we've implemented a work-around, where we trade additional storage
> to suppress redundant updates.
>
> record Holder<V> (V val, boolean changed) {}
> inputStream
> .mapValues(v -> new Holder<>(v, true))
> .aggregate(
> () -> null,
> (k, v, a) -> a == null ? new Holder<>(v.val(), true) : new
> Holder<>(v.val(), !Objects.equals(a.val(), v.val())),
> materialized)
> .toStream()
> .flatMapValues((k, v) -> v.changed() ?
> Collections.singletonList(v.val()) : Collections.emptyList())
> .toTable(materialized)
>
> This seems to mostly give us the semantics we want. The aggregation
> step statefully tracks if we saw this value as the previous state, and
> if not, forwards the new value.
>
> However, we subsequently seem to run into the same bug as highlighted
> in https://issues.apache.org/jira/browse/KAFKA-12508 - in at least
> once configuration, a failure / retry will observe the source topic
> rolling back but not the state store, causing downstream updates to be
> lost.
>
> This makes me wonder - is KAFKA-12508 a more general bug with at least
> once mode and any stateful store - where you can observe state in a
> store from a failed thread in the "future" relative to your source
> topic, since it already wrote to the changelog before failing? KIP-557
> is long reverted and we're able to observe similar behavior just with
> the DSL.
>
> I can't quite put my finger on the risk, but with a bunch of
> aggregates all through our app, it makes me a bit uneasy about
> at-least-once mode. I'm sure at least some of the stateful aggregate
> authors (myself included) did not consider this sort of failure case.
> We will probably try exactly-once mode for peace of mind.
>
> From there, back to emit on change - if there is no other path forward
> envisioned, perhaps we could consider re-enabling KIP-557 only for
> exactly-once streams apps? Then at least some users could benefit from
> it.
>
> Failing that, our hack workaround is somewhat complicated by the need
> to toStream / flatMap / toTable just to suppress "no change". It would
> be nicer if the aggregate could directly return "no update" to the
> KStreamAggregateProcessor, which would then know to skip forwarding
> the record. Null couldn't be used, but we could construct a different
> flag value or other signal.
> If we can't get emit-on-change back into Kafka Streams soon, we'd be
> interested in sponsoring a change like this if the community thinks it
> could be helpful.
>
> Sorry for the meandering train of thought, hopefully others are also
> interested in getting emit-on-change working still :)
>
> Best,
> Steven
the problem you describe it not totally fundamental, but there is no
workaround in KS currently... So right now, only EOS can fix it.
In the end, what you would need is a way to first write to the output
topic (and flush the write), before you update the state store.
Using the PAPI you could conceptually do `context.forward()` before you
call `store.put(...)`, but the KS runtime hides too many low level
controls that it would guarantee that `context.forward(...)` resulted in
successful write into the sink downstream, before it returns and
`state.put(...)` is executed... There is a missing `producer.flush()`
step in-between that you cannot perform.
Does this make sense?
Down the line though, I think that TX state-stores can fix it (KIP-892).
I believe that we can (and should) use this, even for ALOS mode,
allowing us to roll back state of error, ie, all write to the store are
"pending" until we do a commit, and for this case we first flush all
pending producer writes, before we "commit the store changes", and the
KS runtime can do the required flush step.
Thus, while TX state-store itself do not give you EOS, they should
address the issue at hand, and thus also unblock KIP-557 down the line.
Hope this helps,
-Matthias
On 2/25/25 4:22 PM, Steven Schlansker wrote:
> Hi kafka-users,
>
> We are implementing a Kafka Streams app that computes various
> streaming statistics over a corpus of data stored in Kafka topics.
> While some aggregates update often, others like 'min', 'max', or
> histogram buckets could have relatively few distinct updates relative
> to the input data.
>
> With our first implementation, we discovered that the majority of our
> time during catch-up processing is computing the same data redundantly
> over and over, at various stages of our pipeline. We then found that
> KIP-557, Emit-on-change support
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-557%3A+Add+emit+on+change+support+for+Kafka+Streams
> is reverted and doesn't seem to be coming back in the short term.
>
> So, we've implemented a work-around, where we trade additional storage
> to suppress redundant updates.
>
> record Holder<V> (V val, boolean changed) {}
> inputStream
> .mapValues(v -> new Holder<>(v, true))
> .aggregate(
> () -> null,
> (k, v, a) -> a == null ? new Holder<>(v.val(), true) : new
> Holder<>(v.val(), !Objects.equals(a.val(), v.val())),
> materialized)
> .toStream()
> .flatMapValues((k, v) -> v.changed() ?
> Collections.singletonList(v.val()) : Collections.emptyList())
> .toTable(materialized)
>
> This seems to mostly give us the semantics we want. The aggregation
> step statefully tracks if we saw this value as the previous state, and
> if not, forwards the new value.
>
> However, we subsequently seem to run into the same bug as highlighted
> in https://issues.apache.org/jira/browse/KAFKA-12508 - in at least
> once configuration, a failure / retry will observe the source topic
> rolling back but not the state store, causing downstream updates to be
> lost.
>
> This makes me wonder - is KAFKA-12508 a more general bug with at least
> once mode and any stateful store - where you can observe state in a
> store from a failed thread in the "future" relative to your source
> topic, since it already wrote to the changelog before failing? KIP-557
> is long reverted and we're able to observe similar behavior just with
> the DSL.
>
> I can't quite put my finger on the risk, but with a bunch of
> aggregates all through our app, it makes me a bit uneasy about
> at-least-once mode. I'm sure at least some of the stateful aggregate
> authors (myself included) did not consider this sort of failure case.
> We will probably try exactly-once mode for peace of mind.
>
> From there, back to emit on change - if there is no other path forward
> envisioned, perhaps we could consider re-enabling KIP-557 only for
> exactly-once streams apps? Then at least some users could benefit from
> it.
>
> Failing that, our hack workaround is somewhat complicated by the need
> to toStream / flatMap / toTable just to suppress "no change". It would
> be nicer if the aggregate could directly return "no update" to the
> KStreamAggregateProcessor, which would then know to skip forwarding
> the record. Null couldn't be used, but we could construct a different
> flag value or other signal.
> If we can't get emit-on-change back into Kafka Streams soon, we'd be
> interested in sponsoring a change like this if the community thinks it
> could be helpful.
>
> Sorry for the meandering train of thought, hopefully others are also
> interested in getting emit-on-change working still :)
>
> Best,
> Steven
Comments
Post a Comment