Improve already-exists handling and add e2e coverage

pkg/admin/errors.go

@@ -17,6 +17,7 @@ package admin
 import (
 	"errors"
 	"net"
+	"strings"
 
 	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/rest"
 )
@@ -51,12 +52,22 @@ const (
 
 // ErrorReason returns the HTTP status code for the error
 func ErrorReason(err error) Reason {
-	cliError, ok := err.(rest.Error)
-	if !ok {
-		// can't determine error reason as can't convert to a cli error
+	if err == nil {
 		return ReasonUnknown
 	}
-	return Reason(cliError.Code)
+
+	var restErrPtr *rest.Error
+	if errors.As(err, &restErrPtr) && restErrPtr != nil {
+		return Reason(restErrPtr.Code)
+	}
+
+	var restErr rest.Error
+	if errors.As(err, &restErr) {
+		return Reason(restErr.Code)
+	}
+
+	// can't determine error reason as can't convert to a cli error
+	return ReasonUnknown
 }
 
 // IsNotFound returns true if the error indicates the resource is not found on server
@@ -66,7 +77,25 @@ func IsNotFound(err error) bool {
 
 // IsAlreadyExist returns true if the error indicates the resource already exist
 func IsAlreadyExist(err error) bool {
-	return ErrorReason(err) == ReasonAlreadyExist
+	if err == nil {
+		return false
+	}
+
+	reason := ErrorReason(err)
+	if reason == ReasonAlreadyExist {
+		return true
+	}
+	if reason == ReasonInvalidParameter {
+		return isAlreadyExistsMessage(err)
+	}
+	if reason != ReasonUnknown {
+		return false
+	}
+	return isAlreadyExistsMessage(err)
+}
+
+func isAlreadyExistsMessage(err error) bool {
+	return strings.Contains(strings.ToLower(err.Error()), "already exist")
 }
 
 // IsInternalServerError returns true if the error indicates the resource already exist

pkg/admin/errors_test.go

@@ -0,0 +1,62 @@
+package admin
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/rest"
+)
+
+func TestIsAlreadyExist(t *testing.T) {
+	tests := []struct {
+		name     string
+		err      error
+		expected bool
+	}{
+		{
+			name:     "rest.Error with 409",
+			err:      rest.Error{Code: 409, Reason: "already exists"},
+			expected: true,
+		},
+		{
+			name:     "rest.Error pointer with 409",
+			err:      &rest.Error{Code: 409, Reason: "already exists"},
+			expected: true,
+		},
+		{
+			name:     "wrapped rest.Error with 409",
+			err:      fmt.Errorf("wrapped: %w", rest.Error{Code: 409, Reason: "already exists"}),
+			expected: true,
+		},
+		{
+			name:     "rest.Error with 412 and already exists reason",
+			err:      rest.Error{Code: 412, Reason: "This topic already exists"},
+			expected: true,
+		},
+		{
+			name:     "wrapped error with already exists message",
+			err:      fmt.Errorf("wrapped: %w", errors.New("This topic already exists")),
+			expected: true,
+		},
+		{
+			name:     "plain error with already exists message",
+			err:      errors.New("This topic already exists"),
+			expected: true,
+		},
+		{
+			name:     "unrelated error",
+			err:      errors.New("permission denied"),
+			expected: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := IsAlreadyExist(tt.err)
+			if result != tt.expected {
+				t.Fatalf("IsAlreadyExist() = %v, want %v", result, tt.expected)
+			}
+		})
+	}
+}

tests/operator/resources_test.go

@@ -19,7 +19,9 @@ import (
 	"encoding/json"
 	"fmt"
 	"slices"
+	"strings"
 
+	pulsarutils "github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"github.com/onsi/gomega/format"
@@ -126,6 +128,12 @@ var _ = Describe("Resources", func() {
 		ptopic2             *v1alphav1.PulsarTopic
 		ptopicName2         string = "test-topic2"
 		topicName2          string = "persistent://cloud/stage/user2"
+		pexistingTopic      *v1alphav1.PulsarTopic
+		pexistingTopicName  string = "test-existing-topic"
+		existingTopicName   string = "persistent://cloud/stage/existing"
+		pemptySchemaTopic   *v1alphav1.PulsarTopic
+		pemptySchemaName    string = "test-empty-schema-topic"
+		emptySchemaTopic    string = "persistent://cloud/stage/empty-schema"
 		ppermission         *v1alphav1.PulsarPermission
 		ppermissionName     string = "test-permission"
 		ppermission2        *v1alphav1.PulsarPermission
@@ -179,6 +187,15 @@ var _ = Describe("Resources", func() {
 		pnamespace = utils.MakePulsarNamespace(namespaceName, pnamespaceName, pulsarNamespaceName, pconnName, lifecyclePolicy)
 		ptopic = utils.MakePulsarTopic(namespaceName, ptopicName, topicName, pconnName, lifecyclePolicy)
 		ptopic2 = utils.MakePulsarTopic(namespaceName, ptopicName2, topicName2, pconnName, lifecyclePolicy)
+		pexistingTopic = utils.MakePulsarTopic(namespaceName, pexistingTopicName, existingTopicName, pconnName, lifecyclePolicy)
+		pemptySchemaTopic = utils.MakePulsarTopic(namespaceName, pemptySchemaName, emptySchemaTopic, pconnName, lifecyclePolicy)
+		pemptySchemaTopic.Spec.SchemaInfo = &v1alphav1.SchemaInfo{
+			Type:   "STRING",
+			Schema: "",
+		}
+		pemptySchemaTopic.Spec.Properties = map[string]string{
+			"schema-case": "empty",
+		}
 		roles := []string{"ironman"}
 		actions := []string{"produce", "consume", "functions"}
 		ppermission = utils.MakePulsarPermission(namespaceName, ppermissionName, topicName, pconnName, v1alphav1.PulsarResourceTypeTopic, roles, actions, v1alphav1.CleanUpAfterDeletion)
@@ -324,11 +341,19 @@ var _ = Describe("Resources", func() {
 		})
 
 		Context("PulsarTopic operation", Ordered, func() {
+			It("should create existing topic directly in Pulsar", func() {
+				createTopicInPulsar(existingTopicName)
+			})
+
 			It("should create the pulsartopic successfully", Label("Permissions"), func() {
 				err := k8sClient.Create(ctx, ptopic)
 				Expect(err == nil || apierrors.IsAlreadyExists(err)).Should(BeTrue())
 				err = k8sClient.Create(ctx, ptopic2)
 				Expect(err == nil || apierrors.IsAlreadyExists(err)).Should(BeTrue())
+				err = k8sClient.Create(ctx, pexistingTopic)
+				Expect(err == nil || apierrors.IsAlreadyExists(err)).Should(BeTrue())
+				err = k8sClient.Create(ctx, pemptySchemaTopic)
+				Expect(err == nil || apierrors.IsAlreadyExists(err)).Should(BeTrue())
 				err = k8sClient.Create(ctx, partitionedTopic)
 				Expect(err == nil || apierrors.IsAlreadyExists(err)).Should(BeTrue())
 			})
@@ -340,6 +365,18 @@ var _ = Describe("Resources", func() {
 					Expect(k8sClient.Get(ctx, tns, t)).Should(Succeed())
 					return v1alphav1.IsPulsarResourceReady(t)
 				}, "20s", "100ms").Should(BeTrue())
+				Eventually(func() bool {
+					t := &v1alphav1.PulsarTopic{}
+					tns := types.NamespacedName{Namespace: namespaceName, Name: pexistingTopicName}
+					Expect(k8sClient.Get(ctx, tns, t)).Should(Succeed())
+					return v1alphav1.IsPulsarResourceReady(t)
+				}, "20s", "100ms").Should(BeTrue())
+				Eventually(func() bool {
+					t := &v1alphav1.PulsarTopic{}
+					tns := types.NamespacedName{Namespace: namespaceName, Name: pemptySchemaName}
+					Expect(k8sClient.Get(ctx, tns, t)).Should(Succeed())
+					return v1alphav1.IsPulsarResourceReady(t)
+				}, "20s", "100ms").Should(BeTrue())
 				Eventually(func() bool {
 					t := &v1alphav1.PulsarTopic{}
 					tns := types.NamespacedName{Namespace: partitionedTopic.Namespace, Name: partitionedTopic.Name}
@@ -402,6 +439,30 @@ var _ = Describe("Resources", func() {
 				// }, "5s", "100ms").Should(Succeed())
 			})
 
+			It("should not create new schema versions for empty schema", func() {
+				initialCount := mustGetSchemaVersionCount(emptySchemaTopic)
+				Expect(initialCount).Should(BeNumerically(">", 0))
+
+				topic := &v1alphav1.PulsarTopic{}
+				tns := types.NamespacedName{Namespace: namespaceName, Name: pemptySchemaName}
+				Expect(k8sClient.Get(ctx, tns, topic)).Should(Succeed())
+				if topic.Spec.Properties == nil {
+					topic.Spec.Properties = map[string]string{}
+				}
+				topic.Spec.Properties["schema-update"] = "true"
+				Expect(k8sClient.Update(ctx, topic)).Should(Succeed())
+
+				Eventually(func() bool {
+					t := &v1alphav1.PulsarTopic{}
+					Expect(k8sClient.Get(ctx, tns, t)).Should(Succeed())
+					return v1alphav1.IsPulsarResourceReady(t)
+				}, "20s", "100ms").Should(BeTrue())
+
+				Consistently(func() int {
+					return mustGetSchemaVersionCount(emptySchemaTopic)
+				}, "10s", "1s").Should(Equal(initialCount))
+			})
+
 			It("should increase the partitions successfully", func() {
 				curTopic := &v1alphav1.PulsarTopic{}
 				Expect(k8sClient.Get(ctx, types.NamespacedName{
@@ -2633,6 +2694,20 @@ var _ = Describe("Resources", func() {
 				g.Expect(k8sClient.Delete(ctx, t)).Should(Succeed())
 			}).Should(Succeed())
 
+			Eventually(func(g Gomega) {
+				t := &v1alphav1.PulsarTopic{}
+				tns := types.NamespacedName{Namespace: namespaceName, Name: pexistingTopicName}
+				g.Expect(k8sClient.Get(ctx, tns, t)).Should(Succeed())
+				g.Expect(k8sClient.Delete(ctx, t)).Should(Succeed())
+			}).Should(Succeed())
+
+			Eventually(func(g Gomega) {
+				t := &v1alphav1.PulsarTopic{}
+				tns := types.NamespacedName{Namespace: namespaceName, Name: pemptySchemaName}
+				g.Expect(k8sClient.Get(ctx, tns, t)).Should(Succeed())
+				g.Expect(k8sClient.Delete(ctx, t)).Should(Succeed())
+			}).Should(Succeed())
+
 			Eventually(func(g Gomega) {
 				t := &v1alphav1.PulsarTopic{}
 				tns := types.NamespacedName{Namespace: namespaceName, Name: partitionedTopic.Name}
@@ -2696,3 +2771,52 @@ func updateTopicSchema(ctx context.Context, topicName, exampleSchemaDef string)
 	}
 	Expect(k8sClient.Update(ctx, topic)).Should(Succeed())
 }
+
+func createTopicInPulsar(topicName string) {
+	podName := fmt.Sprintf("%s-broker-0", brokerName)
+	containerName := fmt.Sprintf("%s-broker", brokerName)
+	stdout, stderr, err := utils.ExecInPod(k8sConfig, namespaceName, podName, containerName,
+		"./bin/pulsarctl -s http://localhost:8080 --token=$PROXY_TOKEN topics create --non-partitioned "+topicName)
+	if err == nil {
+		return
+	}
+
+	msg := strings.ToLower(stdout + " " + stderr)
+	if strings.Contains(msg, "already exists") || strings.Contains(msg, "already exist") ||
+		strings.Contains(msg, "conflict") || strings.Contains(msg, "409") {
+		return
+	}
+	Expect(err).Should(Succeed())
+}
+
+func mustGetSchemaVersionCount(topic string) int {
+	count, err := getSchemaVersionCount(topic)
+	Expect(err).Should(Succeed())
+	return count
+}
+
+func getSchemaVersionCount(topic string) (int, error) {
+	topicName, err := pulsarutils.GetTopicName(topic)
+	if err != nil {
+		return 0, err
+	}
+	endpoint := fmt.Sprintf("http://localhost:8080/admin/v2/schemas/%s/%s/%s/schemas",
+		topicName.GetTenant(), topicName.GetNamespace(), topicName.GetLocalName())
+	podName := fmt.Sprintf("%s-broker-0", brokerName)
+	containerName := fmt.Sprintf("%s-broker", brokerName)
+	stdout, stderr, err := utils.ExecInPod(k8sConfig, namespaceName, podName, containerName,
+		"curl -s -H \"Authorization: Bearer $PROXY_TOKEN\" "+endpoint)
+	if err != nil {
+		return 0, fmt.Errorf("fetch schema versions failed: %w (stderr: %s)", err, stderr)
+	}
+
+	var response struct {
+		Schemas []struct {
+			Version int64 `json:"version"`
+		} `json:"getSchemaResponses"`
+	}
+	if err := json.Unmarshal([]byte(stdout), &response); err != nil {
+		return 0, err
+	}
+	return len(response.Schemas), nil
+}