fix(acl): Improve error handling

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

Author: avoidik

Makefile

@@ -224,6 +224,7 @@ kind-kafka-setup: $(HELM) $(KIND) $(KUBECTL)
 	@$(KUBECTL) wait --for=condition=ready -n kafka-cluster kafka/dev --timeout=300s
 	@$(KUBECTL) wait --for=condition=ready -n kafka-cluster kafkauser/user --timeout=300s
 	@$(KUBECTL) wait --for=condition=ready -n kafka-cluster kafkatopic/pre-existing --timeout=300s
+	@$(KUBECTL) annotate kafkatopic --all -n kafka-cluster strimzi.io/pause-reconciliation=true --overwrite
 	@$(INFO) Waiting for Kafka broker to be fully initialized...
 	@for i in $$(seq 1 30); do $(KUBECTL) -n kafka-cluster exec kafka-dev-0 -- bash -c 'echo "test" | kafka-console-producer.sh --broker-list localhost:9092 --topic pre-existing' > /dev/null 2>&1 && break || sleep 2; done
 	@$(INFO) Getting service IP and port

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,36 @@ 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)
+	}
+	if len(resp[0].Principal) == 0 {
 		return errors.New("no create response for acl")
 	}
 
 	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
+	}
+	for _, r := range resp {
+		if r.Err != nil {
+			return fmt.Errorf("delete ACL failed: %w", r.Err)
+		}
+	}
+	return nil
 }
 
 // ConvertToJSON performs a json marshalling for ACLs

internal/clients/kafka/acl/acl_test.go

@@ -8,12 +8,38 @@ import (
 	"time"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"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 +503,69 @@ func TestList(t *testing.T) {
 	}
 }
 
+// --- Unit tests for broker-level error propagation (no real Kafka needed) ---
+
+func TestCreateBrokerError(t *testing.T) {
+	t.Parallel()
+	cl := &fakeACLAdmin{
+		createResults: kadm.CreateACLsResults{
+			{Principal: "User:alice", Err: kerr.ClusterAuthorizationFailed},
+		},
+	}
+	err := Create(context.Background(), cl, &baseACL)
+	require.Error(t, err)
+}
+
+func TestCreateEmptyResponse(t *testing.T) {
+	t.Parallel()
+	cl := &fakeACLAdmin{createResults: kadm.CreateACLsResults{}}
+	err := Create(context.Background(), cl, &baseACL)
+	require.Error(t, err)
+}
+
+func TestListBrokerError(t *testing.T) {
+	t.Parallel()
+	cl := &fakeACLAdmin{
+		describeResults: kadm.DescribeACLsResults{
+			{Err: kerr.ClusterAuthorizationFailed},
+		},
+	}
+	got, err := List(context.Background(), cl, &baseACL)
+	require.Error(t, err)
+	assert.Nil(t, got)
+}
+
+func TestListNotFound(t *testing.T) {
+	t.Parallel()
+	cl := &fakeACLAdmin{
+		describeResults: kadm.DescribeACLsResults{
+			{Described: kadm.DescribedACLs{}},
+		},
+	}
+	got, err := List(context.Background(), cl, &baseACL)
+	require.NoError(t, err)
+	assert.Nil(t, got)
+}
+
+func TestListEmptyResponse(t *testing.T) {
+	t.Parallel()
+	cl := &fakeACLAdmin{describeResults: kadm.DescribeACLsResults{}}
+	got, err := List(context.Background(), cl, &baseACL)
+	require.NoError(t, err)
+	assert.Nil(t, got)
+}
+
+func TestDeleteBrokerError(t *testing.T) {
+	t.Parallel()
+	cl := &fakeACLAdmin{
+		deleteResults: kadm.DeleteACLsResults{
+			{Err: kerr.ClusterAuthorizationFailed},
+		},
+	}
+	err := Delete(context.Background(), cl, &baseACL)
+	require.Error(t, err)
+}
+
 // TestListAtProviderNotFound verifies that List returns nil when the ACL does not exist.
 func TestListAtProviderNotFound(t *testing.T) {
 	if len(dataTesting) == 0 {