From e209bb68fca5f4c901b245172cec5401525eb2a5 Mon Sep 17 00:00:00 2001 From: Travis Bischel Date: Tue, 31 Oct 2023 20:32:53 -0600 Subject: [PATCH] kgo: pin AddPartitionsToTxn to v3 when using one transaction KIP-890 has been updated such that v3 must be used by clients. We will pin to v3 unless multiple transactions are being added, or unless any transaction is verify only. Closes #609. --- pkg/kgo/client.go | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/pkg/kgo/client.go b/pkg/kgo/client.go index d11befa9..07bb9adc 100644 --- a/pkg/kgo/client.go +++ b/pkg/kgo/client.go @@ -3420,10 +3420,17 @@ func (cl *addPartitionsToTxnSharder) shard(ctx context.Context, kreq kmsg.Reques var issues []issueShard for id, req := range brokerReqs { - issues = append(issues, issueShard{ - req: req, - broker: id, - }) + if len(req.Transactions) <= 1 || len(req.Transactions) == 1 && !req.Transactions[0].VerifyOnly { + issues = append(issues, issueShard{ + req: &pinReq{Request: req, pinMax: true, max: 3}, + broker: id, + }) + } else { + issues = append(issues, issueShard{ + req: req, + broker: id, + }) + } } for _, unkerr := range unkerrs { issues = append(issues, issueShard{