support v1 saslhandshake with v0 saslauthenticate

Author: rhansen2

protocol/conn.go

@@ -84,6 +84,10 @@ type RawExchanger interface {
 	RawExchange(rw io.ReadWriter) (Message, error)
 }
 
+type DependentMessage interface {
+	DependsOnVersion() ApiKey
+}
+
 func (c *Conn) RoundTrip(msg Message) (Message, error) {
 	correlationID := atomic.AddInt32(&c.idgen, +1)
 	versions, _ := c.versions.Load().(map[ApiKey]int16)
@@ -93,7 +97,14 @@ func (c *Conn) RoundTrip(msg Message) (Message, error) {
 		p.Prepare(apiVersion)
 	}
 
-	if raw, ok := msg.(RawExchanger); ok && raw.Required(int(apiVersion)) {
+	dp, isDependent := msg.(DependentMessage)
+
+	requiredVersion := apiVersion
+	if isDependent {
+		requiredVersion = versions[dp.DependsOnVersion()]
+	}
+
+	if raw, ok := msg.(RawExchanger); ok && raw.Required(int(requiredVersion)) {
 		return raw.RawExchange(c)
 	}
 

protocol/saslauthenticate/saslauthenticate.go

@@ -15,6 +15,10 @@ type Request struct {
 	AuthBytes []byte `kafka:"min=v0,max=v1"`
 }
 
+func (*Request) DependsOnVersion() protocol.ApiKey {
+	return protocol.SaslHandshake
+}
+
 func (r *Request) RawExchange(rw io.ReadWriter) (protocol.Message, error) {
 	if err := r.writeTo(rw); err != nil {
 		return nil, err
@@ -63,4 +67,7 @@ type Response struct {
 
 func (r *Response) ApiKey() protocol.ApiKey { return protocol.SaslAuthenticate }
 
-var _ protocol.RawExchanger = (*Request)(nil)
+var (
+	_ protocol.RawExchanger     = (*Request)(nil)
+	_ protocol.DependentMessage = (*Request)(nil)
+)