Hi Luke,
Thanks for your questions.
> 1. We will store the "targetAssignment" into log now. But as we know,
there's max batch size limit (default 1MB), which means, we cannot support
1M partitions in one group (actually, it should be less than 60k partitions
since we'll store {topicID+partition id}) by default now. How will we
handle that? Do we expect users to adjust the max batch size to support
large partitions in groups, which we don't need this change for old
protocol?
That's right. I have a few ideas to remove this limitation in the
future but I decided to keep them for future improvement. The KIP is
large enough and as the current protocol suffers from the exact same
limitation, it is not a regression.
For the future, my thinking is to split the assignment and to only
write deltas to the log instead of re-writing all of it. We would need
to use transactions for this in the coordinator (similarly to
KIP-868). The challenge is that we have to ensure that those deltas
are all written or completely roll backed. Otherwise, we would have a
weird state with the compaction. This needs obviously more thinking.
> I'm wondering why we should persist the "targetAssignment" data? If we want
to work for coordinator failover, could the new coordinator try to request
for currently owned partitions from each consumer when failed over? I'm not
sure if the consumer will auto send owned partitions to the new
coordinator. If not, maybe we can return an error to client
ConsumerGroupHeartbeat API with REQUIRE_OWNED_PARTITION error, and ask
client to append the currently owned partitions to new coordinates for new
assignment computation. Does that make sense?
The entire reconciliation process depends on it so if we lose it
during a failover, members could be in a weird state. For instance,
they could be in the middle of a transition from their current
assignment to their new target and thus would be blocked. Relying on
members to reconstruct it back does not really work because they don't
have all the information to do so (e.g. new metadata) so we would have
to recompute a new one. This implies that we need to get the owned
partitions from all members and that would take a few seconds until
all members come back in the best case, up to the session timeout in
the worst case. Imagine that a member joins or fails during this time,
the whole process would be stuck. I am afraid storing it is the best
way here.
Best,
David
On Fri, Oct 14, 2022 at 5:11 AM Luke Chen <show...@gmail.com> wrote:
>
> Hi David,
>
> A few more questions:
> 1. We will store the "targetAssignment" into log now. But as we know,
> there's max batch size limit (default 1MB), which means, we cannot support
> 1M partitions in one group (actually, it should be less than 60k partitions
> since we'll store {topicID+partition id}) by default now. How will we
> handle that? Do we expect users to adjust the max batch size to support
> large partitions in groups, which we don't need this change for old
> protocol?
>
> I'm wondering why we should persist the "targetAssignment" data? If we want
> to work for coordinator failover, could the new coordinator try to request
> for currently owned partitions from each consumer when failed over? I'm not
> sure if the consumer will auto send owned partitions to the new
> coordinator. If not, maybe we can return an error to client
> ConsumerGroupHeartbeat API with REQUIRE_OWNED_PARTITION error, and ask
> client to append the currently owned partitions to new coordinates for new
> assignment computation. Does that make sense?
>
> Luke
>
> On Fri, Oct 14, 2022 at 12:22 AM Jun Rao <j...@confluent.io.invalid> wrote:
>
> > Hi, David,
> >
> > Thanks for the reply and the updated KIP. A few more comments on the
> > interfaces and the protocols.
> >
> > 60. On the consumer side, do we need both PartitionAssignor.onAssignment
> > and ConsumerRebalanceListener.onPartitionsAssigned? My understanding is
> > that the former was added for cooperative rebalance, which is now handled
> > by the coordinator. If we do need both, should we make them more consistent
> > (e.g. topic name vs topic id, list vs set vs collection)?
> >
> > 61. group.local.assignors: Could we make it clear that it's the full class
> > name that implements PartitionAssignor?
> >
> > 62. MemberAssignment: It currently has the following method.
> > public Set<TopicPartition> topicPartitions()
> > We are adding List<TopicIdPartition> ownedPartitions. Should we keep the
> > naming and the return type consistent?
> >
> > 63. MemberAssignment.error: should that be reason?
> >
> > 64. group.remote.assignor: The client may not know what assignors the
> > broker supports. Should we default this to what the broker determines (e.g.
> > first assignor listed in group.consumer.assignors)?
> >
> > 65. After the text "When A heartbeats again and acknowledges the
> > revocation, the group coordinator transitions him to epoch 2 and releases
> > foo-2.", we have the following.
> > B - epoch=2, partitions=[foo-2], pending-partitions=[]
> > Should foo-2 be in pending-partitions?
> >
> > 66. In the Online Migration example, is the first occurence of "C -
> > epoch=23, partitions=[foo-2, foo-5, foo-4], pending-partitions=[]" correct?
> > It seems that should happen after C receives a SyncGroup response? If so,
> > subsequent examples have the same issue.
> >
> > 67. ConsumerGroupHeartbeatRequest.RebalanceTimeoutMs : Which config
> > controls this? How is this used by the group coordinator since there is no
> > sync barrier anymore?
> >
> > 68. ConsumerGroupHeartbeatResponse:
> > 68.1 AssignedTopicPartitions and PendingTopicPartitions are of type
> > []TopicPartition. Should they be TopicPartitions?
> > 68.2 Assignment.error. Should we also have an errorMessage field?
> >
> > 69. ConsumerGroupPrepareAssignmentResponse.Members.Assignor: Should it
> > include the selected assignor name?
> >
> > 70. ConsumerGroupInstallAssignmentRequest.GroupEpoch: Should we let the
> > client set this? Intuitively, it seems the coordinator should manage the
> > group epoch.
> >
> > 71. ConsumerGroupDescribeResponse:
> > 71.1 Members.Assignment.Partitions. Should we include the topic name too
> > since it's convenient for building tools? Ditto for TargetAssignment.
> > 71.2 Members: Should we include SubscribedTopicRegex too?
> >
> > 72. OffsetFetchRequest: Is GenerationIdOrMemberEpoch needed since tools may
> > also want to issue this request?
> >
> > Thanks,
> >
> > Jun
> >
