[
https://issues.apache.org/jira/browse/KAFKA-3775?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15326484#comment-15326484
]
Yuto Kawamura commented on KAFKA-3775:
--------------------------------------
[~guozhang] Yes, both of your short and long term measures are sounds
reasonable to me.
Personally I'm satisfied with this overall discussion in terms of to avoid
memory pressure of when there's very less number of instances of Kafka Streams.
As I explained, for now, {{max.poll.records}} worked well at least to stop my
app from dying by OOM. I'm also thinking that it'd be better to have much
explicit way of configuring memory management of Kafka Streams as you said but
that's not urgent for me for now.
Still don't know what to do with concentrated network traffic once it became a
serious level but I will fill up or reopen an issue once it became an actual
problem like killing colocated services by exhausting NIC capacity for example.
So I'm going to close my PR and this issue as it's not likely a good way of
introducing throttling as most of people disagreed to introduce a risk of
unassigned partitions. Expecting we'll continue discussing about the way you
suggested in the proper KIP or issue for that.
Again, thanks for all your help :)
> Throttle maximum number of tasks assigned to a single KafkaStreams
> ------------------------------------------------------------------
>
> Key: KAFKA-3775
> URL: https://issues.apache.org/jira/browse/KAFKA-3775
> Project: Kafka
> Issue Type: Improvement
> Components: streams
> Affects Versions: 0.10.0.0
> Reporter: Yuto Kawamura
> Assignee: Yuto Kawamura
> Fix For: 0.10.1.0
>
>
> As of today, if I start a Kafka Streams app on a single machine which
> consists of single KafkaStreams instance, that instance gets all partitions
> of the target topic assigned.
> As we're using it to process topics which has huge number of partitions and
> message traffic, it is a problem that we don't have a way of throttling the
> maximum amount of partitions assigned to a single instance.
> In fact, when we started a Kafka Streams app which consumes a topic which has
> more than 10MB/sec traffic of each partition we saw that all partitions
> assigned to the first instance and soon the app dead by OOM.
> I know that there's some workarounds considerable here. for example:
> - Start multiple instances at once so the partitions distributed evenly.
> => Maybe works. but as Kafka Streams is a library but not an execution
> framework, there's no predefined procedure of starting Kafka Streams apps so
> some users might wanna take an option to start the first single instance and
> check if it works as expected with lesster number of partitions(I want :p)
> - Adjust config parameters such as {{buffered.records.per.partition}},
> {{max.partition.fetch.bytes}} and {{max.poll.records}} to reduce the heap
> pressure.
> => Maybe works. but still have two problems IMO:
> - Still leads traffic explosion with high throughput processing as it
> accepts all incoming messages from hundreads of partitions.
> - In the first place, by the distributed system principle, it's wired that
> users don't have a away to control maximum "partitions" assigned to a single
> shard(an instance of KafkaStreams here). Users should be allowed to provide
> the maximum amount of partitions that is considered as possible to be
> processed with single instance(or host).
> Here, I'd like to introduce a new configuration parameter
> {{max.tasks.assigned}}, which limits the number of tasks(a notion of
> partition) assigned to the processId(which is the notion of single
> KafkaStreams instance).
> At the same time we need to change StreamPartitionAssignor(TaskAssignor) to
> tolerate the incomplete assignment. That is, Kafka Streams should continue
> working for the part of partitions even there are some partitions left
> unassigned, in order to satisfy this> "user may want to take an option to
> start the first single instance and check if it works as expected with
> lesster number of partitions(I want :p)".
> I've implemented the rough POC for this. PTAL and if it make sense I will
> continue sophisticating it.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
