I second the concern on with the one producer per task approach. At a high-level it seems to make sense but I think Damian is exactly right that that cuts against the general design of the producer. Many people have high input partition counts and will have high task counts as a result. I think processing 1000 partitions should not be an unreasonable thing to want to do.
The tricky bits will be: - Reduced effectiveness of batching (or more latency and memory to get equivalent batching). This doesn't show up in simple benchmarks because much of the penalty is I/O and CPU on the broker and the additional threads from all the producers can make a single-threaded benchmark seem faster. - TCP connection explosion. We maintain one connection per broker. This is already high since each app instance does this. This design though will add an additional multiplicative factor based on the partition count of the input. - Connection and metadata request storms. When an instance with 1000 tasks starts up it is going to try to create many thousands of connections and issue a thousand metadata requests all at once. - Memory usage. We currently default to 64MB per producer. This can be tuned down, but the fact that we are spreading the batching over more producers will fundamentally mean we need a lot more memory to get good perf and the memory usage will change as your task assignment changes so it will be hard to set correctly unless it is done automatically. - Metrics explosion (1000 producer instances, each with their own metrics to monitor). - Thread explosion, 1000 background threads, one per producer, each sending data. -Jay On Wed, Mar 1, 2017 at 3:05 AM, Damian Guy <damian....@gmail.com> wrote: > Hi Guozhang, > > Thanks for the KIP! This is an important feature for Kafka Streams and will > help to unlock a bunch of use cases. > > I have some concerns/questions: > > 1. Producer per task: I'm worried about the overhead this is going to > put on both the streams app and the Kafka Brokers. You can easily > imagine > an app consuming thousands of partitions. What load will this put on the > brokers? Am i correct in assuming that there will be metadata requests > per > Producer? The memory overhead in the streams app will also increase > fairly > significantly. Should we adjust ProducerConfig.BUFFER_MEMORY_CONFIG? > 2. State Store recovery: As we already know, restoring the entire > changelog can take an extremely long time. Even with a fairly small > dataset > and an inappropriately tuned segment size, this can take way too long. > My > concern is that failures happen and then recovery takes "forever" and we > end up in a situation where we need to change the max.poll.interval to > be > some very large number or else we end up in "rebalance hell". I don't > think > this provides a very good user experience. You mention RocksDB > checkpointing in the doc - should we explore this idea some more? i.e., > understand the penalty for checkpointing. Maybe checkpoint every *n* > commits? > 3. What does EoS mean for Caching? If we set the commit interval to > 100ms then the cache is not going to be very effective. Should it just > be > disabled? > > Thanks, > Damian > > On Tue, 28 Feb 2017 at 21:54 Guozhang Wang <wangg...@gmail.com> wrote: > > > Hi all, > > > > I have just created KIP-129 to leverage KIP-98 in Kafka Streams and > provide > > exactly-once processing semantics: > > > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP- > 129%3A+Streams+Exactly-Once+Semantics > > > > This KIP enables Streams users to optionally turn on exactly-once > > processing semantics without changing their app code at all by leveraging > > the transactional messaging features provided in KIP-98. > > > > The above wiki page provides a high-level view of the proposed changes, > > while detailed implementation design can be found in this Google doc: > > > > > > https://docs.google.com/document/d/1pGZ8xtOOyGwDYgH5vA6h19zOMMadu > FK1DAB8_gBYA2c > > > > We would love to hear your comments and suggestions. > > > > Thanks, > > -- Guozhang > > >
