feat: support cluster tag updates (#72)

Issue [#2553](aws-controllers-k8s/community#2553)

Description of changes:
Support cluster tag updates and add e2e tests

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

Author: michaelhtm

apis/v1alpha1/ack-generate-metadata.yaml

@@ -1,13 +1,13 @@
 ack_generate_info:
-  build_date: "2025-07-15T03:10:59Z"
+  build_date: "2025-07-15T20:52:52Z"
   build_hash: c6808295bbb03aac999713ecf1f3aa5cd698a17e
   go_version: go1.24.4
   version: v0.49.0
 api_directory_checksum: 83a6f2d33e6781954a317da21e666cd5195403b7
 api_version: v1alpha1
 aws_sdk_go_version: v1.32.6
 generator_config_info:
-  file_checksum: 6355b93f40e251c132cf1ac3a6a29c89ac40de7c
+  file_checksum: 16526e586defd6129b39dec48357c841c9007aca
   original_file_name: generator.yaml
 last_modification:
   reason: API generation

apis/v1alpha1/generator.yaml

@@ -66,9 +66,6 @@ resources:
           resource: Secret
           service_name: secretsmanager
           path: Status.ACKResourceMetadata.ARN
-      Tags:
-        compare:
-          is_ignored: true
       BrokerNodeGroupInfo.ConnectivityInfo.PublicAccess.Type:
         go_tag: json:"type,omitempty"
       ZookeeperConnectString:
@@ -112,9 +109,6 @@ resources:
           operation: DescribeCluster
           path: ClusterInfo.CurrentVersion
         is_read_only: true
-    tags:
-      # TODO(jaypipes): Ignore tags for now... we will add support later
-      ignore: true
   ServerlessCluster:
     is_arn_primary_key: true
     renames:

generator.yaml

@@ -66,9 +66,6 @@ resources:
           resource: Secret
           service_name: secretsmanager
           path: Status.ACKResourceMetadata.ARN
-      Tags:
-        compare:
-          is_ignored: true
       BrokerNodeGroupInfo.ConnectivityInfo.PublicAccess.Type:
         go_tag: json:"type,omitempty"
       ZookeeperConnectString:
@@ -112,9 +109,6 @@ resources:
           operation: DescribeCluster
           path: ClusterInfo.CurrentVersion
         is_read_only: true
-    tags:
-      # TODO(jaypipes): Ignore tags for now... we will add support later
-      ignore: true
   ServerlessCluster:
     is_arn_primary_key: true
     renames:

pkg/resource/cluster/delta.go

@@ -21,6 +21,7 @@ import (
 	"time"
 
 	svcapitypes "github.com/aws-controllers-k8s/kafka-controller/apis/v1alpha1"
+	"github.com/aws-controllers-k8s/kafka-controller/pkg/sync"
 	ackcompare "github.com/aws-controllers-k8s/runtime/pkg/compare"
 	ackcondition "github.com/aws-controllers-k8s/runtime/pkg/condition"
 	ackerr "github.com/aws-controllers-k8s/runtime/pkg/errors"
@@ -159,6 +160,17 @@ func (rm *resourceManager) customUpdate(
 	}
 
 	switch {
+	case delta.DifferentAt("Spec.Tags"):
+		if err = sync.Tags(
+			ctx,
+			desired.ko.Spec.Tags, latest.ko.Spec.Tags,
+			(*string)(latest.ko.Status.ACKResourceMetadata.ARN),
+			convertToOrderedACKTags, rm.sdkapi, rm.metrics,
+		); err != nil {
+			return updatedRes, nil
+		}
+		return updatedRes, requeueAfterAsyncUpdate()
+
 	case delta.DifferentAt("Spec.ClientAuthentication"):
 		return rm.updateClientAuthentication(ctx, updatedRes, latest)
 	case delta.DifferentAt("Spec.AssociatedSCRAMSecrets"):
@@ -201,7 +213,7 @@ func (rm *resourceManager) updateNumberOfBrokerNodes(
 	if err != nil {
 		return latest, err
 	}
-	message := fmt.Sprintf("kafka is updating broker number of broker nodes")
+	message := "kafka is updating broker number of broker nodes"
 	ackcondition.SetSynced(updatedRes, corev1.ConditionFalse, &message, nil)
 
 	return desired, requeueAfterAsyncUpdate()
@@ -226,7 +238,7 @@ func (rm *resourceManager) updateBrokerType(
 	if err != nil {
 		return nil, err
 	}
-	message := fmt.Sprintf("kafka is updating broker instanceType")
+	message := "kafka is updating broker instanceType"
 	ackcondition.SetSynced(updatedRes, corev1.ConditionFalse, &message, nil)
 
 	return desired, requeueAfterAsyncUpdate()
@@ -418,7 +430,6 @@ func (rm *resourceManager) getAssociatedScramSecrets(
 // BatchAssociateScramSecret or Disassociate. It represents the
 // secretArns that could not be associated and the reason
 type unprocessedSecrets struct {
-	error
 	errorCodes    []string
 	errorMessages []string
 	secretArns    []string

pkg/resource/cluster/hooks.go

@@ -197,3 +197,32 @@ def test_crud(self, simple_cluster):
 
         latest_secrets = cluster.get_associated_scram_secrets(cluster_arn)
         assert len(latest_secrets) == 0
+
+        # Test adding tags to the cluster
+        updates = {
+            "spec": {
+                "tags": {
+                    "tag1": "val1",
+                    "tag2": "val2"
+                }
+            }
+        }
+        k8s.patch_custom_resource(ref, updates)
+        time.sleep(CHECK_STATUS_WAIT_SECONDS)
+        assert k8s.wait_on_condition(
+            ref,
+            "ACK.ResourceSynced",
+            "True",
+            wait_periods=MODIFY_WAIT_AFTER_SECONDS,
+        )
+
+        cr = k8s.get_resource(ref)
+        latest_tags = cluster.get_tags(cluster_arn)
+        desired_tags = cr['spec']['tags']
+        tags.assert_ack_system_tags(
+            tags=latest_tags,
+        )
+        tags.assert_equal_without_ack_tags(
+            expected=desired_tags,
+            actual=latest_tags,
+        )