Merge pull request #2241 from crossplane-contrib/backport-2240-to-release-0.54

[Backport release-0.54] fix(kafkacluster): sort broker endpoints strings for consistent connectionDetails

Author: kkendzia

pkg/controller/kafka/cluster/setup.go

@@ -18,6 +18,7 @@ package cluster
 
 import (
 	"context"
+	"sort"
 	"strings"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -159,8 +160,8 @@ func (u *hooks) postObserve(ctx context.Context, cr *svcapitypes.Cluster, obj *s
 	obs.ConnectionDetails = managed.ConnectionDetails{
 		// see: https://docs.aws.amazon.com/msk/latest/developerguide/client-access.html
 		// no endpoint informations available in DescribeClusterOutput only endpoints for zookeeperPlain/Tls
-		"zookeeperEndpointPlain": []byte(pointer.StringValue(obj.ClusterInfo.ZookeeperConnectString)),
-		"zookeeperEndpointTls":   []byte(pointer.StringValue(obj.ClusterInfo.ZookeeperConnectStringTls)),
+		"zookeeperEndpointPlain": []byte(sortBootstrapBrokerString(pointer.StringValue(obj.ClusterInfo.ZookeeperConnectString))),
+		"zookeeperEndpointTls":   []byte(sortBootstrapBrokerString(pointer.StringValue(obj.ClusterInfo.ZookeeperConnectStringTls))),
 	}
 
 	switch pointer.StringValue(obj.ClusterInfo.State) {
@@ -174,13 +175,13 @@ func (u *hooks) postObserve(ctx context.Context, cr *svcapitypes.Cluster, obj *s
 		if err != nil {
 			return obs, errorutils.Wrap(err, errGetBootstrapBrokers)
 		}
-		obs.ConnectionDetails["clusterEndpointPlain"] = []byte(pointer.StringValue(endpoints.BootstrapBrokerString))
-		obs.ConnectionDetails["clusterEndpointTls"] = []byte(pointer.StringValue(endpoints.BootstrapBrokerStringTls))
-		obs.ConnectionDetails["clusterEndpointIAM"] = []byte(pointer.StringValue(endpoints.BootstrapBrokerStringSaslIam))
-		obs.ConnectionDetails["clusterEndpointScram"] = []byte(pointer.StringValue(endpoints.BootstrapBrokerStringSaslScram))
-		obs.ConnectionDetails["clusterEndpointTlsPublic"] = []byte(pointer.StringValue(endpoints.BootstrapBrokerStringPublicTls))
-		obs.ConnectionDetails["clusterEndpointIAMPublic"] = []byte(pointer.StringValue(endpoints.BootstrapBrokerStringPublicSaslIam))
-		obs.ConnectionDetails["clusterEndpointScramPublic"] = []byte(pointer.StringValue(endpoints.BootstrapBrokerStringPublicSaslScram))
+		obs.ConnectionDetails["clusterEndpointPlain"] = []byte(sortBootstrapBrokerString(pointer.StringValue(endpoints.BootstrapBrokerString)))
+		obs.ConnectionDetails["clusterEndpointTls"] = []byte(sortBootstrapBrokerString(pointer.StringValue(endpoints.BootstrapBrokerStringTls)))
+		obs.ConnectionDetails["clusterEndpointIAM"] = []byte(sortBootstrapBrokerString(pointer.StringValue(endpoints.BootstrapBrokerStringSaslIam)))
+		obs.ConnectionDetails["clusterEndpointScram"] = []byte(sortBootstrapBrokerString(pointer.StringValue(endpoints.BootstrapBrokerStringSaslScram)))
+		obs.ConnectionDetails["clusterEndpointTlsPublic"] = []byte(sortBootstrapBrokerString(pointer.StringValue(endpoints.BootstrapBrokerStringPublicTls)))
+		obs.ConnectionDetails["clusterEndpointIAMPublic"] = []byte(sortBootstrapBrokerString(pointer.StringValue(endpoints.BootstrapBrokerStringPublicSaslIam)))
+		obs.ConnectionDetails["clusterEndpointScramPublic"] = []byte(sortBootstrapBrokerString(pointer.StringValue(endpoints.BootstrapBrokerStringPublicSaslScram)))
 
 	case string(svcapitypes.ClusterState_CREATING):
 		cr.SetConditions(xpv1.Creating())
@@ -1192,3 +1193,17 @@ func generateOpenMonitorinInput(wanted *svcapitypes.OpenMonitoringInfo) *svcsdk.
 	return output
 
 }
+
+// sortBootstrapBrokerString splits the provided string at ",",
+// trims potential spaces, sorts the resulting list of endpoints
+// and returns this a consistent string value with the endpoints joined by "," again.
+func sortBootstrapBrokerString(endpoints string) string {
+	endpointList := strings.Split(endpoints, ",")
+	sort.Strings(endpointList)
+	// for potential edge case, where someday AWS string contains spaces
+	for i, endpoint := range endpointList {
+		endpointList[i] = strings.TrimSpace(endpoint)
+	}
+	sort.Strings(endpointList)
+	return strings.Join(endpointList, ",")
+}

pkg/controller/kafka/cluster/setup_test.go

@@ -0,0 +1,65 @@
+package cluster
+
+import (
+	"testing"
+)
+
+func TestSortBootstrapBrokerString(t *testing.T) {
+	type args struct {
+		input string
+	}
+
+	type want struct {
+		result string
+	}
+
+	cases := map[string]struct {
+		args
+		want
+	}{
+		"EmptyString": {
+			args: args{
+				input: "",
+			},
+			want: want{
+				result: "",
+			},
+		},
+
+		"SingleEndpointString": {
+			args: args{
+				input: "b-1.test.0abcde.c6.kafka.eu-central-1.amazonaws.com:9098",
+			},
+			want: want{
+				result: "b-1.test.0abcde.c6.kafka.eu-central-1.amazonaws.com:9098",
+			},
+		},
+
+		"MultipleEndpointsString": {
+			args: args{
+				input: "b-3.test.0abcde.c6.kafka.eu-central-1.amazonaws.com:9098,b-2.test.0abcde.c6.kafka.eu-central-1.amazonaws.com:9098,b-1.test.0abcde.c6.kafka.eu-central-1.amazonaws.com:9098",
+			},
+			want: want{
+				result: "b-1.test.0abcde.c6.kafka.eu-central-1.amazonaws.com:9098,b-2.test.0abcde.c6.kafka.eu-central-1.amazonaws.com:9098,b-3.test.0abcde.c6.kafka.eu-central-1.amazonaws.com:9098",
+			},
+		},
+
+		"WithSpacesEndpointsString": {
+			args: args{
+				input: "b-3.test.0abcde.c6.kafka.eu-central-1.amazonaws.com:9098, b-2.test.0abcde.c6.kafka.eu-central-1.amazonaws.com:9098, b-1.test.0abcde.c6.kafka.eu-central-1.amazonaws.com:9098",
+			},
+			want: want{
+				result: "b-1.test.0abcde.c6.kafka.eu-central-1.amazonaws.com:9098,b-2.test.0abcde.c6.kafka.eu-central-1.amazonaws.com:9098,b-3.test.0abcde.c6.kafka.eu-central-1.amazonaws.com:9098",
+			},
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			out := sortBootstrapBrokerString(tc.input)
+			if out != tc.result {
+				t.Errorf("For input '%s', expected '%s', but got '%s'", tc.input, tc.result, out)
+			}
+		})
+	}
+}