kgo: add MinVersions

This allows ensuring that you will not be bumped down if you are talking
to an old broker, which is necessary if you absolutely want to use some
new features.

Author: twmb

Diff for: pkg/kgo/broker.go

@@ -293,6 +293,15 @@ func (b *broker) handleReqs() {
 			pr.promise(nil, ErrUnknownRequestKey)
 			continue
 		}
+
+		// If cxn.versions[0] is non-negative, then we loaded API
+		// versions. If the version for this request is negative, we
+		// know the broker cannot handle this request.
+		if cxn.versions[0] >= 0 && cxn.versions[req.Key()] < 0 {
+			pr.promise(nil, ErrBrokerTooOld)
+			continue
+		}
+
 		ourMax := req.MaxVersion()
 		if b.cl.cfg.maxVersions != nil {
 			userMax := b.cl.cfg.maxVersions[req.Key()]
@@ -301,12 +310,24 @@ func (b *broker) handleReqs() {
 			}
 		}
 
-		// If brokerMax is negative, we have no api versions because
-		// the client is pinned pre 0.10.0 and we stick with our max.
+		// If brokerMax is negative at this point, we have no api
+		// versions because the client is pinned pre 0.10.0 and we
+		// stick with our max.
 		version := ourMax
 		if brokerMax := cxn.versions[req.Key()]; brokerMax >= 0 && brokerMax < ourMax {
 			version = brokerMax
 		}
+
+		// If the version now (after potential broker downgrading) is
+		// lower than we desire, we fail the request for the broker is
+		// too old.
+		if b.cl.cfg.minVersions != nil &&
+			int(req.Key()) < len(b.cl.cfg.minVersions) &&
+			version < b.cl.cfg.minVersions[req.Key()] {
+			pr.promise(nil, ErrBrokerTooOld)
+			continue
+		}
+
 		req.SetVersion(version) // always go for highest version
 
 		if !cxn.expiry.IsZero() && time.Now().After(cxn.expiry) {
@@ -661,7 +682,7 @@ func (cxn *brokerCxn) doSasl(authenticate bool) error {
 
 				if err = kerr.ErrorForCode(resp.ErrorCode); err != nil {
 					if resp.ErrorMessage != nil {
-						return fmt.Errorf("%s: %v", *resp.ErrorMessage, err)
+						return fmt.Errorf("%s: %w", *resp.ErrorMessage, err)
 					}
 					return err
 				}

Diff for: pkg/kgo/config.go

@@ -56,6 +56,7 @@ type cfg struct {
 
 	seedBrokers []string
 	maxVersions kversion.Versions
+	minVersions kversion.Versions
 
 	retryBackoff          func(int) time.Duration
 	retries               int
@@ -288,7 +289,7 @@ func SeedBrokers(seeds ...string) Opt {
 }
 
 // MaxVersions sets the maximum Kafka version to try, overriding the
-// internal unbounded (latest) versions.
+// internal unbounded (latest stable) versions.
 //
 // Note that specific max version pinning is required if trying to interact
 // with versions pre 0.10.0. Otherwise, unless using more complicated requests
@@ -300,6 +301,22 @@ func MaxVersions(versions kversion.Versions) Opt {
 	return clientOpt{func(cfg *cfg) { cfg.maxVersions = versions }}
 }
 
+// MinVersions sets the minimum Kafka version a request can be downgraded to,
+// overriding the default of the lowest version.
+//
+// This option is useful if you are issuing requests that you absolutely do not
+// want to be downgraded; that is, if you are relying on features in newer
+// requests, and you are not sure if your brokers can handle those features.
+// By setting a min version, if the client detects it needs to downgrade past
+// the version, it will instead avoid issuing the request.
+//
+// Unlike MaxVersions, if a request is issued that is unknown to the min
+// versions, the request is allowed. It is assumed that there is no lower bound
+// for that request.
+func MinVersions(versions kversion.Versions) Opt {
+	return clientOpt{func(cfg *cfg) { cfg.minVersions = versions }}
+}
+
 // RetryBackoff sets the backoff strategy for how long to backoff for a given
 // amount of retries, overriding the default exponential backoff that ranges
 // from 100ms min to 1s max.

Diff for: pkg/kgo/errors.go

@@ -12,7 +12,12 @@ var (
 
 	// ErrClientTooOld is returned when issuing request that are unknown or
 	// use an unknown version.
-	ErrClientTooOld = errors.New("client is too old; this client does not know what to do with this")
+	ErrClientTooOld = errors.New("client is too old; this client does not know what to do with this request")
+
+	// ErrBrokerTooOld is returned if a connection has loaded broker
+	// ApiVersions and knows that a broker cannot handle the request that
+	// is attempting to be issued.
+	ErrBrokerTooOld = errors.New("broker is too old; the broker has already indicated it will not know how to handle the request")
 
 	// ErrNoResp is the error used if Kafka does not reply to a topic or
 	// partition in a produce request. This error should never be seen.