kgo: allow empty groups when finding coordinator / fetching offsets

twmb · twmb · commit c6f7f9a4281c · 2022-12-12T21:23:00.000-07:00
v1.10.0 introduced a bug when switching to batch loadCoordinators: we would strip empty coordinator keys from the request. Technically, empty coordinator keys are valid. If a request contains an empty coordinator key (empty group), we would not load it and not set a map key for it and then using a field in the map value would panic. Now, we always request all coordinator keys. This also showed a bug in sharded OffsetFetch: fetching offsets for a group with no name would be stripped (now we will just forward whatever error happens). Closes #283.
diff --git a/pkg/kgo/client.go b/pkg/kgo/client.go
@@ -1154,9 +1154,7 @@ func (cl *Client) loadCoordinators(ctx context.Context, typ int8, keys ...string
 
 	toRequest := make(map[string]bool, len(keys)) // true == bypass the cache
 	for _, key := range keys {
-		if len(key) > 0 {
-			toRequest[key] = false
-		}
+		toRequest[key] = false
 	}
 
 	// For each of these keys, we have two cases:
@@ -2324,9 +2322,7 @@ func (cl *offsetFetchSharder) shard(ctx context.Context, kreq kmsg.Request, last
 	}
 	groups := make([]string, 0, len(req.Groups))
 	for i := range req.Groups {
-		if g := req.Groups[i].Group; len(g) > 0 {
-			groups = append(groups, req.Groups[i].Group)
-		}
+		groups = append(groups, req.Groups[i].Group)
 	}
 
 	coordinators := cl.loadCoordinators(ctx, coordinatorTypeGroup, groups...)
@@ -2487,9 +2483,7 @@ func (*findCoordinatorSharder) shard(_ context.Context, kreq kmsg.Request, lastE
 	}
 	req.CoordinatorKeys = req.CoordinatorKeys[:0]
 	for key := range uniq {
-		if len(key) > 0 {
-			req.CoordinatorKeys = append(req.CoordinatorKeys, key)
-		}
+		req.CoordinatorKeys = append(req.CoordinatorKeys, key)
 	}
 
 	splitReq := errors.Is(lastErr, errBrokerTooOld)

Original file line number	Diff line number	Diff line change
`@@ -1154,9 +1154,7 @@ func (cl *Client) loadCoordinators(ctx context.Context, typ int8, keys ...string`
`1154`	`1154`
`1155`	`1155`	`toRequest := make(map[string]bool, len(keys)) // true == bypass the cache`
`1156`	`1156`	`for _, key := range keys {`
`1157`		`- if len(key) > 0 {`
`1158`		`- toRequest[key] = false`
`1159`		`- }`
	`1157`	`+ toRequest[key] = false`
`1160`	`1158`	`}`
`1161`	`1159`
`1162`	`1160`	`// For each of these keys, we have two cases:`
`@@ -2324,9 +2322,7 @@ func (cl *offsetFetchSharder) shard(ctx context.Context, kreq kmsg.Request, last`
`2324`	`2322`	`}`
`2325`	`2323`	`groups := make([]string, 0, len(req.Groups))`
`2326`	`2324`	`for i := range req.Groups {`
`2327`		`- if g := req.Groups[i].Group; len(g) > 0 {`
`2328`		`- groups = append(groups, req.Groups[i].Group)`
`2329`		`- }`
	`2325`	`+ groups = append(groups, req.Groups[i].Group)`
`2330`	`2326`	`}`
`2331`	`2327`
`2332`	`2328`	`coordinators := cl.loadCoordinators(ctx, coordinatorTypeGroup, groups...)`
`@@ -2487,9 +2483,7 @@ func (*findCoordinatorSharder) shard(_ context.Context, kreq kmsg.Request, lastE`
`2487`	`2483`	`}`
`2488`	`2484`	`req.CoordinatorKeys = req.CoordinatorKeys[:0]`
`2489`	`2485`	`for key := range uniq {`
`2490`		`- if len(key) > 0 {`
`2491`		`- req.CoordinatorKeys = append(req.CoordinatorKeys, key)`
`2492`		`- }`
	`2486`	`+ req.CoordinatorKeys = append(req.CoordinatorKeys, key)`
`2493`	`2487`	`}`
`2494`	`2488`
`2495`	`2489`	`splitReq := errors.Is(lastErr, errBrokerTooOld)`