Misc. fixes in response to code review feedback

Author: yolken-segment

alterpartitionreassignments.go

@@ -35,11 +35,9 @@ type AlterPartitionReassignmentsRequestAssignment struct {
 
 // AlterPartitionReassignmentsResponse is a response from the AlterPartitionReassignments API.
 type AlterPartitionReassignmentsResponse struct {
-	// ErrorCode is set to a non-zero value if a top-level error was encountered.
-	ErrorCode int
-
-	// ErrorMessage describes the top-level error that occured.
-	ErrorMessage string
+	// Error is set to a non-nil value including the code and message if a top-level
+	// error was encountered when doing the update.
+	Error error
 
 	// PartitionResults contains the specific results for each partition.
 	PartitionResults []AlterPartitionReassignmentsResponsePartitionResult
@@ -51,11 +49,9 @@ type AlterPartitionReassignmentsResponsePartitionResult struct {
 	// PartitionID is the ID of the partition that was altered.
 	PartitionID int
 
-	// ErrorCode is set to a non-zero value if an error was encountered during the update.
-	ErrorCode int
-
-	// ErrorMessage describes the partition-specific error that occurred.
-	ErrorMessage string
+	// Error is set to a non-nil value including the code and message if an error was encountered
+	// during the update for this partition.
+	Error error
 }
 
 func (c *Client) AlterPartitionReassignments(
@@ -100,18 +96,16 @@ func (c *Client) AlterPartitionReassignments(
 	apiResp := protoResp.(*alterpartitionreassignments.Response)
 
 	resp := &AlterPartitionReassignmentsResponse{
-		ErrorCode:    int(apiResp.ErrorCode),
-		ErrorMessage: apiResp.ErrorMessage,
+		Error: makeError(apiResp.ErrorCode, apiResp.ErrorMessage),
 	}
 
 	for _, topicResult := range apiResp.Results {
 		for _, partitionResult := range topicResult.Partitions {
 			resp.PartitionResults = append(
 				resp.PartitionResults,
 				AlterPartitionReassignmentsResponsePartitionResult{
-					PartitionID:  int(partitionResult.PartitionIndex),
-					ErrorCode:    int(partitionResult.ErrorCode),
-					ErrorMessage: partitionResult.ErrorMessage,
+					PartitionID: int(partitionResult.PartitionIndex),
+					Error:       makeError(partitionResult.ErrorCode, partitionResult.ErrorMessage),
 				},
 			)
 		}

alterpartitionreassignments_test.go

@@ -41,13 +41,6 @@ func TestClientAlterPartitionReassignments(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if resp.ErrorCode != 0 {
-		t.Error(
-			"Bad error code in response",
-			"expected", 0,
-			"got", resp.ErrorCode,
-		)
-	}
 	if len(resp.PartitionResults) != 2 {
 		t.Error(
 			"Unexpected length of partition results",

apiversions.go

@@ -16,8 +16,8 @@ type ApiVersionsRequest struct {
 
 // ApiVersionsResponse is a response from the ApiVersions API.
 type ApiVersionsResponse struct {
-	// ErrorCode is set to a non-zero value if an error was encountered.
-	ErrorCode int
+	// Error is set to a non-nil value if an error was encountered.
+	Error error
 
 	// ApiKeys contains the specific details of each supported API.
 	ApiKeys []ApiVersionsResponseApiKey
@@ -54,7 +54,7 @@ func (c *Client) ApiVersions(
 	apiResp := protoResp.(*apiversions.Response)
 
 	resp := &ApiVersionsResponse{
-		ErrorCode: int(apiResp.ErrorCode),
+		Error: makeError(apiResp.ErrorCode, ""),
 	}
 	for _, apiKey := range apiResp.ApiKeys {
 		resp.ApiKeys = append(

apiversions_test.go

@@ -16,14 +16,6 @@ func TestClientApiVersions(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if resp.ErrorCode != 0 {
-		t.Error(
-			"Unexpected error code",
-			"expected", 0,
-			"got", resp.ErrorCode,
-		)
-	}
-
 	if len(resp.ApiKeys) == 0 {
 		t.Error(
 			"Unexpected apiKeys length",

describegroups.go

@@ -27,6 +27,10 @@ type DescribeGroupsResponse struct {
 
 // DescribeGroupsResponseGroup contains the response details for a single group.
 type DescribeGroupsResponseGroup struct {
+	// Error is set to a non-nil value if there was an error fetching the details
+	// for this group.
+	Error error
+
 	// GroupID is the ID of the group.
 	GroupID string
 
@@ -110,6 +114,7 @@ func (c *Client) DescribeGroups(
 
 	for _, apiGroup := range apiResp.Groups {
 		group := DescribeGroupsResponseGroup{
+			Error:      makeError(apiGroup.ErrorCode, ""),
 			GroupID:    apiGroup.GroupID,
 			GroupState: apiGroup.GroupState,
 			Members:    []DescribeGroupsResponseMember{},

describegroups_test.go

@@ -119,6 +119,13 @@ func TestClientDescribeGroups(t *testing.T) {
 		)
 	}
 	g := resp.Groups[0]
+	if g.Error != nil {
+		t.Fatal(
+			"Wrong error in group response",
+			"expected", nil,
+			"got", g.Error,
+		)
+	}
 
 	if g.GroupID != gid {
 		t.Error(

electleaders.go

@@ -25,8 +25,8 @@ type ElectLeadersRequest struct {
 
 // ElectLeadersResponse is a response from the ElectLeaders API.
 type ElectLeadersResponse struct {
-	// ErrorCode is set to a non-zero value if a top-level error occurred.
-	ErrorCode int
+	// ErrorCode is set to a non-nil value if a top-level error occurred.
+	Error error
 
 	// PartitionResults contains the results for each partition leader election.
 	PartitionResults []ElectLeadersResponsePartitionResult
@@ -37,11 +37,9 @@ type ElectLeadersResponsePartitionResult struct {
 	// Partition is the ID of the partition.
 	Partition int
 
-	// ErrorCode is set to a non-zero value if an error happened for this partition's election.
-	ErrorCode int
-
-	// ErrorMessage describes the partition-specific election error that occurred.
-	ErrorMessage string
+	// Error is set to a non-nil value if an error occurred electing leaders
+	// for this partition.
+	Error error
 }
 
 func (c *Client) ElectLeaders(
@@ -72,17 +70,16 @@ func (c *Client) ElectLeaders(
 	apiResp := protoResp.(*electleaders.Response)
 
 	resp := &ElectLeadersResponse{
-		ErrorCode: int(apiResp.ErrorCode),
+		Error: makeError(apiResp.ErrorCode, ""),
 	}
 
 	for _, topicResult := range apiResp.ReplicaElectionResults {
 		for _, partitionResult := range topicResult.PartitionResults {
 			resp.PartitionResults = append(
 				resp.PartitionResults,
 				ElectLeadersResponsePartitionResult{
-					Partition:    int(partitionResult.PartitionID),
-					ErrorCode:    int(partitionResult.ErrorCode),
-					ErrorMessage: partitionResult.ErrorMessage,
+					Partition: int(partitionResult.PartitionID),
+					Error:     makeError(partitionResult.ErrorCode, partitionResult.ErrorMessage),
 				},
 			)
 		}

electleaders_test.go

@@ -32,13 +32,6 @@ func TestClientElectLeaders(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if resp.ErrorCode != 0 {
-		t.Error(
-			"Bad error code in response",
-			"expected", 0,
-			"got", resp.ErrorCode,
-		)
-	}
 	if len(resp.PartitionResults) != 2 {
 		t.Error(
 			"Unexpected length of partition results",

incrementalalterconfigs.go

@@ -64,12 +64,9 @@ type IncrementalAlterConfigsResponse struct {
 // IncrementalAlterConfigsResponseResource contains the response details for a single resource
 // whose configs were updated.
 type IncrementalAlterConfigsResponseResource struct {
-	// ErrorCode is set to a non-zero value if there was an error updating the configs for this
-	// resource.
-	ErrorCode int
-
-	// ErrorMessage describes any errors that were encountered.
-	ErrorMessage string
+	// Error is set to a non-nil value if an error occurred while updating this specific
+	// config.
+	Error error
 
 	// ResourceType is the type of resource that was updated.
 	ResourceType ResourceType
@@ -125,8 +122,7 @@ func (c *Client) IncrementalAlterConfigs(
 		resp.Resources = append(
 			resp.Resources,
 			IncrementalAlterConfigsResponseResource{
-				ErrorCode:    int(res.ErrorCode),
-				ErrorMessage: res.ErrorMessage,
+				Error:        makeError(res.ErrorCode, res.ErrorMessage),
 				ResourceType: ResourceType(res.ResourceType),
 				ResourceName: res.ResourceName,
 			},

listgroups.go

@@ -16,6 +16,10 @@ type ListGroupsRequest struct {
 
 // ListGroupsResponse is a response from the ListGroups API.
 type ListGroupsResponse struct {
+	// Error is set to a non-nil value if a top-level error occurred while fetching
+	// groups.
+	Error error
+
 	// Groups contains the list of groups.
 	Groups []ListGroupsResponseGroup
 }
@@ -38,7 +42,9 @@ func (c *Client) ListGroups(
 		return nil, err
 	}
 	apiResp := protoResp.(*listgroups.Response)
-	resp := &ListGroupsResponse{}
+	resp := &ListGroupsResponse{
+		Error: makeError(apiResp.ErrorCode, ""),
+	}
 
 	for _, apiGroupInfo := range apiResp.Groups {
 		resp.Groups = append(resp.Groups, ListGroupsResponseGroup{

listgroups_test.go

@@ -87,6 +87,13 @@ func TestClientListGroups(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	if resp.Error != nil {
+		t.Fatal(
+			"Unexpected error in response",
+			"expected", nil,
+			"got", resp.Error,
+		)
+	}
 	hasGroup := false
 	for _, group := range resp.Groups {
 		if group.GroupID == gid {

protocol/alterpartitionreassignments/alterpartitionreassignments.go

@@ -8,12 +8,12 @@ func init() {
 
 // Detailed API definition: https://kafka.apache.org/protocol#The_Messages_AlterPartitionReassignments
 type Request struct {
-	TimeoutMs int32          `kafka:"min=v0,max=v0"`
-	Topics    []RequestTopic `kafka:"min=v0,max=v0"`
-
 	// We need at least one tagged field to indicate that this is a "flexible" message
 	// type.
 	_ struct{} `kafka:"min=v0,max=v0,tag"`
+
+	TimeoutMs int32          `kafka:"min=v0,max=v0"`
+	Topics    []RequestTopic `kafka:"min=v0,max=v0"`
 }
 
 type RequestTopic struct {
@@ -35,14 +35,14 @@ func (r *Request) Broker(cluster protocol.Cluster) (protocol.Broker, error) {
 }
 
 type Response struct {
+	// We need at least one tagged field to indicate that this is a "flexible" message
+	// type.
+	_ struct{} `kafka:"min=v0,max=v0,tag"`
+
 	ThrottleTimeMs int32            `kafka:"min=v0,max=v0"`
 	ErrorCode      int16            `kafka:"min=v0,max=v0"`
 	ErrorMessage   string           `kafka:"min=v0,max=v0,nullable"`
 	Results        []ResponseResult `kafka:"min=v0,max=v0"`
-
-	// We need at least one tagged field to indicate that this is a "flexible" message
-	// type.
-	_ struct{} `kafka:"min=v0,max=v0,tag"`
 }
 
 type ResponseResult struct {