Declare an error function and deal with topic

streamnative · Aug 21, 2024 · b100bfb · b100bfb
1 parent 6ab2c06
commit b100bfb
Show file tree

Hide file tree

Showing 5 changed files with 18 additions and 18 deletions.
diff --git a/pkg/admin/errors.go b/pkg/admin/errors.go
@@ -15,7 +15,9 @@
 package admin
 
 import (
+	"errors"
 	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/rest"
+	"net"
 )
 
 // Reason indicates the status code
@@ -70,3 +72,9 @@ func IsAlreadyExist(err error) bool {
 func IsInternalServerError(err error) bool {
 	return ErrorReason(err) == ReasonInternalServerError
 }
+
+// IsNoSuchHostError returns true if operator cannot connect the resource host
+func IsNoSuchHostError(err error) bool {
+	var dnsErr *net.DNSError
+	return errors.As(err, &dnsErr) && dnsErr.Err == "no such host"
+}
diff --git a/pkg/connection/reconcile_geo_replication.go b/pkg/connection/reconcile_geo_replication.go
@@ -16,10 +16,7 @@ package connection
 
 import (
 	"context"
-	"errors"
 	"fmt"
-	"net"
-
 	"github.com/go-logr/logr"
 	"github.com/streamnative/pulsar-resources-operator/pkg/feature"
 	"github.com/streamnative/pulsar-resources-operator/pkg/utils"
@@ -156,8 +153,7 @@ func (r *PulsarGeoReplicationReconciler) ReconcileGeoReplication(ctx context.Con
 				// Delete the cluster that created with destination cluster info.
 				// TODO it can only be deleted after the cluster has been removed from the tenant, namespace, and topic
 				if err := pulsarAdmin.DeleteCluster(destClusterName); err != nil && !admin.IsNotFound(err) {
-					var dnsErr *net.DNSError
-					if errors.As(err, &dnsErr) && dnsErr.Err == "no such host" {
+					if admin.IsNoSuchHostError(err) {
 						log.Info("Pulsar cluster has been deleted")
 					} else {
 						log.Error(err, "Failed to delete geo replication cluster")

diff --git a/pkg/connection/reconcile_namespace.go b/pkg/connection/reconcile_namespace.go
@@ -18,11 +18,9 @@ import (
 	"context"
 	"fmt"
 	"github.com/go-logr/logr"
-	"github.com/pkg/errors"
 	"github.com/streamnative/pulsar-resources-operator/pkg/feature"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/types"
-	"net"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
@@ -93,8 +91,7 @@ func (r *PulsarNamespaceReconciler) ReconcileNamespace(ctx context.Context, puls
 		if namespace.Status.GeoReplicationEnabled {
 			log.Info("GeoReplication is enabled. Reset namespace cluster", "LifecyclePolicy", namespace.Spec.LifecyclePolicy, "ClusterName", r.conn.connection.Spec.ClusterName)
 			if err := pulsarAdmin.SetNamespaceClusters(namespace.Spec.Name, []string{r.conn.connection.Spec.ClusterName}); err != nil {
-				var dnsErr *net.DNSError
-				if errors.As(err, &dnsErr) && dnsErr.Err == "no such host" {
+				if admin.IsNoSuchHostError(err) {
 					log.Info("Pulsar cluster has been deleted")
 				} else {
 					log.Error(err, "Failed to reset the cluster for namespace")
@@ -105,8 +102,7 @@ func (r *PulsarNamespaceReconciler) ReconcileNamespace(ctx context.Context, puls
 
 		if namespace.Spec.LifecyclePolicy != resourcev1alpha1.KeepAfterDeletion {
 			if err := pulsarAdmin.DeleteNamespace(namespace.Spec.Name); err != nil && !admin.IsNotFound(err) {
-				var dnsErr *net.DNSError
-				if errors.As(err, &dnsErr) && dnsErr.Err == "no such host" {
+				if admin.IsNoSuchHostError(err) {
 					log.Info("Pulsar cluster has been deleted")
 				} else {
 					log.Error(err, "Failed to delete namespace")

diff --git a/pkg/connection/reconcile_tenant.go b/pkg/connection/reconcile_tenant.go
@@ -16,10 +16,7 @@ package connection
 
 import (
 	"context"
-	"errors"
 	"fmt"
-	"net"
-
 	"github.com/go-logr/logr"
 	"github.com/streamnative/pulsar-resources-operator/pkg/feature"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -90,8 +87,7 @@ func (r *PulsarTenantReconciler) ReconcileTenant(ctx context.Context, pulsarAdmi
 		log.Info("Deleting tenant", "LifecyclePolicy", tenant.Spec.LifecyclePolicy)
 		if tenant.Spec.LifecyclePolicy != resourcev1alpha1.KeepAfterDeletion {
 			if err := pulsarAdmin.DeleteTenant(tenant.Spec.Name); err != nil && !admin.IsNotFound(err) {
-				var dnsErr *net.DNSError
-				if errors.As(err, &dnsErr) && dnsErr.Err == "no such host" {
+				if admin.IsNoSuchHostError(err) {
 					log.Info("Pulsar cluster has been deleted")
 				} else {
 					log.Error(err, "Failed to delete tenant")

diff --git a/pkg/connection/reconcile_topic.go b/pkg/connection/reconcile_topic.go
@@ -99,8 +99,12 @@ func (r *PulsarTopicReconciler) ReconcileTopic(ctx context.Context, pulsarAdmin
 			if topic.Status.GeoReplicationEnabled {
 				log.Info("GeoReplication is enabled. Reset topic cluster first", "LifecyclePolicy", topic.Spec.LifecyclePolicy, "ClusterName", r.conn.connection.Spec.ClusterName)
 				if err := pulsarAdmin.SetTopicClusters(topic.Spec.Name, topic.Spec.Persistent, []string{r.conn.connection.Spec.ClusterName}); err != nil {
-					log.Error(err, "Failed to reset the cluster for topic")
-					return err
+					if admin.IsNoSuchHostError(err) {
+						log.Info("Pulsar cluster has been deleted")
+					} else {
+						log.Error(err, "Failed to reset the cluster for topic")
+						return err
+					}
 				}
 			}