remove stream resources after drop from Postgres manifest (#2563)

* remove stream resources after drop from Postgres manifest

pkg/cluster/cluster.go

@@ -1082,7 +1082,7 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 	}
 
 	// streams
-	if len(newSpec.Spec.Streams) > 0 {
+	if len(newSpec.Spec.Streams) > 0 || len(oldSpec.Spec.Streams) != len(newSpec.Spec.Streams) {
 		if err := c.syncStreams(); err != nil {
 			c.logger.Errorf("could not sync streams: %v", err)
 			updateFailed = true

pkg/cluster/streams.go

@@ -327,7 +327,8 @@ func (c *Cluster) syncStreams() error {
 	if len(slotsToSync) > 0 {
 		requiredPatroniConfig.Slots = slotsToSync
 	} else {
-		return nil
+		// try to delete existing stream resources
+		return c.deleteStreams()
 	}
 
 	c.logger.Debug("syncing logical replication slots")

pkg/cluster/streams_test.go

@@ -455,4 +455,21 @@ func TestUpdateFabricEventStream(t *testing.T) {
 	if match, _ := sameStreams(streams.Items[0].Spec.EventStreams, result.Spec.EventStreams); !match {
 		t.Errorf("Malformed FabricEventStream after disabling event recovery, expected %#v, got %#v", streams.Items[0], result)
 	}
+
+	mockClient := k8sutil.NewMockKubernetesClient()
+	cluster.KubeClient.CustomResourceDefinitionsGetter = mockClient.CustomResourceDefinitionsGetter
+
+	// remove streams from manifest
+	pgPatched.Spec.Streams = nil
+	pgUpdated, err := cluster.KubeClient.Postgresqls(namespace).Update(
+		context.TODO(), pgPatched, metav1.UpdateOptions{})
+	assert.NoError(t, err)
+
+	cluster.Postgresql.Spec = pgUpdated.Spec
+	cluster.syncStreams()
+
+	streamList, err := cluster.KubeClient.FabricEventStreams(namespace).List(context.TODO(), listOptions)
+	if len(streamList.Items) > 0 || err != nil {
+		t.Errorf("stream resource has not been removed or unexpected error %v", err)
+	}
 }