kgo: fix batch coordinator fetching

twmb · twmb · commit ca67da4d12a6 · 2022-12-12T21:27:28.000-07:00
v1.10.0 introduced batch coordinator fetching, but had a bug: when
fetching many coordinators in a batch, we would not use the proper
NodeID (we would use the top level NodeID, not the per-coordinator
NodeID).

This worked for non-batch coordinator fetching because we mirrored the
first coordinator to the top level response, if the request/response was
batched.
diff --git a/pkg/kgo/client.go b/pkg/kgo/client.go
@@ -1229,12 +1229,12 @@ func (cl *Client) loadCoordinators(ctx context.Context, typ int8, keys ...string
 					}
 				}
 			} else {
-				resp, _ := shard.Resp.(*kmsg.FindCoordinatorResponse)
+				resp := shard.Resp.(*kmsg.FindCoordinatorResponse)
 				for _, rc := range resp.Coordinators {
 					c, ok := key2load[rc.Key]
 					if ok {
 						c.err = kerr.ErrorForCode(rc.ErrorCode)
-						c.node = resp.NodeID
+						c.node = rc.NodeID
 					}
 				}
 			}

Original file line number	Diff line number	Diff line change
`@@ -1229,12 +1229,12 @@ func (cl *Client) loadCoordinators(ctx context.Context, typ int8, keys ...string`
`1229`	`1229`	`}`
`1230`	`1230`	`}`
`1231`	`1231`	`} else {`
`1232`		`- resp, _ := shard.Resp.(*kmsg.FindCoordinatorResponse)`
	`1232`	`+ resp := shard.Resp.(*kmsg.FindCoordinatorResponse)`
`1233`	`1233`	`for _, rc := range resp.Coordinators {`
`1234`	`1234`	`c, ok := key2load[rc.Key]`
`1235`	`1235`	`if ok {`
`1236`	`1236`	`c.err = kerr.ErrorForCode(rc.ErrorCode)`
`1237`		`- c.node = resp.NodeID`
	`1237`	`+ c.node = rc.NodeID`
`1238`	`1238`	`}`
`1239`	`1239`	`}`
`1240`	`1240`	`}`