Should this be Retriable? in theory, others could leave the group while we are retrying? Not 100% sure if we want any retriable exception in join group, which is time sensitive.

cc @hachikuji for second opinion :)

In the KIP discussions we settled on fatally failing a consumer that encounters this error.

Does the consumer retry Retriable exceptions up to a certain point? I'm not sure. If we could retry up to 3 times and then fatally fail, that would be somewhat better

I think it probably should be fatal. If a user is bumping up against the limit, we want them to know about it. Not sure there is much benefit trying to operate too finely at the edge.

nit: Can you configure your IDE not to reorder imports?

Sure, reverted these back

Any reason this isn't in GroupConfig?

Completely missed it, my bad

few style nits:

• the if-else style here is not consistent
• use of NoMemberId vs JoinGroupRequest.UNKNOWN_MEMBER_ID. I'm not sure why we have both, but we probably want to be consistent with the other two places where we return an error here.
• Do we really need the "return" here? Scala style recommends against multiple exit points.

re: NoMemberId, I had seen it from the removeMemberAndUpdateGroup. I also think it should use JoinGroupRequest.UNKNOWN_MEMBER_ID, especially now that the lines are right next to each other after the rebase

Agreed with the style, it reads better as an else if for sure

Do we really need to shrink? If we detect a size issue and force a rebalance, everyone is forced to re-join and then the "group too large" logic in the JoinGroup path will kick in.

Personally, I'd leave existing groups alone. If the size causes issues, they'll timeout and rebalance themselves anyway. And if there are no issues, why introduce new issues.

@hachikuji should probably check the logic here, since reloading of groups has a bunch of logic (including transactions) that I'm not comfortable with.

Yeah, this is the trickiest part of the KIP. My naive initial reasoning was that we might want to maintain the same leader but now that I think of it that is more or less pointless (unless it loads some state in its initial PartitionAssignor) and we most likely will pick another leader from the subsequent rebalance

It's not a bad idea to preserve the leader, but I think Gwen's suggestion seems simpler. I think the main thing I was hoping is that the resizing could be graceful. The existing members should be able to commit their offsets before getting kicked out. It seems like we don't get that with this approach since we are forcefully removing some members.

By kicking out the extra members, like Jason said those members lose their opportunities to commit the last batch. However on the other hand, this should only happen when the server operator believes that an absurd group size is configured that should be prevented. In this sense, the group is interpreted as over capacity, so allowing rest of consumers take over the jobs should be very light lifting, which I don't see an obvious negative impact. Thoughts?

In this sense, the group is interpreted as over capacity, so allowing rest of consumers take over the jobs should be very light lifting
Sorry, I couldn't understand this exactly. Do you mean that it's reasonable to "shrink" the group in one way or another and not let it continue over capacity?

My thinking is to go with a forced rebalance

do we really want to throw? Rather than just silently do nothing?

I don't have a strong opinion about this

Perhaps a log message (warning) would suffice?

Should we just use assert here? As long as the change is robust, this should never happen which makes me believe adding assertion will be a better approach to avoid future changes affecting group max size logic.

The argument name maxSize suggests this should be a no-op if the group is already smaller. I'd just remove the check.

I'm worried about iterating the entire member map and not sure it is necessary.
I think we can assume that the number of members we remove is much smaller than the total number of members (i.e. the group is just slightly oversized).

Maybe this is cleaner:

• calculate how many members to remove.
• find N members out of members.keys
• call remove() on each (thereby not rewriting lots of logic, not moving around lots of stuff in memory, etc).

WDYT?

Yeah this is pretty inefficient.
My thinking is that we should run through this code path only on broker rolls with a new config. But regarding your other comment, shrinking may turn out to be completely unnecessary. I'll wait for @hachikuji to give his thoughts as well before removing it.

I guess this part of the logic is to add as many as members while making sure the leader is successfully joining the newMembers group. How about we just put the leader into the newMembers list first and continue the iteration without worrying about it? Note that double putting of leader shouldn't be a problem. This should make the logic simpler here as I perceived.

I agree that @gwenshap's idea is more efficient. Otherwise, if we are sticking to the existing logic, we might be able to leverage the retain(...) function and avoid creating a second hash map.

retain() is a cool function but I will go with @gwenshap's suggestion as it makes the updating of supportedProtocols easier

super nit: "maximum number of..." amount usually refers to things that aren't counted individually.

English! :)

I'm not sure I understand what the changes to receive records are doing here...

receiveRecords used to accept a topic parameter which it did not use at all, so I removed it

If you're asking in a more general sense - I added a way to handle exceptions in receiveRecords through a callback. This was to ease implementation since I needed to run receiveRecords() in parallel. I played a lot with these tests, now I clearly see that callback logic isn't needed.

