Merge pull request #2217 from amaanx86/bugfix/persists-messages-flake

test: fix the flaky persists messages system test

Author: Zerpet

test/system/system_test.go

@@ -269,8 +269,14 @@ CONSOLE_LOG=new`
 		)
 
 		BeforeEach(func(ctx SpecContext) {
-			cluster = newRabbitmqCluster(namespace, "persistence-rabbit")
-			Expect(createRabbitmqCluster(ctx, rmqClusterClient, cluster)).To(Succeed())
+			// The AfterEach wait can be cut short by the spec deadline, so retry the create too.
+			Eventually(ctx, func() error {
+				cluster = newRabbitmqCluster(namespace, "persistence-rabbit")
+				return createRabbitmqCluster(ctx, rmqClusterClient, cluster)
+			}).
+				WithTimeout(clusterCreationTimeout).
+				WithPolling(2 * time.Second).
+				Should(Succeed())
 
 			waitForRabbitmqRunning(cluster)
 
@@ -283,7 +289,9 @@ CONSOLE_LOG=new`
 		})
 
 		AfterEach(func(ctx SpecContext) {
-			Expect(rmqClusterClient.Delete(ctx, cluster)).To(Succeed())
+			// A FlakeAttempts retry recreates this name milliseconds after we return, so block
+			// until the cluster is actually gone.
+			deleteRabbitmqClusterAndWait(ctx, rmqClusterClient, cluster)
 		})
 
 		It("persists messages", FlakeAttempts(3), func(ctx SpecContext) {
@@ -304,7 +312,8 @@ CONSOLE_LOG=new`
 				Expect(err).NotTo(HaveOccurred())
 				Expect(message.Payload).To(Equal("hello"))
 			})
-		}, SpecTimeout(time.Minute*3))
+			// The deadline covers the whole attempt, teardown included, and resets on each retry.
+		}, SpecTimeout(time.Minute*5))
 	})
 
 	Context("Persistence expansion", Label("persistence_expansion"), func() {

test/system/utils_test.go

@@ -50,6 +50,7 @@ import (
 	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/message"
 	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream"
 	corev1 "k8s.io/api/core/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
@@ -62,6 +63,11 @@ import (
 const podCreationTimeout = 10 * time.Minute
 const portReadinessTimeout = 1 * time.Minute
 const k8sQueryTimeout = 1 * time.Minute
+const clusterDeletionTimeout = 90 * time.Second
+
+// A create is rejected while a cluster of the same name is still terminating, so the retry needs
+// to cover a deletion.
+const clusterCreationTimeout = clusterDeletionTimeout
 
 type featureFlag struct {
 	Name  string
@@ -507,6 +513,32 @@ func createRabbitmqCluster(ctx context.Context, client client.Client, rabbitmqCl
 	return client.Create(ctx, rabbitmqCluster)
 }
 
+// deleteRabbitmqClusterAndWait deletes a RabbitmqCluster and waits for it to disappear. Delete only
+// marks it: the operator's finalizer keeps the object in Terminating for a few seconds, and
+// recreating the same name before then fails with "object is being deleted ... already exists".
+func deleteRabbitmqClusterAndWait(ctx context.Context, c client.Client, cluster *rabbitmqv1beta1.RabbitmqCluster) {
+	GinkgoHelper()
+
+	Expect(client.IgnoreNotFound(c.Delete(ctx, cluster))).To(Succeed())
+
+	key := types.NamespacedName{Name: cluster.Name, Namespace: cluster.Namespace}
+	Eventually(ctx, func() error {
+		var current rabbitmqv1beta1.RabbitmqCluster
+		err := c.Get(ctx, key, &current)
+		if k8serrors.IsNotFound(err) {
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		return fmt.Errorf("RabbitmqCluster %s is still present (deletionTimestamp: %v, finalizers: %v)",
+			key, current.DeletionTimestamp, current.Finalizers)
+	}).
+		WithTimeout(clusterDeletionTimeout).
+		WithPolling(2 * time.Second).
+		Should(Succeed())
+}
+
 func statefulSetPodName(cluster *rabbitmqv1beta1.RabbitmqCluster, index int) string {
 	return cluster.ChildResourceName(strings.Join([]string{"server", strconv.Itoa(index)}, "-"))
 }