Add altertopicpartitions error handler (#119)

segmentio · Feb 23, 2023 · 342c7c3 · 342c7c3
1 parent 7d30f24
commit 342c7c3
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 0 deletions.
diff --git a/pkg/admin/brokerclient.go b/pkg/admin/brokerclient.go
@@ -514,6 +514,9 @@ func (c *BrokerAdminClient) AssignPartitions(
 	if err = resp.Error; err != nil {
 		return err
 	}
+	if err = util.AlterPartitionReassignmentsRequestAssignmentError(resp.PartitionResults); err != nil {
+		return err
+	}
 
 	return err
 }

diff --git a/pkg/util/error.go b/pkg/util/error.go
@@ -34,3 +34,18 @@ func IncrementalAlterConfigsResponseResourcesError(resources []kafka.Incremental
 	}
 	return nil
 }
+
+func AlterPartitionReassignmentsRequestAssignmentError(results []kafka.AlterPartitionReassignmentsResponsePartitionResult) error {
+	errors := map[int]error{}
+	var hasErrors bool
+	for _, result := range results {
+		if result.Error != nil {
+			hasErrors = true
+			errors[result.PartitionID] = result.Error
+		}
+	}
+	if hasErrors {
+		return fmt.Errorf("%+v", errors)
+	}
+	return nil
+}