kafka-streams / window expiration because of shuffling

[Mathieu D <ma...@gmail.com>]

Hello there,

We're processing IOT device data, each device sending several metrics.

When we upgrade our streams app, we set a brand new 'application.id' to
reprocess a bit of past data, to warm up the state stores and aggregations
to make sure all outputs will be valid. Downstream is designed for "at
least once" so no problem with this bit of reprocessing.

When this restart+reprocessing occurs, we observe a peak of "Skipping
record for expired window" / "Skipping record for expired segment" warning
in logs.

My understanding so far is this:
- a part of our topology is keyed by deviceId.
- during the reprocessing, some tasks are moving faster for some
partitions, which means there's a substantial difference between the
various stream-times across tasks
- at some point in the topology, we re-key the data by (deviceId, metric)
for "group by metric" aggregations
- this shuffles the data:  deviceId1 was in partition 1 with eventTime1,
deviceId2 was in partition 2 with eventTime2, and now by the magic of
hashing a (device,metric) key, they are pushed together in the same
partitionX. If eventTime2 is far ahead of eventTime1, then all windows
will  expire at once.

Is this analysis correct ?

Then, what's the proper way to avoid this ? Manually do a .repartition()
with a custom partitioner after each .selectKey((device, metric)), and
before going through aggregations ?

Any other advice ?

Thanks for your insights

Mathieu

Re: kafka-streams / window expiration because of shuffling

[Mathieu D <ma...@gmail.com>]

Hi John

Thanks for your message.
Indeed there's a potential feature here. Should I log it on Jira ?

Mathieu

Le ven. 20 nov. 2020 à 21:53, John Roesler <vv...@apache.org> a écrit :

> Hi Mathieu,
>
> Ah, that is unfortunate. I believe your analysis is correct. In general,
> we have no good solution to the problem of upstream tasks moving ahead of
> each other and causing disorder in the repartition topics. Guozhang has
> done a substantial amount of thinking on this subject, though, and has some
> ideas for how we can improve the behavior.
>
> However, your situation is a special case, since the data actually doesn’t
> need to be shuffled. Ideally, there would be a way to say “I assert that
> the partitioning doesn’t need to change despite this key change”. That
> actually seems like a good feature for the repartition operator.
>
> In the absence of that feature, I think you’re on the right track. If you
> specify a partitioner that produces exactly the same partitioning as the
> source, you _should_ be able to avoid any shuffling, although there will
> still be a repartition topic there.
>
> I hope this helps,
> John
>
> On Fri, Nov 20, 2020, at 04:14, Mathieu D wrote:
> > Hello there,
> >
> > We're processing IOT device data, each device sending several metrics.
> >
> > When we upgrade our streams app, we set a brand new 'application.id' to
> > reprocess a bit of past data, to warm up the state stores and
> aggregations
> > to make sure all outputs will be valid. Downstream is designed for "at
> > least once" so no problem with this bit of reprocessing.
> >
> > When this restart+reprocessing occurs, we observe a peak of "Skipping
> > record for expired window" / "Skipping record for expired segment"
> warning
> > in logs.
> >
> > My understanding so far is this:
> > - a part of our topology is keyed by deviceId.
> > - during the reprocessing, some tasks are moving faster for some
> > partitions, which means there's a substantial difference between the
> > various stream-times across tasks
> > - at some point in the topology, we re-key the data by (deviceId, metric)
> > for "group by metric" aggregations
> > - this shuffles the data:  deviceId1 was in partition 1 with eventTime1,
> > deviceId2 was in partition 2 with eventTime2, and now by the magic of
> > hashing a (device,metric) key, they are pushed together in the same
> > partitionX. If eventTime2 is far ahead of eventTime1, then all windows
> > will  expire at once.
> >
> > Is this analysis correct ?
> >
> > Then, what's the proper way to avoid this ? Manually do a .repartition()
> > with a custom partitioner after each .selectKey((device, metric)), and
> > before going through aggregations ?
> >
> > Any other advice ?
> >
> > Thanks for your insights
> >
> > Mathieu
> >
>

Re: kafka-streams / window expiration because of shuffling

[John Roesler <vv...@apache.org>]

Hi Mathieu,

Ah, that is unfortunate. I believe your analysis is correct. In general, we have no good solution to the problem of upstream tasks moving ahead of each other and causing disorder in the repartition topics. Guozhang has done a substantial amount of thinking on this subject, though, and has some ideas for how we can improve the behavior. 

However, your situation is a special case, since the data actually doesn’t need to be shuffled. Ideally, there would be a way to say “I assert that the partitioning doesn’t need to change despite this key change”. That actually seems like a good feature for the repartition operator.

In the absence of that feature, I think you’re on the right track. If you specify a partitioner that produces exactly the same partitioning as the source, you _should_ be able to avoid any shuffling, although there will still be a repartition topic there. 

I hope this helps,
John

On Fri, Nov 20, 2020, at 04:14, Mathieu D wrote:
> Hello there,
> 
> We're processing IOT device data, each device sending several metrics.
> 
> When we upgrade our streams app, we set a brand new 'application.id' to
> reprocess a bit of past data, to warm up the state stores and aggregations
> to make sure all outputs will be valid. Downstream is designed for "at
> least once" so no problem with this bit of reprocessing.
> 
> When this restart+reprocessing occurs, we observe a peak of "Skipping
> record for expired window" / "Skipping record for expired segment" warning
> in logs.
> 
> My understanding so far is this:
> - a part of our topology is keyed by deviceId.
> - during the reprocessing, some tasks are moving faster for some
> partitions, which means there's a substantial difference between the
> various stream-times across tasks
> - at some point in the topology, we re-key the data by (deviceId, metric)
> for "group by metric" aggregations
> - this shuffles the data:  deviceId1 was in partition 1 with eventTime1,
> deviceId2 was in partition 2 with eventTime2, and now by the magic of
> hashing a (device,metric) key, they are pushed together in the same
> partitionX. If eventTime2 is far ahead of eventTime1, then all windows
> will  expire at once.
> 
> Is this analysis correct ?
> 
> Then, what's the proper way to avoid this ? Manually do a .repartition()
> with a custom partitioner after each .selectKey((device, metric)), and
> before going through aggregations ?
> 
> Any other advice ?
> 
> Thanks for your insights
> 
> Mathieu
>