Kafka Streams: a back-pressure question for windowed streams

[Tianji Li <sk...@gmail.com>]

Hi there,

I have a doubt regarding how to realize 'back-pressure' for windowed
streams.

Say I have a pipeline that consumes from a topic on a windowed basis, then
do some processing (whenever punctuate is called), and produces into
another topic.

If the incoming rates from all consumers is 10M/second, and the processing
rate of all the punctuates is something like 5M/second, then two things can
happen:

- If a in-memory store is used, on-heap memory will be drained gradually
and finally GC kicks in which leads to unnecessary rebalancing + other
things.

- If off-heap (RocksDB) is used, then over time, punctuate() will take
longer and longer time, and finally performance will be terrible +
something else that I do not know yet.

I understand the reason of these behaviors is that kafka Streams does
back-pressure by checking consumers buffer sizes, and StreamThread's buffer
size, but does NOT check state stores.

I think a solution for this is to 'add more Kafka Streams instance'. By
this I mean maybe today I need a processing rate of 1M/second, and tomorrow
I need 5M/second. Then a mechanism is needed for Kafka Streams to detect
this, and inform people who can add new instances either manually or better
automatically. And while waiting for people to react, the current running
Kafka Streams applications should not crash but can slow down a little bit
(by checking the state stores conditions, say number of records cached, or
total time taken for previous punctuates??).

Am I understanding correctly?

Thanks
Tianji

Re: Kafka Streams: a back-pressure question for windowed streams

[Michael Noll <mi...@confluent.io>]

Hi there!

In short, Kafka Streams ensures that your application consumes only as much
data (or: as fast) as it can process it.

The main "problem" you might encounter is not that you run into issues with
state stores (like in-memory stores or RocksDB stores), but -- which is a
more general issue -- that your application's *current capacity* (e.g. "I
am running (only) 10 instances of my app") doesn't allow it process the
data as fast as the data is coming in.  As you mentioned, you'd need to
monitor your application (Kafka/Streams exposes several such metrics) and,
if needed, launch additional instances of your application ("Ok, incoming
data load increased by 200%, so it looks like I need to run 10 + 20 = 30
instances of my application").  What you'd typically monitor is the
so-called "consumer lag" of your application.  In a nutshell, if the
consumer lag is 0 (zero), then your app processes the data as fast as it is
arriving.  If the consumer lag is "large" or, more importantly, growing
over time, then this means your application is not capable any longer to
keep up with the incoming data load.

FWIW, we provide guidance on how to properly size and plan the capacity of
your application:
http://docs.confluent.io/current/streams/sizing.html

Hope this helps,
Michael







On Thu, Apr 20, 2017 at 3:46 PM, Tianji Li <sk...@gmail.com> wrote:

> Hi there,
>
> I have a doubt regarding how to realize 'back-pressure' for windowed
> streams.
>
> Say I have a pipeline that consumes from a topic on a windowed basis, then
> do some processing (whenever punctuate is called), and produces into
> another topic.
>
> If the incoming rates from all consumers is 10M/second, and the processing
> rate of all the punctuates is something like 5M/second, then two things can
> happen:
>
> - If a in-memory store is used, on-heap memory will be drained gradually
> and finally GC kicks in which leads to unnecessary rebalancing + other
> things.
>
> - If off-heap (RocksDB) is used, then over time, punctuate() will take
> longer and longer time, and finally performance will be terrible +
> something else that I do not know yet.
>
> I understand the reason of these behaviors is that kafka Streams does
> back-pressure by checking consumers buffer sizes, and StreamThread's buffer
> size, but does NOT check state stores.
>
> I think a solution for this is to 'add more Kafka Streams instance'. By
> this I mean maybe today I need a processing rate of 1M/second, and tomorrow
> I need 5M/second. Then a mechanism is needed for Kafka Streams to detect
> this, and inform people who can add new instances either manually or better
> automatically. And while waiting for people to react, the current running
> Kafka Streams applications should not crash but can slow down a little bit
> (by checking the state stores conditions, say number of records cached, or
> total time taken for previous punctuates??).
>
> Am I understanding correctly?
>
> Thanks
> Tianji
>