kgo: bugfix batched FindCoordinator requests against older brokers

All credit to @douglasbouttell for the excellent find here.

Batched FindCoordinator requests against brokers that do not support
batched requests no longer worked at of 1.19.3. The problem was the
removal of pinReq, and well, a bug in the full removal.

This fixes the problem and adds a test test that fails without the fix.

I checked all linters I know of (opting into all revive, staticcheck,
gocritic, golangci-lint linters) and none of them catch this, for valid
reasons. It's hard to check.

I then went back and forth with Claude to generate some code that would
check it, and surprisingly came up with something. However, since it
generates a _few_ false positives and I don't want to learn enough of
the ast code to fully maintain the linter, I'm going to comment the code
on #1034 and this PR and if anybody wants to take it and run with it,
feel free. The one other variable the lint flagged was a false positive.

Closes #1034.

Author: twmb

pkg/kgo/client.go

@@ -3227,7 +3227,7 @@ func (*findCoordinatorSharder) shard(_ context.Context, kreq kmsg.Request, lastE
 		sreq.CoordinatorType = req.CoordinatorType
 		sreq.CoordinatorKey = key
 		issues = append(issues, issueShard{
-			req: req,
+			req: sreq,
 			pin: &pinReq{pinMax: true, max: 3},
 			any: true,
 		})

pkg/kgo/client_test.go

@@ -228,3 +228,32 @@ func TestPing(t *testing.T) {
 		t.Errorf("unable to ping: %v", err)
 	}
 }
+
+func TestIssue1034(t *testing.T) {
+	t.Parallel()
+
+	cl, _ := newTestClient()
+	defer cl.Close()
+
+	req := kmsg.NewPtrFindCoordinatorRequest()
+	req.CoordinatorType = 0
+	req.CoordinatorKeys = []string{"foo", "bar", "biz"}
+	resp, err := req.RequestWith(context.Background(), cl)
+	if err != nil {
+		t.Fatalf("unable to request coordinators: %v", err)
+	}
+	need := make(map[string]struct{})
+	for _, k := range req.CoordinatorKeys {
+		need[k] = struct{}{}
+	}
+	for _, c := range resp.Coordinators {
+		if c.Key == "" {
+			t.Errorf("response coordinator erroneously has an empty key")
+			continue
+		}
+		delete(need, c.Key)
+	}
+	if len(need) > 0 {
+		t.Errorf("coordinator key responses missing: %v", need)
+	}
+}