seems to be consistently failing for me with:
java.util.concurrent.TimeoutException: Futures timed out after [1 millisecond]

I do not see any failures. Passed 100/100.

This is now failing for me too.

Aha, this is why the tests are failing - left over from the merge with KIP-394 ;)

Would this be easier if we name a constructor as GroupMaxSizeReachedException(String groupId)?

I also like what @abbccdda suggests (even though most of the existing exception constructors just take a message string).

I also wonder if the message ... at full capacity ... provides enough context to users who may not be aware of this new feature. I think a clearer message that emphasizes on number of consumers in the group already at the configured maximum would be less confusing.

I also wonder if the message ... at full capacity ... provides enough context to users who may not be aware of this new feature. I think a clearer message that emphasizes on number of consumers in the group already at the configured maximum would be less confusing.

This sounds good to me.

I was thinking about a groupId as a parameter as well and was wondering how to configure it in Errors.java for when the server uses that. I've updated that, please see it again.

Should we just use assert here? As long as the change is robust, this should never happen which makes me believe adding assertion will be a better approach to avoid future changes affecting group max size logic.

Could we just do a two-element (memberId, member) in the foreach iteration to avoid extra definition here?

Since leaderWasAdded is only used twice, we might just want to use !leader.isEmpty for straightforward logic lookup.

Just want to understand here: (I'm not 100% sure) does a plain assignment for members = newMembers affect the final result?

s/maxSize/groupMaxSize

I guess this part of the logic is to add as many as members while making sure the leader is successfully joining the newMembers group. How about we just put the leader into the newMembers list first and continue the iteration without worrying about it? Note that double putting of leader shouldn't be a problem. This should make the logic simpler here as I perceived.

Do we need to create a separate function? Since checkExceptionDuringRebalance is used only once.

I initially had it separate as it had a lot of lines of code but now that it's less than 10 lines I think it's a good idea to merge it back in

By kicking out the extra members, like Jason said those members lose their opportunities to commit the last batch. However on the other hand, this should only happen when the server operator believes that an absurd group size is configured that should be prevented. In this sense, the group is interpreted as over capacity, so allowing rest of consumers take over the jobs should be very light lifting, which I don't see an obvious negative impact. Thoughts?

Let's just remove the return type.

Perhaps a log message (warning) would suffice?

This test fails for me about 50% of the time when I run it locally. I wonder if you see this flakiness on your side.
Sample error messages:

java.lang.AssertionError: expected:<102> but was:<105>

org.scalatest.junit.JUnitTestFailedError: Expected to only receive one exception of type
class org.apache.kafka.common.errors.GroupMaxSizeReachedExceptionduring consumption.
Received: ArrayBuffer(org.apache.kafka.clients.consumer.CommitFailedException: Commit
cannot be completed since the group has already rebalanced and assigned the partitions to
another member. This means that the time between subsequent calls to poll() was longer
than the configured max.poll.interval.ms, which typically implies that the poll loop is
spending too much time message processing. You can address this either by increasing
max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with
max.poll.records.)

java.lang.AssertionError: expected:<102> but was:<136>

Not 50% of the time but I do see some flakiness now. What do you think would be a good way to go around this? Checking for at least X records rather than exact?

After the changes, this test has passed 50/50 times for me

Yes, it now passes for me too (20/20). Thanks!

I also like what @abbccdda suggests (even though most of the existing exception constructors just take a message string).

I also wonder if the message ... at full capacity ... provides enough context to users who may not be aware of this new feature. I think a clearer message that emphasizes on number of consumers in the group already at the configured maximum would be less confusing.

I agree that @gwenshap's idea is more efficient. Otherwise, if we are sticking to the existing logic, we might be able to leverage the retain(...) function and avoid creating a second hash map.

I do not see any failures. Passed 100/100.

nit: ... as many members, or even ... already has the configured maximum number of members.

It seems this test doesn't properly handle the failure.

This is now failing for me too.

Yes, it now passes for me too (20/20). Thanks!

Since this is non-retriable exception, we could consolidate it into code block L546 as fatal error.

We don't need to check anything here I guess, we should just need to attempt removing the member id (because no member.id match will do a no-op IMO.

I think we just need to kick out real members until the groupIsFull is met. Probably we could just use groupIsFull to keep rejecting new members, while doesn't care about pending members to make the handling logic simpler.

I don't think that would work.
If we were to detect that groupIsOverCapacity() and start kicking out members until groupIsFull, we might kick out members before they had a chance to commit offsets.

When we receive aJoinGroup request from a member, we know that it has attempted to commit offsets. (see ConsumerCoordinator#onJoinPrepare)