Merge pull request #925 from rhansen2/cyx-fix-handle-error-in-response

fix: handle error in response

Author: rhansen2

protocol/describeconfigs/describeconfigs.go

@@ -1,6 +1,7 @@
 package describeconfigs
 
 import (
+	"fmt"
 	"strconv"
 
 	"github.com/segmentio/kafka-go/protocol"
@@ -87,11 +88,19 @@ func (r *Response) Merge(requests []protocol.Message, results []interface{}) (
 	response := &Response{}
 
 	for _, result := range results {
-		brokerResp := result.(*Response)
-		response.Resources = append(
-			response.Resources,
-			brokerResp.Resources...,
-		)
+		switch v := result.(type) {
+		case *Response:
+			response.Resources = append(
+				response.Resources,
+				v.Resources...,
+			)
+
+		case error:
+			return nil, v
+
+		default:
+			panic(fmt.Sprintf("unknown result type in Merge: %T", result))
+		}
 	}
 
 	return response, nil
@@ -123,6 +132,4 @@ type ResponseConfigSynonym struct {
 	ConfigSource int8   `kafka:"min=v1,max=v3"`
 }
 
-var (
-	_ protocol.BrokerMessage = (*Request)(nil)
-)
+var _ protocol.BrokerMessage = (*Request)(nil)

protocol/describeconfigs/describeconfigs_test.go

@@ -0,0 +1,77 @@
+package describeconfigs
+
+import (
+	"errors"
+	"io"
+	"reflect"
+	"testing"
+
+	"github.com/segmentio/kafka-go/protocol"
+)
+
+func TestResponse_Merge(t *testing.T) {
+	t.Run("happy path", func(t *testing.T) {
+		r := &Response{}
+
+		r1 := &Response{
+			Resources: []ResponseResource{
+				{ResourceName: "r1"},
+			},
+		}
+		r2 := &Response{
+			Resources: []ResponseResource{
+				{ResourceName: "r2"},
+			},
+		}
+
+		got, err := r.Merge([]protocol.Message{&Request{}}, []interface{}{r1, r2})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		want := &Response{
+			Resources: []ResponseResource{
+				{ResourceName: "r1"},
+				{ResourceName: "r2"},
+			},
+		}
+
+		if !reflect.DeepEqual(want, got) {
+			t.Fatalf("wanted response: \n%+v, got \n%+v", want, got)
+		}
+	})
+
+	t.Run("with errors", func(t *testing.T) {
+		r := &Response{}
+
+		r1 := &Response{
+			Resources: []ResponseResource{
+				{ResourceName: "r1"},
+			},
+		}
+
+		_, err := r.Merge([]protocol.Message{&Request{}}, []interface{}{r1, io.EOF})
+		if !errors.Is(err, io.EOF) {
+			t.Fatalf("wanted err io.EOF, got %v", err)
+		}
+	})
+
+	t.Run("panic with unexpected type", func(t *testing.T) {
+		defer func() {
+			msg := recover()
+			if msg != "unknown result type in Merge: string" {
+				t.Fatal("unexpected panic", msg)
+			}
+		}()
+		r := &Response{}
+
+		r1 := &Response{
+			Resources: []ResponseResource{
+				{ResourceName: "r1"},
+			},
+		}
+
+		_, _ = r.Merge([]protocol.Message{&Request{}}, []interface{}{r1, "how did a string got here"})
+		t.Fatal("did not panic")
+	})
+}