ah, right - we can add as many strategies as we want. On Wed, May 25, 2016 at 10:54 AM, Joel Koshy <jjkosh...@gmail.com> wrote:
> > Yes it would be a protocol bump. > > > > Sorry - I'm officially confused. I think it may not be required - since the > more compact format would be associated with a new assignment strategy - > right? > > > > smaller than the plaintext PAL, but the post-compressed binary PAL is > just > > 25% smaller than the post-compressed plaintext PAL. IOW using a symbol > > table helps a lot but further compression on that already compact format > > would yield only marginal return. > > > > > So basically I feel we could get pretty far with a more compact field > > format for assignment and if we do that then we would potentially not > even > > want to do any compression. > > > > Also just wanted to add that this compression on the binary PAL did help > but the compression ratio was obviously not as high as plaintext > compression. > > > > > > On Tue, May 24, 2016 at 4:19 PM, Gwen Shapira <g...@confluent.io> wrote: > > > >> Regarding the change to the assignment field. It would be a protocol > bump, > >> otherwise consumers will not know how to parse the bytes the broker is > >> returning, right? > >> Or did I misunderstand the suggestion? > >> > >> On Tue, May 24, 2016 at 2:52 PM, Guozhang Wang <wangg...@gmail.com> > >> wrote: > >> > >> > I think for just solving issue 1), Jun's suggestion is sufficient and > >> > simple. So I'd prefer that approach. > >> > > >> > In addition, Jason's optimization on the assignment field would be > good > >> for > >> > 2) and 3) as well, and I like that optimization for its simplicity and > >> no > >> > format change as well. And in the future I'm in favor of considering > to > >> > change the in-memory cache format as Jiangjie suggested. > >> > > >> > Guozhang > >> > > >> > > >> > On Tue, May 24, 2016 at 12:42 PM, Becket Qin <becket....@gmail.com> > >> wrote: > >> > > >> > > Hi Jason, > >> > > > >> > > There are a few problems we want to solve here: > >> > > 1. The group metadata is too big to be appended to the log. > >> > > 2. Reduce the memory footprint on the broker > >> > > 3. Reduce the bytes transferred over the wire. > >> > > > >> > > To solve (1), I like your idea of having separate messages per > member. > >> > The > >> > > proposal (Onur's option 8) is to break metadata into small records > in > >> the > >> > > same uncompressed message set so each record is small. I agree it > >> would > >> > be > >> > > ideal if we are able to store the metadata separately for each > >> member. I > >> > > was also thinking about storing the metadata into multiple messages, > >> too. > >> > > What concerns me was that having multiple messages seems breaking > the > >> > > atomicity. I am not sure how we are going to deal with the potential > >> > > issues. For example, What if group metadata is replicated but the > >> member > >> > > metadata is not? It might be fine depending on the implementation > >> though, > >> > > but I am not sure. > >> > > > >> > > For (2) we want to store the metadata onto the disk, which is what > we > >> > have > >> > > to do anyway. The only question is in what format should we store > >> them. > >> > > > >> > > To address (3) we want to have the metadata to be compressed, which > is > >> > > contradict to the the above solution of (1). > >> > > > >> > > I think Jun's suggestion is probably still the simplest. To avoid > >> > changing > >> > > the behavior for consumers, maybe we can do that only for > >> offset_topic, > >> > > i.e, if the max fetch bytes of the fetch request is smaller than the > >> > > message size on the offset topic, we always return at least one full > >> > > message. This should avoid the unexpected problem on the client side > >> > > because supposedly only tools and brokers will fetch from the the > >> > internal > >> > > topics, > >> > > > >> > > As a modification to what you suggested, one solution I was thinking > >> was > >> > to > >> > > have multiple messages in a single compressed message. That means > for > >> > > SyncGroupResponse we still need to read the entire compressed > messages > >> > and > >> > > extract the inner messages, which seems not quite different from > >> having a > >> > > single message containing everything. But let me just put it here > and > >> see > >> > > if that makes sense. > >> > > > >> > > We can have a map of GroupMetadataKey -> GroupMetadataValueOffset. > >> > > > >> > > The GroupMetadataValue is stored in a compressed message. The inner > >> > > messages are the following: > >> > > > >> > > Inner Message 0: Version GroupId Generation > >> > > > >> > > Inner Message 1: MemberId MemberMetadata_1 (we can compress the > bytes > >> > here) > >> > > > >> > > Inner Message 2: MemberId MemberMetadata_2 > >> > > .... > >> > > Inner Message N: MemberId MemberMetadata_N > >> > > > >> > > The MemberMetadata format is the following: > >> > > MemberMetadata => Version Generation ClientId Host Subscription > >> > > Assignment > >> > > > >> > > So DescribeGroupResponse will just return the entire compressed > >> > > GroupMetadataMessage. SyncGroupResponse will return the > corresponding > >> > inner > >> > > message. > >> > > > >> > > Thanks, > >> > > > >> > > Jiangjie (Becket) Qin > >> > > > >> > > > >> > > > >> > > On Tue, May 24, 2016 at 9:14 AM, Jason Gustafson < > ja...@confluent.io> > >> > > wrote: > >> > > > >> > > > Hey Becket, > >> > > > > >> > > > I like your idea to store only the offset for the group metadata > in > >> > > memory. > >> > > > I think it would be safe to keep it in memory for a short time > after > >> > the > >> > > > rebalance completes, but after that, it's only real purpose is to > >> > answer > >> > > > DescribeGroup requests, so your proposal makes a lot of sense to > me. > >> > > > > >> > > > As for the specific problem with the size of the group metadata > >> message > >> > > for > >> > > > the MM case, if we cannot succeed in reducing the size of the > >> > > > subscription/assignment (which I think is still probably the best > >> > > > alternative if it can work), then I think there are some options > for > >> > > > changing the message format (option #8 in Onur's initial e-mail). > >> > > > Currently, the key used for storing the group metadata is this: > >> > > > > >> > > > GroupMetadataKey => Version GroupId > >> > > > > >> > > > And the value is something like this (some details elided): > >> > > > > >> > > > GroupMetadataValue => Version GroupId Generation [MemberMetadata] > >> > > > MemberMetadata => ClientId Host Subscription Assignment > >> > > > > >> > > > I don't think we can change the key without a lot of pain, but it > >> seems > >> > > > like we can change the value format. Maybe we can take the > >> > > > subscription/assignment payloads out of the value and introduce a > >> new > >> > > > "MemberMetadata" message for each member in the group. For > example: > >> > > > > >> > > > MemberMetadataKey => Version GroupId MemberId > >> > > > > >> > > > MemberMetadataValue => Version Generation ClientId Host > Subscription > >> > > > Assignment > >> > > > > >> > > > When a new generation is created, we would first write the group > >> > metadata > >> > > > message which includes the generation and all of the memberIds, > and > >> > then > >> > > > we'd write the member metadata messages. To answer the > DescribeGroup > >> > > > request, we'd read the group metadata at the cached offset and, > >> > depending > >> > > > on the version, all of the following member metadata. This would > be > >> > more > >> > > > complex to maintain, but it seems doable if it comes to it. > >> > > > > >> > > > Thanks, > >> > > > Jason > >> > > > > >> > > > On Mon, May 23, 2016 at 6:15 PM, Becket Qin <becket....@gmail.com > > > >> > > wrote: > >> > > > > >> > > > > It might worth thinking a little further. We have discussed this > >> > before > >> > > > > that we want to avoid holding all the group metadata in memory. > >> > > > > > >> > > > > I am thinking about the following end state: > >> > > > > > >> > > > > 1. Enable compression on the offset topic. > >> > > > > 2. Instead of holding the entire group metadata in memory on the > >> > > brokers, > >> > > > > each broker only keeps a [group -> Offset] map, the offset > points > >> to > >> > > the > >> > > > > message in the offset topic which holds the latest metadata of > the > >> > > group. > >> > > > > 3. DescribeGroupResponse will read from the offset topic > directly > >> > like > >> > > a > >> > > > > normal consumption, except that only exactly one message will be > >> > > > returned. > >> > > > > 4. SyncGroupResponse will read the message, extract the > assignment > >> > part > >> > > > and > >> > > > > send back the partition assignment. We can compress the > partition > >> > > > > assignment before sends it out if we want. > >> > > > > > >> > > > > Jiangjie (Becket) Qin > >> > > > > > >> > > > > On Mon, May 23, 2016 at 5:08 PM, Jason Gustafson < > >> ja...@confluent.io > >> > > > >> > > > > wrote: > >> > > > > > >> > > > > > > > >> > > > > > > Jason, doesn't gzip (or other compression) basically do > this? > >> If > >> > > the > >> > > > > > topic > >> > > > > > > is a string and the topic is repeated throughout, won't > >> > compression > >> > > > > > > basically replace all repeated instances of it with an index > >> > > > reference > >> > > > > to > >> > > > > > > the full string? > >> > > > > > > >> > > > > > > >> > > > > > Hey James, yeah, that's probably true, but keep in mind that > the > >> > > > > > compression happens on the broker side. It would be nice to > >> have a > >> > > more > >> > > > > > compact representation so that get some benefit over the wire > as > >> > > well. > >> > > > > This > >> > > > > > seems to be less of a concern here, so the bigger gains are > >> > probably > >> > > > from > >> > > > > > reducing the number of partitions that need to be listed > >> > > individually. > >> > > > > > > >> > > > > > -Jason > >> > > > > > > >> > > > > > On Mon, May 23, 2016 at 4:23 PM, Onur Karaman < > >> > > > > > onurkaraman.apa...@gmail.com> > >> > > > > > wrote: > >> > > > > > > >> > > > > > > When figuring out these optimizations, it's worth keeping in > >> mind > >> > > the > >> > > > > > > improvements when the message is uncompressed vs when it's > >> > > > compressed. > >> > > > > > > > >> > > > > > > When uncompressed: > >> > > > > > > Fixing the Assignment serialization to instead be a topic > >> index > >> > > into > >> > > > > the > >> > > > > > > corresponding member's subscription list would usually be a > >> good > >> > > > thing. > >> > > > > > > > >> > > > > > > I think the proposal is only worse when the topic names are > >> > small. > >> > > > The > >> > > > > > > Type.STRING we use in our protocol for the assignment's > >> > > > TOPIC_KEY_NAME > >> > > > > is > >> > > > > > > limited in length to Short.MAX_VALUE, so our strings are > first > >> > > > > prepended > >> > > > > > > with 2 bytes to indicate the string size. > >> > > > > > > > >> > > > > > > The new proposal does worse when: > >> > > > > > > 2 + utf_encoded_string_payload_size < index_type_size > >> > > > > > > in other words when: > >> > > > > > > utf_encoded_string_payload_size < index_type_size - 2 > >> > > > > > > > >> > > > > > > If the index type ends up being Type.INT32, then the > proposal > >> is > >> > > > worse > >> > > > > > when > >> > > > > > > the topic is length 1. > >> > > > > > > If the index type ends up being Type.INT64, then the > proposal > >> is > >> > > > worse > >> > > > > > when > >> > > > > > > the topic is less than length 6. > >> > > > > > > > >> > > > > > > When compressed: > >> > > > > > > As James Cheng brought up, I'm not sure how things change > when > >> > > > > > compression > >> > > > > > > comes into the picture. This would be worth investigating. > >> > > > > > > > >> > > > > > > On Mon, May 23, 2016 at 4:05 PM, James Cheng < > >> > wushuja...@gmail.com > >> > > > > >> > > > > > wrote: > >> > > > > > > > >> > > > > > > > > >> > > > > > > > > On May 23, 2016, at 10:59 AM, Jason Gustafson < > >> > > > ja...@confluent.io> > >> > > > > > > > wrote: > >> > > > > > > > > > >> > > > > > > > > 2. Maybe there's a better way to lay out the assignment > >> > without > >> > > > > > needing > >> > > > > > > > to > >> > > > > > > > > explicitly repeat the topic? For example, the leader > could > >> > sort > >> > > > the > >> > > > > > > > topics > >> > > > > > > > > for each member and just use an integer to represent the > >> > index > >> > > of > >> > > > > > each > >> > > > > > > > > topic within the sorted list (note this depends on the > >> > > > subscription > >> > > > > > > > > including the full topic list). > >> > > > > > > > > > >> > > > > > > > > Assignment -> [TopicIndex [Partition]] > >> > > > > > > > > > >> > > > > > > > > >> > > > > > > > Jason, doesn't gzip (or other compression) basically do > >> this? > >> > If > >> > > > the > >> > > > > > > topic > >> > > > > > > > is a string and the topic is repeated throughout, won't > >> > > compression > >> > > > > > > > basically replace all repeated instances of it with an > index > >> > > > > reference > >> > > > > > to > >> > > > > > > > the full string? > >> > > > > > > > > >> > > > > > > > -James > >> > > > > > > > > >> > > > > > > > > You could even combine these two options so that you > have > >> > only > >> > > 3 > >> > > > > > > integers > >> > > > > > > > > for each topic assignment: > >> > > > > > > > > > >> > > > > > > > > Assignment -> [TopicIndex MinPartition MaxPartition] > >> > > > > > > > > > >> > > > > > > > > There may even be better options with a little more > >> thought. > >> > > All > >> > > > of > >> > > > > > > this > >> > > > > > > > is > >> > > > > > > > > just part of the client-side protocol, so it wouldn't > >> require > >> > > any > >> > > > > > > version > >> > > > > > > > > bumps on the broker. What do you think? > >> > > > > > > > > > >> > > > > > > > > Thanks, > >> > > > > > > > > Jason > >> > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > On Mon, May 23, 2016 at 9:17 AM, Guozhang Wang < > >> > > > wangg...@gmail.com > >> > > > > > > >> > > > > > > > wrote: > >> > > > > > > > > > >> > > > > > > > >> The original concern is that regex may not be > efficiently > >> > > > > supported > >> > > > > > > > >> across-languages, but if there is a neat workaround I > >> would > >> > > love > >> > > > > to > >> > > > > > > > learn. > >> > > > > > > > >> > >> > > > > > > > >> Guozhang > >> > > > > > > > >> > >> > > > > > > > >> On Mon, May 23, 2016 at 5:31 AM, Ismael Juma < > >> > > ism...@juma.me.uk > >> > > > > > >> > > > > > > wrote: > >> > > > > > > > >> > >> > > > > > > > >>> +1 to Jun's suggestion. > >> > > > > > > > >>> > >> > > > > > > > >>> Having said that, as a general point, I think we > should > >> > > > consider > >> > > > > > > > >> supporting > >> > > > > > > > >>> topic patterns in the wire protocol. It requires some > >> > > thinking > >> > > > > for > >> > > > > > > > >>> cross-language support, but it seems surmountable and > it > >> > > could > >> > > > > make > >> > > > > > > > >> certain > >> > > > > > > > >>> operations a lot more efficient (the fact that a basic > >> > regex > >> > > > > > > > subscription > >> > > > > > > > >>> causes the consumer to request metadata for all topics > >> is > >> > not > >> > > > > > great). > >> > > > > > > > >>> > >> > > > > > > > >>> Ismael > >> > > > > > > > >>> > >> > > > > > > > >>> On Sun, May 22, 2016 at 11:49 PM, Guozhang Wang < > >> > > > > > wangg...@gmail.com> > >> > > > > > > > >>> wrote: > >> > > > > > > > >>> > >> > > > > > > > >>>> I like Jun's suggestion in changing the handling > >> logics of > >> > > > > single > >> > > > > > > > large > >> > > > > > > > >>>> message on the consumer side. > >> > > > > > > > >>>> > >> > > > > > > > >>>> As for the case of "a single group subscribing to > 3000 > >> > > > topics", > >> > > > > > with > >> > > > > > > > >> 100 > >> > > > > > > > >>>> consumers the 2.5Mb Gzip size is reasonable to me > (when > >> > > > storing > >> > > > > in > >> > > > > > > ZK, > >> > > > > > > > >> we > >> > > > > > > > >>>> also have the znode limit which is set to 1Mb by > >> default, > >> > > > though > >> > > > > > > > >>> admittedly > >> > > > > > > > >>>> it is only for one consumer). And if we do the change > >> as > >> > Jun > >> > > > > > > > suggested, > >> > > > > > > > >>>> 2.5Mb on follower's memory pressure is OK I think. > >> > > > > > > > >>>> > >> > > > > > > > >>>> > >> > > > > > > > >>>> Guozhang > >> > > > > > > > >>>> > >> > > > > > > > >>>> > >> > > > > > > > >>>> On Sat, May 21, 2016 at 12:51 PM, Onur Karaman < > >> > > > > > > > >>>> onurkaraman.apa...@gmail.com > >> > > > > > > > >>>>> wrote: > >> > > > > > > > >>>> > >> > > > > > > > >>>>> Results without compression: > >> > > > > > > > >>>>> 1 consumer 292383 bytes > >> > > > > > > > >>>>> 5 consumers 1079579 bytes * the tipping point > >> > > > > > > > >>>>> 10 consumers 1855018 bytes > >> > > > > > > > >>>>> 20 consumers 2780220 bytes > >> > > > > > > > >>>>> 30 consumers 3705422 bytes > >> > > > > > > > >>>>> 40 consumers 4630624 bytes > >> > > > > > > > >>>>> 50 consumers 5555826 bytes > >> > > > > > > > >>>>> 60 consumers 6480788 bytes > >> > > > > > > > >>>>> 70 consumers 7405750 bytes > >> > > > > > > > >>>>> 80 consumers 8330712 bytes > >> > > > > > > > >>>>> 90 consumers 9255674 bytes > >> > > > > > > > >>>>> 100 consumers 10180636 bytes > >> > > > > > > > >>>>> > >> > > > > > > > >>>>> So it looks like gzip compression shrinks the > message > >> > size > >> > > by > >> > > > > 4x. > >> > > > > > > > >>>>> > >> > > > > > > > >>>>> On Sat, May 21, 2016 at 9:47 AM, Jun Rao < > >> > j...@confluent.io > >> > > > > >> > > > > > wrote: > >> > > > > > > > >>>>> > >> > > > > > > > >>>>>> Onur, > >> > > > > > > > >>>>>> > >> > > > > > > > >>>>>> Thanks for the investigation. > >> > > > > > > > >>>>>> > >> > > > > > > > >>>>>> Another option is to just fix how we deal with the > >> case > >> > > > when a > >> > > > > > > > >>> message > >> > > > > > > > >>>> is > >> > > > > > > > >>>>>> larger than the fetch size. Today, if the fetch > size > >> is > >> > > > > smaller > >> > > > > > > > >> than > >> > > > > > > > >>>> the > >> > > > > > > > >>>>>> fetch size, the consumer will get stuck. Instead, > we > >> can > >> > > > > simply > >> > > > > > > > >>> return > >> > > > > > > > >>>>> the > >> > > > > > > > >>>>>> full message if it's larger than the fetch size w/o > >> > > > requiring > >> > > > > > the > >> > > > > > > > >>>>> consumer > >> > > > > > > > >>>>>> to manually adjust the fetch size. On the broker > >> side, > >> > to > >> > > > > serve > >> > > > > > a > >> > > > > > > > >>> fetch > >> > > > > > > > >>>>>> request, we already do an index lookup and then > scan > >> the > >> > > > log a > >> > > > > > bit > >> > > > > > > > >> to > >> > > > > > > > >>>>> find > >> > > > > > > > >>>>>> the message with the requested offset. We can just > >> check > >> > > the > >> > > > > > size > >> > > > > > > > >> of > >> > > > > > > > >>>> that > >> > > > > > > > >>>>>> message and return the full message if its size is > >> > larger > >> > > > than > >> > > > > > the > >> > > > > > > > >>>> fetch > >> > > > > > > > >>>>>> size. This way, fetch size is really for > performance > >> > > > > > optimization, > >> > > > > > > > >>> i.e. > >> > > > > > > > >>>>> in > >> > > > > > > > >>>>>> the common case, we will not return more bytes than > >> > fetch > >> > > > > size, > >> > > > > > > but > >> > > > > > > > >>> if > >> > > > > > > > >>>>>> there is a large message, we will return more bytes > >> than > >> > > the > >> > > > > > > > >>> specified > >> > > > > > > > >>>>>> fetch size. In practice, large messages are rare. > >> So, it > >> > > > > > shouldn't > >> > > > > > > > >>>>> increase > >> > > > > > > > >>>>>> the memory consumption on the client too much. > >> > > > > > > > >>>>>> > >> > > > > > > > >>>>>> Jun > >> > > > > > > > >>>>>> > >> > > > > > > > >>>>>> On Sat, May 21, 2016 at 3:34 AM, Onur Karaman < > >> > > > > > > > >>>>>> onurkaraman.apa...@gmail.com> > >> > > > > > > > >>>>>> wrote: > >> > > > > > > > >>>>>> > >> > > > > > > > >>>>>>> Hey everyone. So I started doing some tests on the > >> new > >> > > > > > > > >>>>>> consumer/coordinator > >> > > > > > > > >>>>>>> to see if it could handle more strenuous use cases > >> like > >> > > > > > mirroring > >> > > > > > > > >>>>>> clusters > >> > > > > > > > >>>>>>> with thousands of topics and thought I'd share > >> > whatever I > >> > > > > have > >> > > > > > so > >> > > > > > > > >>>> far. > >> > > > > > > > >>>>>>> > >> > > > > > > > >>>>>>> The scalability limit: the amount of group > metadata > >> we > >> > > can > >> > > > > fit > >> > > > > > > > >> into > >> > > > > > > > >>>> one > >> > > > > > > > >>>>>>> message > >> > > > > > > > >>>>>>> > >> > > > > > > > >>>>>>> Some background: > >> > > > > > > > >>>>>>> Client-side assignment is implemented in two > phases > >> > > > > > > > >>>>>>> 1. a PreparingRebalance phase that identifies > >> members > >> > of > >> > > > the > >> > > > > > > > >> group > >> > > > > > > > >>>> and > >> > > > > > > > >>>>>>> aggregates member subscriptions. > >> > > > > > > > >>>>>>> 2. an AwaitingSync phase that waits for the group > >> > leader > >> > > to > >> > > > > > > > >> decide > >> > > > > > > > >>>>> member > >> > > > > > > > >>>>>>> assignments based on the member subscriptions > across > >> > the > >> > > > > group. > >> > > > > > > > >>>>>>> - The leader announces this decision with a > >> > > > > SyncGroupRequest. > >> > > > > > > > >> The > >> > > > > > > > >>>>>>> GroupCoordinator handles SyncGroupRequests by > >> appending > >> > > all > >> > > > > > group > >> > > > > > > > >>>> state > >> > > > > > > > >>>>>>> into a single message under the __consumer_offsets > >> > topic. > >> > > > > This > >> > > > > > > > >>>> message > >> > > > > > > > >>>>> is > >> > > > > > > > >>>>>>> keyed on the group id and contains each member > >> > > subscription > >> > > > > as > >> > > > > > > > >> well > >> > > > > > > > >>>> as > >> > > > > > > > >>>>>> the > >> > > > > > > > >>>>>>> decided assignment for each member. > >> > > > > > > > >>>>>>> > >> > > > > > > > >>>>>>> The environment: > >> > > > > > > > >>>>>>> - one broker > >> > > > > > > > >>>>>>> - one __consumer_offsets partition > >> > > > > > > > >>>>>>> - offsets.topic.compression.codec=1 // this is > gzip > >> > > > > > > > >>>>>>> - broker has my pending KAFKA-3718 patch that > >> actually > >> > > > makes > >> > > > > > use > >> > > > > > > > >> of > >> > > > > > > > >>>>>>> offsets.topic.compression.codec: > >> > > > > > > > >>>>>> https://github.com/apache/kafka/pull/1394 > >> > > > > > > > >>>>>>> - around 3000 topics. This is an actual subset of > >> > topics > >> > > > from > >> > > > > > one > >> > > > > > > > >>> of > >> > > > > > > > >>>>> our > >> > > > > > > > >>>>>>> clusters. > >> > > > > > > > >>>>>>> - topics have 8 partitions > >> > > > > > > > >>>>>>> - topics are 25 characters long on average > >> > > > > > > > >>>>>>> - one group with a varying number of consumers > each > >> > > > hardcoded > >> > > > > > > > >> with > >> > > > > > > > >>>> all > >> > > > > > > > >>>>>> the > >> > > > > > > > >>>>>>> topics just to make the tests more consistent. > >> > > wildcarding > >> > > > > with > >> > > > > > > > >> .* > >> > > > > > > > >>>>> should > >> > > > > > > > >>>>>>> have the same effect once the subscription hits > the > >> > > > > coordinator > >> > > > > > > > >> as > >> > > > > > > > >>>> the > >> > > > > > > > >>>>>>> subscription has already been fully expanded out > to > >> the > >> > > > list > >> > > > > of > >> > > > > > > > >>>> topics > >> > > > > > > > >>>>> by > >> > > > > > > > >>>>>>> the consumers. > >> > > > > > > > >>>>>>> - I added some log messages to Log.scala to print > >> out > >> > the > >> > > > > > message > >> > > > > > > > >>>> sizes > >> > > > > > > > >>>>>>> after compression > >> > > > > > > > >>>>>>> - there are no producers at all and auto commits > are > >> > > > > disabled. > >> > > > > > > > >> The > >> > > > > > > > >>>> only > >> > > > > > > > >>>>>>> topic with messages getting added is the > >> > > __consumer_offsets > >> > > > > > topic > >> > > > > > > > >>> and > >> > > > > > > > >>>>>>> they're only from storing group metadata while > >> > processing > >> > > > > > > > >>>>>>> SyncGroupRequests. > >> > > > > > > > >>>>>>> > >> > > > > > > > >>>>>>> Results: > >> > > > > > > > >>>>>>> The results below show that we exceed the 1000012 > >> byte > >> > > > > > > > >>>>>>> KafkaConfig.messageMaxBytes limit relatively > quickly > >> > > > (between > >> > > > > > > > >> 30-40 > >> > > > > > > > >>>>>>> consumers): > >> > > > > > > > >>>>>>> 1 consumer 54739 bytes > >> > > > > > > > >>>>>>> 5 consumers 261524 bytes > >> > > > > > > > >>>>>>> 10 consumers 459804 bytes > >> > > > > > > > >>>>>>> 20 consumers 702499 bytes > >> > > > > > > > >>>>>>> 30 consumers 930525 bytes > >> > > > > > > > >>>>>>> 40 consumers 1115657 bytes * the tipping point > >> > > > > > > > >>>>>>> 50 consumers 1363112 bytes > >> > > > > > > > >>>>>>> 60 consumers 1598621 bytes > >> > > > > > > > >>>>>>> 70 consumers 1837359 bytes > >> > > > > > > > >>>>>>> 80 consumers 2066934 bytes > >> > > > > > > > >>>>>>> 90 consumers 2310970 bytes > >> > > > > > > > >>>>>>> 100 consumers 2542735 bytes > >> > > > > > > > >>>>>>> > >> > > > > > > > >>>>>>> Note that the growth itself is pretty gradual. > >> Plotting > >> > > the > >> > > > > > > > >> points > >> > > > > > > > >>>>> makes > >> > > > > > > > >>>>>> it > >> > > > > > > > >>>>>>> look roughly linear w.r.t the number of consumers: > >> > > > > > > > >>>>>>> > >> > > > > > > > >>>>>>> > >> > > > > > > > >>>>>> > >> > > > > > > > >>>>> > >> > > > > > > > >>>> > >> > > > > > > > >>> > >> > > > > > > > >> > >> > > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > https://www.wolframalpha.com/input/?i=(1,+54739),+(5,+261524),+(10,+459804),+(20,+702499),+(30,+930525),+(40,+1115657),+(50,+1363112),+(60,+1598621),+(70,+1837359),+(80,+2066934),+(90,+2310970),+(100,+2542735) > >> > > > > > > > >>>>>>> > >> > > > > > > > >>>>>>> Also note that these numbers aren't averages or > >> medians > >> > > or > >> > > > > > > > >> anything > >> > > > > > > > >>>>> like > >> > > > > > > > >>>>>>> that. It's just the byte size from a given run. I > >> did > >> > run > >> > > > > them > >> > > > > > a > >> > > > > > > > >>> few > >> > > > > > > > >>>>>> times > >> > > > > > > > >>>>>>> and saw similar results. > >> > > > > > > > >>>>>>> > >> > > > > > > > >>>>>>> Impact: > >> > > > > > > > >>>>>>> Even after adding gzip to the __consumer_offsets > >> topic > >> > > with > >> > > > > my > >> > > > > > > > >>>> pending > >> > > > > > > > >>>>>>> KAFKA-3718 patch, the AwaitingSync phase of the > >> group > >> > > fails > >> > > > > > with > >> > > > > > > > >>>>>>> RecordTooLargeException. This means the combined > >> size > >> > of > >> > > > each > >> > > > > > > > >>>> member's > >> > > > > > > > >>>>>>> subscriptions and assignments exceeded the > >> > > > > > > > >>>> KafkaConfig.messageMaxBytes > >> > > > > > > > >>>>> of > >> > > > > > > > >>>>>>> 1000012 bytes. The group ends up dying. > >> > > > > > > > >>>>>>> > >> > > > > > > > >>>>>>> Options: > >> > > > > > > > >>>>>>> 1. Config change: reduce the number of consumers > in > >> the > >> > > > > group. > >> > > > > > > > >> This > >> > > > > > > > >>>>> isn't > >> > > > > > > > >>>>>>> always a realistic answer in more strenuous use > >> cases > >> > > like > >> > > > > > > > >>>> MirrorMaker > >> > > > > > > > >>>>>>> clusters or for auditing. > >> > > > > > > > >>>>>>> 2. Config change: split the group into smaller > >> groups > >> > > which > >> > > > > > > > >>> together > >> > > > > > > > >>>>> will > >> > > > > > > > >>>>>>> get full coverage of the topics. This gives each > >> group > >> > > > > member a > >> > > > > > > > >>>> smaller > >> > > > > > > > >>>>>>> subscription.(ex: g1 has topics starting with a-m > >> while > >> > > g2 > >> > > > > has > >> > > > > > > > >>> topics > >> > > > > > > > >>>>>>> starting ith n-z). This would be operationally > >> painful > >> > to > >> > > > > > manage. > >> > > > > > > > >>>>>>> 3. Config change: split the topics among members > of > >> the > >> > > > > group. > >> > > > > > > > >>> Again > >> > > > > > > > >>>>> this > >> > > > > > > > >>>>>>> gives each group member a smaller subscription. > This > >> > > would > >> > > > > also > >> > > > > > > > >> be > >> > > > > > > > >>>>>>> operationally painful to manage. > >> > > > > > > > >>>>>>> 4. Config change: bump up > >> KafkaConfig.messageMaxBytes > >> > (a > >> > > > > > > > >>> topic-level > >> > > > > > > > >>>>>>> config) and KafkaConfig.replicaFetchMaxBytes (a > >> > > > broker-level > >> > > > > > > > >>> config). > >> > > > > > > > >>>>>>> Applying messageMaxBytes to just the > >> __consumer_offsets > >> > > > topic > >> > > > > > > > >> seems > >> > > > > > > > >>>>>>> relatively harmless, but bumping up the > broker-level > >> > > > > > > > >>>>> replicaFetchMaxBytes > >> > > > > > > > >>>>>>> would probably need more attention. > >> > > > > > > > >>>>>>> 5. Config change: try different compression > codecs. > >> > Based > >> > > > on > >> > > > > 2 > >> > > > > > > > >>>> minutes > >> > > > > > > > >>>>> of > >> > > > > > > > >>>>>>> googling, it seems like lz4 and snappy are faster > >> than > >> > > gzip > >> > > > > but > >> > > > > > > > >>> have > >> > > > > > > > >>>>>> worse > >> > > > > > > > >>>>>>> compression, so this probably won't help. > >> > > > > > > > >>>>>>> 6. Implementation change: support sending the > regex > >> > over > >> > > > the > >> > > > > > wire > >> > > > > > > > >>>>> instead > >> > > > > > > > >>>>>>> of the fully expanded topic subscriptions. I think > >> > people > >> > > > > said > >> > > > > > in > >> > > > > > > > >>> the > >> > > > > > > > >>>>>> past > >> > > > > > > > >>>>>>> that different languages have subtle differences > in > >> > > regex, > >> > > > so > >> > > > > > > > >> this > >> > > > > > > > >>>>>> doesn't > >> > > > > > > > >>>>>>> play nicely with cross-language groups. > >> > > > > > > > >>>>>>> 7. Implementation change: maybe we can reverse the > >> > > mapping? > >> > > > > > > > >> Instead > >> > > > > > > > >>>> of > >> > > > > > > > >>>>>>> mapping from member to subscriptions, we can map a > >> > > > > subscription > >> > > > > > > > >> to > >> > > > > > > > >>> a > >> > > > > > > > >>>>> list > >> > > > > > > > >>>>>>> of members. > >> > > > > > > > >>>>>>> 8. Implementation change: maybe we can try to > break > >> > apart > >> > > > the > >> > > > > > > > >>>>>> subscription > >> > > > > > > > >>>>>>> and assignments from the same SyncGroupRequest > into > >> > > > multiple > >> > > > > > > > >>> records? > >> > > > > > > > >>>>>> They > >> > > > > > > > >>>>>>> can still go to the same message set and get > >> appended > >> > > > > together. > >> > > > > > > > >>> This > >> > > > > > > > >>>>> way > >> > > > > > > > >>>>>>> the limit become the segment size, which shouldn't > >> be a > >> > > > > > problem. > >> > > > > > > > >>> This > >> > > > > > > > >>>>> can > >> > > > > > > > >>>>>>> be tricky to get right because we're currently > >> keying > >> > > these > >> > > > > > > > >>> messages > >> > > > > > > > >>>> on > >> > > > > > > > >>>>>> the > >> > > > > > > > >>>>>>> group, so I think records from the same rebalance > >> might > >> > > > > > > > >>> accidentally > >> > > > > > > > >>>>>>> compact one another, but my understanding of > >> compaction > >> > > > isn't > >> > > > > > > > >> that > >> > > > > > > > >>>>> great. > >> > > > > > > > >>>>>>> > >> > > > > > > > >>>>>>> Todo: > >> > > > > > > > >>>>>>> It would be interesting to rerun the tests with no > >> > > > > compression > >> > > > > > > > >> just > >> > > > > > > > >>>> to > >> > > > > > > > >>>>>> see > >> > > > > > > > >>>>>>> how much gzip is helping but it's getting late. > >> Maybe > >> > > > > tomorrow? > >> > > > > > > > >>>>>>> > >> > > > > > > > >>>>>>> - Onur > >> > > > > > > > >>>>>>> > >> > > > > > > > >>>>>> > >> > > > > > > > >>>>> > >> > > > > > > > >>>> > >> > > > > > > > >>>> > >> > > > > > > > >>>> > >> > > > > > > > >>>> -- > >> > > > > > > > >>>> -- Guozhang > >> > > > > > > > >>>> > >> > > > > > > > >>> > >> > > > > > > > >> > >> > > > > > > > >> > >> > > > > > > > >> > >> > > > > > > > >> -- > >> > > > > > > > >> -- Guozhang > >> > > > > > > > >> > >> > > > > > > > > >> > > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > > >> > > >> > -- > >> > -- Guozhang > >> > > >> > > > > >
