fix error when delete mgh then recreate it (#1288)

Signed-off-by: root <[email protected]>
stolostron · Dec 20, 2024 · 9571226 · 9571226
1 parent 0fd0844
commit 9571226
Show file tree

Hide file tree

Showing 3 changed files with 28 additions and 2 deletions.
diff --git a/operator/pkg/config/storage_config.go b/operator/pkg/config/storage_config.go
@@ -142,7 +142,11 @@ func GetPGConnectionFromBuildInPostgres(ctx context.Context, client client.Clien
 // SetStorageConnection update the postgres connection
 func SetStorageConnection(conn *PostgresConnection) bool {
 	log.Debugf("Set Storage Connection: %v", conn == nil)
-	if conn != nil && !reflect.DeepEqual(conn, postgresConn) {
+	if conn == nil {
+		postgresConn = nil
+		return true
+	}
+	if !reflect.DeepEqual(conn, postgresConn) {
 		postgresConn = conn
 		log.Debugf("Update Storage Connection")
 		return true

diff --git a/operator/pkg/config/transport_config.go b/operator/pkg/config/transport_config.go
@@ -45,7 +45,11 @@ var (
 
 func SetTransporterConn(conn *transport.KafkaConfig) bool {
 	log.Debug("set Transporter Conn")
-	if conn != nil && !reflect.DeepEqual(conn, transporterConn) {
+	if conn == nil {
+		transporterConn = nil
+		return true
+	}
+	if !reflect.DeepEqual(conn, transporterConn) {
 		transporterConn = conn
 		log.Debug("update Transporter Conn")
 		return true

diff --git a/operator/pkg/controllers/transporter/protocol/strimzi_kafka_controller.go b/operator/pkg/controllers/transporter/protocol/strimzi_kafka_controller.go
@@ -10,6 +10,7 @@ import (
 
 	kafkav1beta2 "github.com/RedHatInsights/strimzi-client-go/apis/kafka.strimzi.io/v1beta2"
 	subv1alpha1 "github.com/operator-framework/api/pkg/operators/v1alpha1"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -293,6 +294,23 @@ func (r *KafkaController) pruneStrimziResources(ctx context.Context) (ctrl.Resul
 	}
 	log.Infof("kafka cluster deleted")
 
+	// Delete kafka pvc
+	kafkaPvc := &corev1.PersistentVolumeClaimList{}
+	pvcListOpts := []client.ListOption{
+		client.MatchingLabels(map[string]string{
+			"strimzi.io/cluster": KafkaClusterName,
+		}),
+	}
+	if err := r.c.List(ctx, kafkaPvc, pvcListOpts...); err != nil {
+		return ctrl.Result{}, err
+	}
+	for idx := range kafkaPvc.Items {
+		log.Infof("Delete kafka pvc %v", kafkaPvc.Items[idx].Name)
+		if err := r.c.Delete(ctx, &kafkaPvc.Items[idx]); err != nil && !errors.IsNotFound(err) {
+			return ctrl.Result{}, err
+		}
+	}
+
 	kafkaSub := &subv1alpha1.Subscription{}
 	err := r.c.Get(ctx, types.NamespacedName{
 		Namespace: utils.GetDefaultNamespace(),