Hi Matthias,
Thanks for your detailed answer. I have one more question (sorry to be so annoying!). If we had a topology making calls to a third party takes, let's say, 5 seconds to process requests and we pull a batch of 20 records, the app would be unresponsive in the event of a rebalance for up to 100 seconds (in the worst possible scenario). If the third party slows down or even goes down and we timeout after 25 seconds per call, the unresponsive period could grow to 500 seconds. I imagine that it means other consumers would not consume either until the rebalance is completed successfully. Is that correct? If it is, then we will assign explicit, narrower values to max.poll.timeout.ms to avoid halting the stream in that scenario. Thanks. ________________________________ From: Matthias J. Sax <matth...@confluent.io> Sent: 27 December 2017 19:54:38 To: users@kafka.apache.org Subject: Re: Kafka Streams - max.poll.interval.ms defaults to Integer.MAX_VALUE It would -- however, this is not an issue for KafkaStreams as we make sure the thread is either still alive or properly shut down. Thus, if an error happens and the thread really dies, KafkaStreams ensures that the heartbeat thread is stopped and thus, a rebalance would not block forever as it drops out of the group via session-timeout. And as long as a KafkaStreams instance does restore, the rebalance should block by design. Note, that all this does not hold for newer versions of KafkaStreams anymore. -Matthias On 12/27/17 6:55 AM, Javier Holguera wrote: > Hi Matthias, > > Thanks for your answer. It makes a lot of sense. > > Just a follow-up question. KIP-62 says: "we give the client as much as > max.poll.interval.ms to handle a batch of records, this is also the maximum > time before a consumer can be expected to rejoin the group in the worst > case". Does it mean that a broker would wait Integer.MAX_VALUE for a client > to report in the event of a rebalance? That sounds improbable, so I must be > missing something. > > Thanks. > > > -----Original Message----- > From: Matthias J. Sax [mailto:matth...@confluent.io] > Sent: Friday, December 22, 2017 9:13 PM > To: users@kafka.apache.org > Subject: Re: Kafka Streams - max.poll.interval.ms defaults to > Integer.MAX_VALUE > > The value was change to make Streams application robust against large state > restore phases during rebalance. > > Ie, it is targeted to exactly "fix" 2. If an application needs to restore > state, this state restore might take longer than the max.poll.interval.ms > parameter and thus, even if the application is in a good state it drops out > of the group. This results in rebalance "storms". The consumer default of 30 > seconds is too small for most applications and thus we set it to MAX_VALUE -- > if you have a good estimate on the max expected state restore time, you can > safely set the timeout to an appropriate value. > > Note, in Kafka 0.11 and 1.0 Kafka Streams state restore was largely improved > and it should not be an issue there to reduce the timeout accordingly. > > > -Matthias > > On 12/20/17 7:14 AM, Javier Holguera wrote: >> Hi, >> >> According to the documentation, "max.poll.interval.ms" defaults to >> Integer.MAX_VALUE for Kafka Streams since 0.10.2.1. >> >> Considering that the "max.poll.interval.ms" is: >> >> 1. A "processing timeout" to control an upper limit for processing a >> batch of records AND >> 2. The rebalance timeout that the client will communicate to the >> broker, according to KIP-62 >> >> How do Kafka Streams application detect slow consumers that are taking too >> long to process a batch of messages? What replaces the existing mechanism >> with a smaller "max.poll.interval.ms" where the application will willingly >> abandon the consumer group when the timeout expires? >> >> From the broker perspective, what does it mean that the application >> communicates a "rebalance timeout" of Integer.MAX_VALUE? I can imagine it >> will not wait for that long in a rebalance. What happens then? >> >> Thanks. >> >
