chore(acl): Improve error handling

Signed-off-by: Viacheslav Vasilyev <avoidik@gmail.com>

Author: avoidik

internal/clients/kafka/acl/acl.go

@@ -14,6 +14,14 @@ import (
 	"github.com/crossplane-contrib/provider-kafka/internal/clients/kafka"
 )
 
+// adminClient is the subset of kadm.Client methods used by this package.
+// *kadm.Client satisfies this interface without any changes to callers.
+type adminClient interface {
+	CreateACLs(ctx context.Context, b *kadm.ACLBuilder) (kadm.CreateACLsResults, error)
+	DeleteACLs(ctx context.Context, b *kadm.ACLBuilder) (kadm.DeleteACLsResults, error)
+	DescribeACLs(ctx context.Context, b *kadm.ACLBuilder) (kadm.DescribeACLsResults, error)
+}
+
 // AccessControlList is a holistic representation of a Kafka ACL with configurable
 // fields
 type AccessControlList struct {
@@ -58,7 +66,7 @@ func buildACLBuilder(accessControlList *AccessControlList) (*kadm.ACLBuilder, er
 }
 
 // List lists all the ACLs in Kafka
-func List(ctx context.Context, cl *kadm.Client, accessControlList *AccessControlList) (*AccessControlList, error) {
+func List(ctx context.Context, cl adminClient, accessControlList *AccessControlList) (*AccessControlList, error) {
 	ab, err := buildACLBuilder(accessControlList)
 	if err != nil {
 		return nil, err
@@ -68,7 +76,13 @@ func List(ctx context.Context, cl *kadm.Client, accessControlList *AccessControl
 	if err != nil {
 		return nil, fmt.Errorf("describe ACLs failed: %w", err)
 	}
-	if len(resp) == 0 || len(resp[0].Described) == 0 {
+	if len(resp) == 0 {
+		return nil, nil
+	}
+	if resp[0].Err != nil {
+		return nil, fmt.Errorf("describe ACLs failed: %w", resp[0].Err)
+	}
+	if len(resp[0].Described) == 0 {
 		return nil, nil
 	}
 
@@ -84,7 +98,7 @@ func List(ctx context.Context, cl *kadm.Client, accessControlList *AccessControl
 }
 
 // Create creates an ACL from the Kafka side
-func Create(ctx context.Context, cl *kadm.Client, accessControlList *AccessControlList) error {
+func Create(ctx context.Context, cl adminClient, accessControlList *AccessControlList) error {
 	ab, err := buildACLBuilder(accessControlList)
 	if err != nil {
 		return err
@@ -94,22 +108,31 @@ func Create(ctx context.Context, cl *kadm.Client, accessControlList *AccessContr
 	if err != nil {
 		return err
 	}
-	if len(resp) == 0 || len(resp[0].Principal) == 0 {
+	if len(resp) == 0 {
 		return errors.New("no create response for acl")
 	}
+	if resp[0].Err != nil {
+		return fmt.Errorf("create ACL failed: %w", resp[0].Err)
+	}
 
 	return nil
 }
 
 // Delete deletes an ACL from the Kafka side
-func Delete(ctx context.Context, cl *kadm.Client, accessControlList *AccessControlList) error {
+func Delete(ctx context.Context, cl adminClient, accessControlList *AccessControlList) error {
 	ab, err := buildACLBuilder(accessControlList)
 	if err != nil {
 		return err
 	}
 
-	_, err = cl.DeleteACLs(ctx, ab)
-	return err
+	resp, err := cl.DeleteACLs(ctx, ab)
+	if err != nil {
+		return err
+	}
+	if len(resp) > 0 && resp[0].Err != nil {
+		return fmt.Errorf("delete ACL failed: %w", resp[0].Err)
+	}
+	return nil
 }
 
 // ConvertToJSON performs a json marshalling for ACLs

internal/clients/kafka/acl/acl_test.go

@@ -8,12 +8,36 @@ import (
 	"time"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/twmb/franz-go/pkg/kadm"
+	"github.com/twmb/franz-go/pkg/kerr"
 	"k8s.io/apimachinery/pkg/util/json"
 
 	"github.com/crossplane-contrib/provider-kafka/apis/v1alpha1"
 	"github.com/crossplane-contrib/provider-kafka/internal/clients/kafka"
 )
 
+// fakeACLAdmin is an in-process implementation of adminClient for unit tests.
+type fakeACLAdmin struct {
+	createResults kadm.CreateACLsResults
+	createErr     error
+	describeResults kadm.DescribeACLsResults
+	describeErr   error
+	deleteResults kadm.DeleteACLsResults
+	deleteErr     error
+}
+
+func (f *fakeACLAdmin) CreateACLs(_ context.Context, _ *kadm.ACLBuilder) (kadm.CreateACLsResults, error) {
+	return f.createResults, f.createErr
+}
+
+func (f *fakeACLAdmin) DescribeACLs(_ context.Context, _ *kadm.ACLBuilder) (kadm.DescribeACLsResults, error) {
+	return f.describeResults, f.describeErr
+}
+
+func (f *fakeACLAdmin) DeleteACLs(_ context.Context, _ *kadm.ACLBuilder) (kadm.DeleteACLsResults, error) {
+	return f.deleteResults, f.deleteErr
+}
+
 var dataTesting = []byte(os.Getenv("KAFKA_CONFIG"))
 
 var baseACL = AccessControlList{
@@ -477,6 +501,81 @@ func TestList(t *testing.T) {
 	}
 }
 
+// --- Unit tests for broker-level error propagation (no real Kafka needed) ---
+
+func TestCreateBrokerError(t *testing.T) {
+	cl := &fakeACLAdmin{
+		createResults: kadm.CreateACLsResults{
+			{Principal: "User:alice", Err: kerr.ClusterAuthorizationFailed},
+		},
+	}
+	err := Create(context.Background(), cl, &baseACL)
+	if err == nil {
+		t.Fatal("Create() expected error for broker-level Err, got nil")
+	}
+}
+
+func TestCreateEmptyResponse(t *testing.T) {
+	cl := &fakeACLAdmin{createResults: kadm.CreateACLsResults{}}
+	err := Create(context.Background(), cl, &baseACL)
+	if err == nil {
+		t.Fatal("Create() expected error for empty response, got nil")
+	}
+}
+
+func TestListBrokerError(t *testing.T) {
+	cl := &fakeACLAdmin{
+		describeResults: kadm.DescribeACLsResults{
+			{Err: kerr.ClusterAuthorizationFailed},
+		},
+	}
+	got, err := List(context.Background(), cl, &baseACL)
+	if err == nil {
+		t.Fatal("List() expected error for broker-level Err, got nil")
+	}
+	if got != nil {
+		t.Errorf("List() expected nil result on error, got %v", got)
+	}
+}
+
+func TestListNotFound(t *testing.T) {
+	cl := &fakeACLAdmin{
+		describeResults: kadm.DescribeACLsResults{
+			{Described: kadm.DescribedACLs{}},
+		},
+	}
+	got, err := List(context.Background(), cl, &baseACL)
+	if err != nil {
+		t.Fatalf("List() unexpected error: %v", err)
+	}
+	if got != nil {
+		t.Errorf("List() expected nil for no matching ACL, got %v", got)
+	}
+}
+
+func TestListEmptyResponse(t *testing.T) {
+	cl := &fakeACLAdmin{describeResults: kadm.DescribeACLsResults{}}
+	got, err := List(context.Background(), cl, &baseACL)
+	if err != nil {
+		t.Fatalf("List() unexpected error: %v", err)
+	}
+	if got != nil {
+		t.Errorf("List() expected nil for empty response, got %v", got)
+	}
+}
+
+func TestDeleteBrokerError(t *testing.T) {
+	cl := &fakeACLAdmin{
+		deleteResults: kadm.DeleteACLsResults{
+			{Err: kerr.ClusterAuthorizationFailed},
+		},
+	}
+	err := Delete(context.Background(), cl, &baseACL)
+	if err == nil {
+		t.Fatal("Delete() expected error for broker-level Err, got nil")
+	}
+}
+
 // TestListAtProviderNotFound verifies that List returns nil when the ACL does not exist.
 func TestListAtProviderNotFound(t *testing.T) {
 	if len(dataTesting) == 0 {