join&sync: fix for 2.5.0
Turns out that ProtocolType and Protocol need to be sent from all
members. Looking back at the KIP, I'm not sure why I ever thought this
was not the case.
Anyway, test works on 2.5.0 now.
@@ -892,12 +892,12 @@ start:
var plan balancePlan
var protocol string
- leader := resp.LeaderID == resp.MemberID
- if leader {
if resp.Protocol != nil {
protocol = *resp.Protocol
}
+ leader := resp.LeaderID == resp.MemberID
+ if leader {
g.cl.cfg.logger.Log(LogLevelInfo, "joined, balancing group",
"memberID", g.memberID,
"instanceID", g.instanceID,
@@ -940,21 +940,15 @@ func (g *groupConsumer) syncGroup(leader bool, plan balancePlan, protocol string
Generation: g.generation,
MemberID: g.memberID,
InstanceID: g.instanceID,
- ProtocolType: nil,
- Protocol: nil,
+ ProtocolType: &clientGroupProtocol,
+ Protocol: &protocol,
GroupAssignment: plan.intoAssignment(), // nil unless we are the leader
}
- if leader {
- req.ProtocolType = &clientGroupProtocol
- req.Protocol = &protocol
g.cl.cfg.logger.Log(LogLevelInfo, "syncing",
"protocol_type", clientGroupProtocol,
"protocol", protocol,
)
- } else {
- g.cl.cfg.logger.Log(LogLevelInfo, "syncing")
- }
kresp, err := g.cl.Request(g.ctx, &req)
if err != nil {
