Repository: Altinity/clickhouse-operator

Issue:
PodDisruptionBudget "maxUnavailable" doesn't prevent from downtime
### Description

Currently, `PodDisruptionBudget` allows configuring only `maxUnavailable`, which doesn't prevent downtime while evicting shard pods. This is reproducible with a simple example having a single shard and 2 replicas and default `maxUnavailable: 1`.

```
     layout:
        shards:
        - internalReplication: "true"
          replicas:
          - templates:
              podTemplate: clickhouse-in-zone-a
          - templates:
              podTemplate: clickhouse-in-zone-b
```

While evicting both pods at a time, there is a short gap when 2nd pod can be evicted while the first pod is gone. 

The status of `pdb` before start of the eviction:

```
{
  "conditions": [
    {
      "lastTransitionTime": "2025-05-11T16:05:17Z",
      "message": "",
      "observedGeneration": 1,
      "reason": "SufficientPods",
      "status": "True",
      "type": "DisruptionAllowed"
    }
  ],
  "currentHealthy": 2, // <- main part
  "desiredHealthy": 1, // <- main part
  "disruptionsAllowed": 1, // <- main part
  "expectedPods": 2,
  "observedGeneration": 1
}
```
The status of `pdb` when 1st pod was deleted:
```
{
  "conditions": [
    {
      "lastTransitionTime": "2025-05-11T16:05:58Z",
      "message": "",
      "observedGeneration": 1,
      "reason": "SufficientPods",
      "status": "True",
      "type": "DisruptionAllowed"
    }
  ],
  "currentHealthy": 1, // <- main part
  "desiredHealthy": 0, // <- main part
  "disruptionsAllowed": 1, // <- main part
  "expectedPods": 1,
  "observedGeneration": 1
}
```
Notice that at some point, pdb observes`"expectedPods": 1`. While it happens, 2nd pod is allowed to be evicted as it doesn't violate the pod's disruption budget based on the current state.

Based on k8s docs, we can't be using `maxUnavailable` with the pod setup we have:

https://kubernetes.io/docs/tasks/run-application/configure-pdb/#arbitrary-controllers-and-selectors
```
You can use a PDB with pods controlled by another resource, by an "operator", or bare pods, but with these restrictions:

- only .spec.minAvailable can be used, not .spec.maxUnavailable.
- only an integer value can be used with .spec.minAvailable, not a percentage.
```

Additionally, `The eviction API will disallow eviction of any pod covered by multiple PDBs, so most users will want to avoid overlapping selectors` so we can't have multiple PDBs defined.

### Problem
- `maxUnavailable` cannot be used reliably while managing arbitrary pods
- default PDB creation is not configurable, preventing from customization

### Solution

- allow to disable default PDB creation, enabling the user to define PDBs manually
Or
- create `minAvailable` PDB per shard instead





Discussion:
@alex-zaitsev: Duplicate of https://github.com/Altinity/clickhouse-operator/issues/1225@varnastadeus: @alex-zaitsev I am not sure it's really a duplicate, the current configuration is prone to downtime and its more like a bug than a feature request. Having per shard PDB with `maxUnavailable: 1` will still have the same issue. Ideally, we would like to have the freedom to define PDBs on our own as there are multiple configuration options I can think of. What if we would have an option to disable PDB creation from operator side?

Solution patch:
diff --git a/deploy/builder/templates-install-bundle/clickhouse-operator-install-yaml-template-01-section-crd-01-chi-chit.yaml b/deploy/builder/templates-install-bundle/clickhouse-operator-install-yaml-template-01-section-crd-01-chi-chit.yaml
index acd0a1684..ad7327eee 100644
--- a/deploy/builder/templates-install-bundle/clickhouse-operator-install-yaml-template-01-section-crd-01-chi-chit.yaml
+++ b/deploy/builder/templates-install-bundle/clickhouse-operator-install-yaml-template-01-section-crd-01-chi-chit.yaml
@@ -799,6 +799,15 @@ spec:
                                     required:
                                       - name
                                       - key
+                          pdbManaged:
+                            <<: *TypeStringBool
+                            description: |
+                              Specifies whether the Pod Disruption Budget (PDB) should be managed.
+                              During the next installation, if PDB management is enabled, the operator will
+                              attempt to retrieve any existing PDB. If none is found, it will create a new one
+                              and initiate a reconciliation loop. If PDB management is disabled, the existing PDB
+                              will remain intact, and the reconciliation loop will not be executed. By default,
+                              PDB management is enabled.
                           pdbMaxUnavailable:
                             type: integer
                             description: |
diff --git a/deploy/builder/templates-install-bundle/clickhouse-operator-install-yaml-template-01-section-crd-03-chk.yaml b/deploy/builder/templates-install-bundle/clickhouse-operator-install-yaml-template-01-section-crd-03-chk.yaml
index e93374339..2d261b899 100644
--- a/deploy/builder/templates-install-bundle/clickhouse-operator-install-yaml-template-01-section-crd-03-chk.yaml
+++ b/deploy/builder/templates-install-bundle/clickhouse-operator-install-yaml-template-01-section-crd-03-chk.yaml
@@ -528,6 +528,15 @@ spec:
                             description: |
                               optional, allows define content of any setting file inside each `Pod` on current cluster during generate `ConfigMap` which will mount in `/etc/clickhouse-server/config.d/` or `/etc/clickhouse-server/conf.d/` or `/etc/clickhouse-server/users.d/`
                               override top-level `chi.spec.configuration.files`
+                          pdbManaged:
+                            <<: *TypeStringBool
+                            description: |
+                              Specifies whether the Pod Disruption Budget (PDB) should be managed.
+                              During the next installation, if PDB management is enabled, the operator will
+                              attempt to retrieve any existing PDB. If none is found, it will create a new one
+                              and initiate a reconciliation loop. If PDB management is disabled, the existing PDB
+                              will remain intact, and the reconciliation loop will not be executed. By default,
+                              PDB management is enabled.
                           templates:
                             <<: *TypeTemplateNames
                             description: |
diff --git a/go.mod b/go.mod
index ea95d8e32..c7e8221f1 100644
--- a/go.mod
+++ b/go.mod
@@ -82,6 +82,7 @@ require (
 	github.com/prometheus/common v0.45.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/stretchr/objx v0.5.0 // indirect
 	github.com/xo/terminfo v0.0.0-20210125001918-ca9a967f8778 // indirect
 	go.opentelemetry.io/otel/trace v1.24.0 // indirect
 	go.uber.org/atomic v1.10.0 // indirect
diff --git a/go.sum b/go.sum
index 6d4cc8fb5..cfaad41cb 100644
--- a/go.sum
+++ b/go.sum
@@ -418,6 +418,7 @@ github.com/spf13/viper v1.3.2/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DM
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/objx v0.5.0 h1:1zr/of2m5FGMsad5YfcqgdqdWrIhu+EBEJRhR1U7z/c=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
 github.com/stretchr/testify v0.0.0-20161117074351-18a02ba4a312/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v0.0.0-20170130113145-4d4bfba8f1d1/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
diff --git a/pkg/apis/clickhouse-keeper.altinity.com/v1/type_cluster.go b/pkg/apis/clickhouse-keeper.altinity.com/v1/type_cluster.go
index 702730e0a..ecc37000e 100644
--- a/pkg/apis/clickhouse-keeper.altinity.com/v1/type_cluster.go
+++ b/pkg/apis/clickhouse-keeper.altinity.com/v1/type_cluster.go
@@ -23,10 +23,11 @@ import (
 type Cluster struct {
 	Name string `json:"name,omitempty"         yaml:"name,omitempty"`
 
-	Settings  *apiChi.Settings      `json:"settings,omitempty"          yaml:"settings,omitempty"`
-	Files     *apiChi.Settings      `json:"files,omitempty"             yaml:"files,omitempty"`
-	Templates *apiChi.TemplatesList `json:"templates,omitempty"         yaml:"templates,omitempty"`
-	Layout    *ChkClusterLayout     `json:"layout,omitempty"       yaml:"layout,omitempty"`
+	Settings   *apiChi.Settings      `json:"settings,omitempty"          yaml:"settings,omitempty"`
+	Files      *apiChi.Settings      `json:"files,omitempty"             yaml:"files,omitempty"`
+	Templates  *apiChi.TemplatesList `json:"templates,omitempty"         yaml:"templates,omitempty"`
+	Layout     *ChkClusterLayout     `json:"layout,omitempty"       yaml:"layout,omitempty"`
+	PDBManaged *types.StringBool     `json:"pdbManaged,omitempty"        yaml:"pdbManaged,omitempty"`
 
 	Runtime ChkClusterRuntime `json:"-" yaml:"-"`
 }
@@ -128,6 +129,11 @@ func (cluster *Cluster) GetRuntime() apiChi.IClusterRuntime {
 	return &cluster.Runtime
 }
 
+// GetPDBManaged is a getter
+func (cluster *Cluster) GetPDBManaged() *types.StringBool {
+	return cluster.PDBManaged
+}
+
 // GetPDBMaxUnavailable is a getter
 func (cluster *Cluster) GetPDBMaxUnavailable() *types.Int32 {
 	return types.NewInt32(1)
diff --git a/pkg/apis/clickhouse.altinity.com/v1/interface.go b/pkg/apis/clickhouse.altinity.com/v1/interface.go
index 4e5d62138..4863584de 100644
--- a/pkg/apis/clickhouse.altinity.com/v1/interface.go
+++ b/pkg/apis/clickhouse.altinity.com/v1/interface.go
@@ -116,6 +116,7 @@ type ICluster interface {
 	GetInsecure() *types.StringBool
 	GetSecure() *types.StringBool
 	GetSecret() *ClusterSecret
+	GetPDBManaged() *types.StringBool
 	GetPDBMaxUnavailable() *types.Int32
 
 	WalkShards(f func(index int, shard IShard) error) []error
diff --git a/pkg/apis/clickhouse.altinity.com/v1/type_cluster.go b/pkg/apis/clickhouse.altinity.com/v1/type_cluster.go
index 91ea3764d..836f91007 100644
--- a/pkg/apis/clickhouse.altinity.com/v1/type_cluster.go
+++ b/pkg/apis/clickhouse.altinity.com/v1/type_cluster.go
@@ -29,6 +29,7 @@ type Cluster struct {
 	Insecure          *types.StringBool `json:"insecure,omitempty"          yaml:"insecure,omitempty"`
 	Secure            *types.StringBool `json:"secure,omitempty"            yaml:"secure,omitempty"`
 	Secret            *ClusterSecret    `json:"secret,omitempty"            yaml:"secret,omitempty"`
+	PDBManaged        *types.StringBool `json:"pdbManaged,omitempty"        yaml:"pdbManaged,omitempty"`
 	PDBMaxUnavailable *types.Int32      `json:"pdbMaxUnavailable,omitempty" yaml:"pdbMaxUnavailable,omitempty"`
 	Reconcile         ClusterReconcile  `json:"reconcile"                   yaml:"reconcile"`
 	Layout            *ChiClusterLayout `json:"layout,omitempty"            yaml:"layout,omitempty"`
@@ -170,6 +171,11 @@ func (cluster *Cluster) GetRuntime() IClusterRuntime {
 	return &cluster.Runtime
 }
 
+// GetPDBManaged is a getter
+func (cluster *Cluster) GetPDBManaged() *types.StringBool {
+	return cluster.PDBManaged
+}
+
 // GetPDBMaxUnavailable is a getter
 func (cluster *Cluster) GetPDBMaxUnavailable() *types.Int32 {
 	return cluster.PDBMaxUnavailable
diff --git a/pkg/controller/chi/worker-reconciler-chi.go b/pkg/controller/chi/worker-reconciler-chi.go
index ed54bfab9..9274c6dc6 100644
--- a/pkg/controller/chi/worker-reconciler-chi.go
+++ b/pkg/controller/chi/worker-reconciler-chi.go
@@ -566,6 +566,10 @@ func (w *worker) reconcileClusterSecret(ctx context.Context, cluster *api.Cluste
 }
 
 func (w *worker) reconcileClusterPodDisruptionBudget(ctx context.Context, cluster *api.Cluster) error {
+	if cluster.GetPDBManaged().IsFalse() {
+		return nil
+	}
+
 	pdb := w.task.Creator().CreatePodDisruptionBudget(cluster)
 	if err := w.reconcilePDB(ctx, cluster, pdb); err == nil {
 		w.task.RegistryReconciled().RegisterPDB(pdb.GetObjectMeta())
diff --git a/pkg/controller/chk/worker-chk-reconciler.go b/pkg/controller/chk/worker-chk-reconciler.go
index c06305d51..1ff27693c 100644
--- a/pkg/controller/chk/worker-chk-reconciler.go
+++ b/pkg/controller/chk/worker-chk-reconciler.go
@@ -403,14 +403,22 @@ func (w *worker) reconcileCluster(ctx context.Context, cluster *apiChk.Cluster)
 
 	w.reconcileClusterSecret(ctx, cluster)
 
+	w.reconcileClusterPodDisruptionBudget(ctx, cluster)
+
+	return nil
+}
+
+func (w *worker) reconcileClusterPodDisruptionBudget(ctx context.Context, cluster *apiChk.Cluster) {
+	if cluster.GetPDBManaged().IsFalse() {
+		return
+	}
+
 	pdb := w.task.Creator().CreatePodDisruptionBudget(cluster)
 	if err := w.reconcilePDB(ctx, cluster, pdb); err == nil {
 		w.task.RegistryReconciled().RegisterPDB(pdb.GetObjectMeta())
 	} else {
 		w.task.RegistryFailed().RegisterPDB(pdb.GetObjectMeta())
 	}
-
-	return nil
 }
 
 func (w *worker) reconcileClusterSecret(ctx context.Context, cluster *apiChk.Cluster) {
diff --git a/pkg/model/chi/normalizer/normalizer.go b/pkg/model/chi/normalizer/normalizer.go
index 126dd3b9f..718f61a8f 100644
--- a/pkg/model/chi/normalizer/normalizer.go
+++ b/pkg/model/chi/normalizer/normalizer.go
@@ -807,6 +807,7 @@ func (n *Normalizer) normalizeClusterStage2(cluster *chi.Cluster) *chi.Cluster {
 	cluster.Files = n.normalizeConfigurationFiles(cluster.Files, cluster)
 
 	cluster.SchemaPolicy = n.normalizeClusterSchemaPolicy(cluster.SchemaPolicy)
+	cluster.PDBManaged = n.normalizePDBManaged(cluster.PDBManaged)
 	cluster.PDBMaxUnavailable = n.normalizePDBMaxUnavailable(cluster.PDBMaxUnavailable)
 	cluster.Reconcile = n.normalizeClusterReconcile(cluster.Reconcile)
 
@@ -876,6 +877,16 @@ func (n *Normalizer) normalizeClusterSchemaPolicy(policy *chi.SchemaPolicy) *chi
 	return policy
 }
 
+// normalizePDBManaged normalizes PDBManaged
+func (n *Normalizer) normalizePDBManaged(pdbManaged *types.StringBool) *types.StringBool {
+	if pdbManaged.IsValid() {
+		return pdbManaged
+	}
+
+	// unknown value, fallback to true
+	return types.NewStringBool(true)
+}
+
 // normalizePDBMaxUnavailable normalizes PDBMaxUnavailable
 func (n *Normalizer) normalizePDBMaxUnavailable(value *types.Int32) *types.Int32 {
 	return value.Normalize(1)
diff --git a/pkg/model/chk/normalizer/normalizer.go b/pkg/model/chk/normalizer/normalizer.go
index 3f923cfbd..c5a89ac3b 100644
--- a/pkg/model/chk/normalizer/normalizer.go
+++ b/pkg/model/chk/normalizer/normalizer.go
@@ -495,9 +495,21 @@ func (n *Normalizer) normalizeCluster(cluster *chk.Cluster) *chk.Cluster {
 		return nil
 	})
 
+	cluster.PDBManaged = n.normalizePDBManaged(cluster.PDBManaged)
+
 	return cluster
 }
 
+// normalizePDBManaged normalizes PDBManaged
+func (n *Normalizer) normalizePDBManaged(pdbManaged *types.StringBool) *types.StringBool {
+	if pdbManaged.IsValid() {
+		return pdbManaged
+	}
+
+	// unknown value, fallback to true
+	return types.NewStringBool(true)
+}
+
 // normalizeClusterLayoutShardsCountAndReplicasCount ensures at least 1 shard and 1 replica counters
 func (n *Normalizer) normalizeClusterLayoutShardsCountAndReplicasCount(clusterLayout *chk.ChkClusterLayout) *chk.ChkClusterLayout {
 	// Ensure layout
diff --git a/pkg/model/zookeeper/connection.go b/pkg/model/zookeeper/connection.go
index f309db088..38babca22 100644
--- a/pkg/model/zookeeper/connection.go
+++ b/pkg/model/zookeeper/connection.go
@@ -18,6 +18,7 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
+	"errors"
 	"fmt"
 	"math/rand"
 	"net"
@@ -33,70 +34,94 @@ import (
 	api "github.com/altinity/clickhouse-operator/pkg/apis/clickhouse.altinity.com/v1"
 )
 
+// ZKClient abstracts the subset of zk.Conn methods used by Connection for testability.
+type ZKClient interface {
+	Get(path string) ([]byte, *zk.Stat, error)
+	Set(path string, data []byte, version int32) (*zk.Stat, error)
+	Create(path string, data []byte, flags int32, acl []zk.ACL) (string, error)
+	Delete(path string, version int32) error
+	Exists(path string) (bool, *zk.Stat, error)
+	AddAuth(scheme string, auth []byte) error
+	Close()
+}
+
+// Assert that zk.Conn implements ZKClient
+var _ ZKClient = (*zk.Conn)(nil)
+
 type Connection struct {
 	nodes api.ZookeeperNodes
 	ConnectionParams
 	sema       *semaphore.Weighted
 	mu         sync.Mutex
-	connection *zk.Conn
+	connection ZKClient
+
+	// retryDelayFn is configurable for testing
+	retryDelayFn func(i int)
 }
 
+// NewConnection creates a new Zookeeper connection with the provided nodes and parameters.
 func NewConnection(nodes api.ZookeeperNodes, _params ...*ConnectionParams) *Connection {
-	var params *ConnectionParams
-	if len(_params) > 0 {
-		params = _params[0]
-	}
-	params = params.Normalize()
+	params := BuildConnectionParams(_params...)
 	return &Connection{
 		nodes:            nodes,
 		sema:             semaphore.NewWeighted(params.MaxConcurrentRequests),
 		ConnectionParams: *params,
+		retryDelayFn: func(i int) {
+			time.Sleep(time.Duration(i)*time.Second + time.Duration(rand.Int63n(int64(1*time.Second))))
+		},
 	}
 }
 
+// Get retrieves data from the specified path in Zookeeper.
 func (c *Connection) Get(ctx context.Context, path string) (data []byte, stat *zk.Stat, err error) {
-	err = c.retry(ctx, func(connection *zk.Conn) error {
+	err = c.retry(ctx, func(connection ZKClient) error {
 		data, stat, err = connection.Get(path)
 		return err
 	})
 	return
 }
 
-func (c *Connection) Exists(ctx context.Context, path string) bool {
-	exists, _, _ := c.Details(ctx, path)
-	return exists
+// Exists checks if the specified path exists in Zookeeper.
+func (c *Connection) Exists(ctx context.Context, path string) (bool, error) {
+	exists, _, err := c.Details(ctx, path)
+	return exists, err
 }
 
+// Details retrieves existence and stat information for the specified path in Zookeeper.
 func (c *Connection) Details(ctx context.Context, path string) (exists bool, stat *zk.Stat, err error) {
-	err = c.retry(ctx, func(connection *zk.Conn) error {
+	err = c.retry(ctx, func(connection ZKClient) error {
 		exists, stat, err = connection.Exists(path)
 		return err
 	})
 	return
 }
 
+// Create creates a new node at the specified path with the given value, flags, and ACL.
 func (c *Connection) Create(ctx context.Context, path string, value []byte, flags int32, acl []zk.ACL) (pathCreated string, err error) {
-	err = c.retry(ctx, func(connection *zk.Conn) error {
+	err = c.retry(ctx, func(connection ZKClient) error {
 		pathCreated, err = connection.Create(path, value, flags, acl)
 		return err
 	})
 	return
 }
 
+// Set updates the value of the node at the specified path with the given version.
 func (c *Connection) Set(ctx context.Context, path string, value []byte, version int32) (stat *zk.Stat, err error) {
-	err = c.retry(ctx, func(connection *zk.Conn) error {
+	err = c.retry(ctx, func(connection ZKClient) error {
 		stat, err = connection.Set(path, value, version)
 		return err
 	})
 	return
 }
 
+// Delete removes the node at the specified path with the given version.
 func (c *Connection) Delete(ctx context.Context, path string, version int32) error {
-	return c.retry(ctx, func(connection *zk.Conn) error {
+	return c.retry(ctx, func(connection ZKClient) error {
 		return connection.Delete(path, version)
 	})
 }
 
+// Close closes the Zookeeper connection if it exists. If the connection is nil, it does nothing.
 func (c *Connection) Close() error {
 	if c == nil {
 		return nil
@@ -109,41 +134,55 @@ func (c *Connection) Close() error {
 	return nil
 }
 
-func (c *Connection) retry(ctx context.Context, fn func(*zk.Conn) error) error {
+func (c *Connection) retry(ctx context.Context, fn func(connection ZKClient) error) error {
 	if err := c.sema.Acquire(ctx, 1); err != nil {
 		return err
 	}
 	defer c.sema.Release(1)
 
+	var errs []error
 	for i := 0; i < c.MaxRetriesNum; i++ {
 		if i > 0 {
 			// Progressive delay before each retry
-			time.Sleep(time.Duration(i)*time.Second + time.Duration(rand.Int63n(int64(1*time.Second))))
+			c.retryDelayFn(i)
 		}
 
 		connection, err := c.ensureConnection(ctx)
 		if err != nil {
+			errs = append(errs, fmt.Errorf("retry %d: connection error: %w", i+1, err))
 			continue // Retry
 		}
 
 		err = fn(connection)
+		if err == nil {
+			// Success - return nil, no need for caller to know about errors
+			return nil
+		}
+
+		// Handle specific error cases
 		if err == zk.ErrConnectionClosed {
 			c.mu.Lock()
 			if c.connection == connection {
 				c.connection = nil
 			}
 			c.mu.Unlock()
+			errs = append(errs, fmt.Errorf("retry %d: connection closed: %w", i+1, err))
 			continue // Retry
 		}
 
-		// Got result
-		return err
+		// Collect the errors
+		errs = append(errs, fmt.Errorf("retry %d: %w", i+1, err))
+	}
+
+	// All retries failed - wrap all errors
+	if len(errs) == 0 {
+		return fmt.Errorf("max retries number reached: %d", c.MaxRetriesNum)
 	}
 
-	return fmt.Errorf("max retries number reached")
+	return fmt.Errorf("all retries (%d) failed: %w", c.MaxRetriesNum, errors.Join(errs...))
 }
 
-func (c *Connection) ensureConnection(ctx context.Context) (*zk.Conn, error) {
+func (c *Connection) ensureConnection(ctx context.Context) (ZKClient, error) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
@@ -181,7 +220,7 @@ func (c *Connection) connectionAddAuth(ctx context.Context) {
 	}
 }
 
-func (c *Connection) connectionEventsProcessor(connection *zk.Conn, events <-chan zk.Event) {
+func (c *Connection) connectionEventsProcessor(connection ZKClient, events <-chan zk.Event) {
 	for event := range events {
 		shouldCloseConnection := false
 		switch event.State {
@@ -206,7 +245,7 @@ func (c *Connection) connectionEventsProcessor(connection *zk.Conn, events <-cha
 	}
 }
 
-func (c *Connection) dial(ctx context.Context) (*zk.Conn, <-chan zk.Event, error) {
+func (c *Connection) dial(ctx context.Context) (ZKClient, <-chan zk.Event, error) {
 	ctx, cancel := context.WithTimeout(ctx, c.TimeoutConnect)
 	defer cancel()
 
@@ -232,7 +271,7 @@ func (c *Connection) dial(ctx context.Context) (*zk.Conn, <-chan zk.Event, error
 	}
 }
 
-func (c *Connection) connect(servers []string) (*zk.Conn, <-chan zk.Event, error) {
+func (c *Connection) connect(servers []string) (ZKClient, <-chan zk.Event, error) {
 	optionsDialer := zk.WithDialer(net.DialTimeout)
 	if c.CertFile != "" && c.KeyFile != "" {
 		if len(servers) > 1 {
diff --git a/pkg/model/zookeeper/connection_params.go b/pkg/model/zookeeper/connection_params.go
index ea2d581a0..751fb87fd 100644
--- a/pkg/model/zookeeper/connection_params.go
+++ b/pkg/model/zookeeper/connection_params.go
@@ -36,6 +36,14 @@ type ConnectionParams struct {
 	AuthFile string
 }
 
+func BuildConnectionParams(_params ...*ConnectionParams) *ConnectionParams {
+	var params *ConnectionParams
+	if len(_params) > 0 {
+		params = _params[0]
+	}
+	return params.Normalize()
+}
+
 func (p *ConnectionParams) Normalize() *ConnectionParams {
 	if p == nil {
 		// Overwrite nil pointer with struct to be returned
diff --git a/pkg/model/zookeeper/path_manager.go b/pkg/model/zookeeper/path_manager.go
index d2ed3a44c..f456296c5 100644
--- a/pkg/model/zookeeper/path_manager.go
+++ b/pkg/model/zookeeper/path_manager.go
@@ -61,7 +61,11 @@ func (p *PathManager) Ensure(path string) {
 	subPath := ""
 	for _, folder := range pathParts {
 		subPath += "/" + folder
-		if p.Connection.Exists(ctx, subPath) {
+		if ok, err := p.Connection.Exists(ctx, subPath); !ok {
+			if err != nil {
+				log.Warning("received error while checking zk path: %s err: %v", subPath, err)
+			}
+		} else {
 			log.Info("zk path already exists: %s", subPath)
 			continue // for
 		}



Test patch:
diff --git a/pkg/model/zookeeper/connection_test.go b/pkg/model/zookeeper/connection_test.go
new file mode 100644
index 000000000..46a15164f
--- /dev/null
+++ b/pkg/model/zookeeper/connection_test.go
@@ -0,0 +1,748 @@
+package zookeeper
+
+import (
+	"context"
+	"testing"
+
+	api "github.com/altinity/clickhouse-operator/pkg/apis/clickhouse.altinity.com/v1"
+	"github.com/go-zookeeper/zk"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+func TestConnection_Get(t *testing.T) {
+	// Sample stat for successful responses
+	sampleStat := &zk.Stat{
+		Czxid:          123,
+		Mzxid:          124,
+		Ctime:          1234567890,
+		Mtime:          1234567891,
+		Version:        1,
+		Cversion:       0,
+		Aversion:       0,
+		EphemeralOwner: 0,
+		DataLength:     4,
+		NumChildren:    0,
+		Pzxid:          125,
+	}
+
+	tests := []struct {
+		name           string
+		path           string
+		setupMock      func(*MockZKClient)
+		expectedData   []byte
+		expectedStat   *zk.Stat
+		expectedErrors []error
+		retries        int
+	}{
+		{
+			name: "success: successful get operation",
+			path: "/test/path",
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Get", "/test/path").Return([]byte("test data"), sampleStat, nil).Once()
+			},
+			expectedData: []byte("test data"),
+			expectedStat: sampleStat,
+			retries:      1, // 1 means a single attempt
+		},
+		{
+			name: "success: retry scenario - 2 errors then success",
+			path: "/test/retry-path",
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Get", "/test/retry-path").Return([]byte(nil), (*zk.Stat)(nil), zk.ErrAPIError).Once()
+				mockClient.On("Get", "/test/retry-path").Return([]byte(nil), (*zk.Stat)(nil), zk.ErrNoAuth).Once()
+				mockClient.On("Get", "/test/retry-path").Return([]byte("retry success"), sampleStat, nil).Once()
+			},
+			expectedData: []byte("retry success"),
+			expectedStat: sampleStat,
+			retries:      3,
+		},
+		{
+			name: "error: persistent connection error, no retries",
+			path: "/nonexistent/path",
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Get", "/nonexistent/path").Return([]byte(nil), (*zk.Stat)(nil), zk.ErrNoNode).Once()
+			},
+			expectedData:   nil,
+			expectedStat:   nil,
+			expectedErrors: []error{zk.ErrNoNode},
+			retries:        1,
+		},
+
+		{
+			name: "error: persistent connection error, with retries",
+			path: "/nonexistent/path",
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Get", "/nonexistent/path").Return([]byte(nil), (*zk.Stat)(nil), zk.ErrNoNode).Once()
+				mockClient.On("Get", "/nonexistent/path").Return([]byte(nil), (*zk.Stat)(nil), zk.ErrAPIError).Once()
+				mockClient.On("Get", "/nonexistent/path").Return([]byte(nil), (*zk.Stat)(nil), zk.ErrNoAuth).Once()
+			},
+			expectedData:   nil,
+			expectedStat:   nil,
+			expectedErrors: []error{zk.ErrNoNode, zk.ErrAPIError, zk.ErrNoAuth},
+			retries:        3,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create mock client
+			mockClient := new(MockZKClient)
+
+			// Setup mock expectations
+			tt.setupMock(mockClient)
+
+			// Create connection with mock client
+			conn := newTestConnection(api.ZookeeperNodes{}, mockClient, &ConnectionParams{MaxRetriesNum: tt.retries})
+
+			// Call the method under test
+			ctx := context.Background()
+			data, stat, err := conn.Get(ctx, tt.path)
+
+			// Assert expectations
+			if len(tt.expectedErrors) > 0 {
+				assert.Error(t, err)
+				// Verify that all expected errors are present in the aggregated error
+				for _, expectedError := range tt.expectedErrors {
+					assert.ErrorIs(t, err, expectedError, "Expected error %v should be present in aggregated error", expectedError)
+				}
+				assert.Nil(t, data)
+				assert.Nil(t, stat)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.expectedData, data)
+				assert.Equal(t, tt.expectedStat, stat)
+			}
+
+			// Verify all mock expectations were met
+			mockClient.AssertExpectations(t)
+		})
+	}
+}
+
+func TestConnection_Exists(t *testing.T) {
+	// Sample stat for successful responses
+	sampleStat := &zk.Stat{
+		Czxid:          123,
+		Mzxid:          124,
+		Ctime:          1234567890,
+		Mtime:          1234567891,
+		Version:        1,
+		Cversion:       0,
+		Aversion:       0,
+		EphemeralOwner: 0,
+		DataLength:     4,
+		NumChildren:    0,
+		Pzxid:          125,
+	}
+
+	tests := []struct {
+		name           string
+		path           string
+		setupMock      func(*MockZKClient)
+		expectedExists bool
+		expectedErrors []error
+		retries        int
+	}{
+		{
+			name: "success: node exists",
+			path: "/test/path",
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Exists", "/test/path").Return(true, sampleStat, nil).Once()
+			},
+			expectedExists: true,
+			retries:        1,
+		},
+		{
+			name: "success: node does not exist",
+			path: "/nonexistent/path",
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Exists", "/nonexistent/path").Return(false, (*zk.Stat)(nil), nil).Once()
+			},
+			expectedExists: false,
+			retries:        1,
+		},
+		{
+			name: "success: retry scenario - 2 errors then success",
+			path: "/test/retry-path",
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Exists", "/test/retry-path").Return(false, (*zk.Stat)(nil), zk.ErrAPIError).Once()
+				mockClient.On("Exists", "/test/retry-path").Return(false, (*zk.Stat)(nil), zk.ErrNoAuth).Once()
+				mockClient.On("Exists", "/test/retry-path").Return(true, sampleStat, nil).Once()
+			},
+			expectedExists: true,
+			retries:        3,
+		},
+		{
+			name: "error: persistent connection error, no retries",
+			path: "/test/path",
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Exists", "/test/path").Return(false, (*zk.Stat)(nil), zk.ErrConnectionClosed).Once()
+			},
+			expectedExists: false,
+			expectedErrors: []error{zk.ErrConnectionClosed},
+			retries:        1,
+		},
+		{
+			name: "error: all retries fail with different errors",
+			path: "/test/path",
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Exists", "/test/path").Return(false, (*zk.Stat)(nil), zk.ErrAPIError).Once()
+				mockClient.On("Exists", "/test/path").Return(false, (*zk.Stat)(nil), zk.ErrNoAuth).Once()
+				mockClient.On("Exists", "/test/path").Return(false, (*zk.Stat)(nil), zk.ErrBadVersion).Once()
+			},
+			expectedExists: false,
+			expectedErrors: []error{zk.ErrAPIError, zk.ErrNoAuth, zk.ErrBadVersion},
+			retries:        3,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create mock client
+			mockClient := new(MockZKClient)
+
+			// Setup mock expectations
+			tt.setupMock(mockClient)
+
+			// Create connection with mock client
+			conn := newTestConnection(api.ZookeeperNodes{}, mockClient, &ConnectionParams{MaxRetriesNum: tt.retries})
+
+			// Call the method under test
+			ctx := context.Background()
+			exists, err := conn.Exists(ctx, tt.path)
+
+			// Assert expectations
+			if len(tt.expectedErrors) > 0 {
+				assert.Error(t, err)
+				// Verify that all expected errors are present in the aggregated error
+				for _, expectedError := range tt.expectedErrors {
+					assert.ErrorIs(t, err, expectedError, "Expected error %v should be present in aggregated error", expectedError)
+				}
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.expectedExists, exists)
+			}
+
+			// Verify all mock expectations were met
+			mockClient.AssertExpectations(t)
+		})
+	}
+}
+
+func TestConnection_Details(t *testing.T) {
+	// Sample stat for successful responses
+	sampleStat := &zk.Stat{
+		Czxid:          123,
+		Mzxid:          124,
+		Ctime:          1234567890,
+		Mtime:          1234567891,
+		Version:        1,
+		Cversion:       0,
+		Aversion:       0,
+		EphemeralOwner: 0,
+		DataLength:     4,
+		NumChildren:    0,
+		Pzxid:          125,
+	}
+
+	tests := []struct {
+		name           string
+		path           string
+		setupMock      func(*MockZKClient)
+		expectedExists bool
+		expectedStat   *zk.Stat
+		expectedErrors []error
+		retries        int
+	}{
+		{
+			name: "success: node exists with stat",
+			path: "/test/path",
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Exists", "/test/path").Return(true, sampleStat, nil).Once()
+			},
+			expectedExists: true,
+			expectedStat:   sampleStat,
+			retries:        1,
+		},
+		{
+			name: "success: node does not exist",
+			path: "/nonexistent/path",
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Exists", "/nonexistent/path").Return(false, (*zk.Stat)(nil), nil).Once()
+			},
+			expectedExists: false,
+			expectedStat:   nil,
+			retries:        1,
+		},
+		{
+			name: "success: retry scenario - 2 errors then success",
+			path: "/test/retry-path",
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Exists", "/test/retry-path").Return(false, (*zk.Stat)(nil), zk.ErrAPIError).Once()
+				mockClient.On("Exists", "/test/retry-path").Return(false, (*zk.Stat)(nil), zk.ErrNoAuth).Once()
+				mockClient.On("Exists", "/test/retry-path").Return(true, sampleStat, nil).Once()
+			},
+			expectedExists: true,
+			expectedStat:   sampleStat,
+			retries:        3,
+		},
+		{
+			name: "error: persistent error, no retries",
+			path: "/test/path",
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Exists", "/test/path").Return(false, (*zk.Stat)(nil), zk.ErrConnectionClosed).Once()
+			},
+			expectedExists: false,
+			expectedStat:   nil,
+			expectedErrors: []error{zk.ErrConnectionClosed},
+			retries:        1,
+		},
+		{
+			name: "error: all retries fail with different errors",
+			path: "/test/path",
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Exists", "/test/path").Return(false, (*zk.Stat)(nil), zk.ErrAPIError).Once()
+				mockClient.On("Exists", "/test/path").Return(false, (*zk.Stat)(nil), zk.ErrNoAuth).Once()
+				mockClient.On("Exists", "/test/path").Return(false, (*zk.Stat)(nil), zk.ErrBadVersion).Once()
+			},
+			expectedExists: false,
+			expectedStat:   nil,
+			expectedErrors: []error{zk.ErrAPIError, zk.ErrNoAuth, zk.ErrBadVersion},
+			retries:        3,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create mock client
+			mockClient := new(MockZKClient)
+
+			// Setup mock expectations
+			tt.setupMock(mockClient)
+
+			// Create connection with mock client
+			conn := newTestConnection(api.ZookeeperNodes{}, mockClient, &ConnectionParams{MaxRetriesNum: tt.retries})
+
+			// Call the method under test
+			ctx := context.Background()
+			exists, stat, err := conn.Details(ctx, tt.path)
+
+			// Assert expectations
+			if len(tt.expectedErrors) > 0 {
+				assert.Error(t, err)
+				// Verify that all expected errors are present in the aggregated error
+				for _, expectedError := range tt.expectedErrors {
+					assert.ErrorIs(t, err, expectedError, "Expected error %v should be present in aggregated error", expectedError)
+				}
+				assert.False(t, exists)
+				assert.Nil(t, stat)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.expectedExists, exists)
+				assert.Equal(t, tt.expectedStat, stat)
+			}
+
+			// Verify all mock expectations were met
+			mockClient.AssertExpectations(t)
+		})
+	}
+}
+
+func TestConnection_Create(t *testing.T) {
+	tests := []struct {
+		name           string
+		path           string
+		value          []byte
+		flags          int32
+		acl            []zk.ACL
+		setupMock      func(*MockZKClient)
+		expectedPath   string
+		expectedErrors []error
+		retries        int
+	}{
+		{
+			name:  "success: create node",
+			path:  "/test/path",
+			value: []byte("test data"),
+			flags: 0,
+			acl:   zk.WorldACL(zk.PermAll),
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Create", "/test/path", []byte("test data"), int32(0), zk.WorldACL(zk.PermAll)).Return("/test/path", nil).Once()
+			},
+			expectedPath: "/test/path",
+			retries:      1,
+		},
+		{
+			name:  "success: retry scenario - 2 errors then success",
+			path:  "/test/retry-path",
+			value: []byte("retry data"),
+			flags: 0,
+			acl:   zk.WorldACL(zk.PermAll),
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Create", "/test/retry-path", []byte("retry data"), int32(0), zk.WorldACL(zk.PermAll)).Return("", zk.ErrAPIError).Once()
+				mockClient.On("Create", "/test/retry-path", []byte("retry data"), int32(0), zk.WorldACL(zk.PermAll)).Return("", zk.ErrNoAuth).Once()
+				mockClient.On("Create", "/test/retry-path", []byte("retry data"), int32(0), zk.WorldACL(zk.PermAll)).Return("/test/retry-path", nil).Once()
+			},
+			expectedPath: "/test/retry-path",
+			retries:      3,
+		},
+		{
+			name:  "error: persistent error",
+			path:  "/test/error",
+			value: []byte("data"),
+			flags: 0,
+			acl:   zk.WorldACL(zk.PermAll),
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Create", "/test/error", []byte("data"), int32(0), zk.WorldACL(zk.PermAll)).Return("", zk.ErrNoNode).Once()
+			},
+			expectedPath:   "",
+			expectedErrors: []error{zk.ErrNoNode},
+			retries:        1,
+		},
+		{
+			name:  "error: all retries fail with different errors",
+			path:  "/test/error-path",
+			value: []byte("data"),
+			flags: 0,
+			acl:   zk.WorldACL(zk.PermAll),
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Create", "/test/error-path", []byte("data"), int32(0), zk.WorldACL(zk.PermAll)).Return("", zk.ErrNodeExists).Once()
+				mockClient.On("Create", "/test/error-path", []byte("data"), int32(0), zk.WorldACL(zk.PermAll)).Return("", zk.ErrAPIError).Once()
+				mockClient.On("Create", "/test/error-path", []byte("data"), int32(0), zk.WorldACL(zk.PermAll)).Return("", zk.ErrNoAuth).Once()
+			},
+			expectedPath:   "",
+			expectedErrors: []error{zk.ErrNodeExists, zk.ErrAPIError, zk.ErrNoAuth},
+			retries:        3,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create mock client
+			mockClient := new(MockZKClient)
+
+			// Setup mock expectations
+			tt.setupMock(mockClient)
+
+			// Create connection with mock client
+			conn := newTestConnection(api.ZookeeperNodes{}, mockClient, &ConnectionParams{MaxRetriesNum: tt.retries})
+
+			// Call the method under test
+			ctx := context.Background()
+			pathCreated, err := conn.Create(ctx, tt.path, tt.value, tt.flags, tt.acl)
+
+			// Assert expectations
+			if len(tt.expectedErrors) > 0 {
+				assert.Error(t, err)
+				// Verify that all expected errors are present in the aggregated error
+				for _, expectedError := range tt.expectedErrors {
+					assert.ErrorIs(t, err, expectedError, "Expected error %v should be present in aggregated error", expectedError)
+				}
+				assert.Empty(t, pathCreated)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.expectedPath, pathCreated)
+			}
+
+			// Verify all mock expectations were met
+			mockClient.AssertExpectations(t)
+		})
+	}
+}
+
+func TestConnection_Set(t *testing.T) {
+	// Sample stat for successful responses
+	sampleStat := &zk.Stat{
+		Czxid:          123,
+		Mzxid:          124,
+		Ctime:          1234567890,
+		Mtime:          1234567891,
+		Version:        2,
+		Cversion:       0,
+		Aversion:       0,
+		EphemeralOwner: 0,
+		DataLength:     9,
+		NumChildren:    0,
+		Pzxid:          125,
+	}
+
+	tests := []struct {
+		name           string
+		path           string
+		value          []byte
+		version        int32
+		setupMock      func(*MockZKClient)
+		expectedStat   *zk.Stat
+		expectedErrors []error
+		retries        int
+	}{
+		{
+			name:    "success: set data",
+			path:    "/test/path",
+			value:   []byte("new data"),
+			version: 1,
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Set", "/test/path", []byte("new data"), int32(1)).Return(sampleStat, nil).Once()
+			},
+			expectedStat: sampleStat,
+			retries:      1,
+		},
+		{
+			name:    "success: retry scenario - 2 errors then success",
+			path:    "/test/retry-path",
+			value:   []byte("retry data"),
+			version: 1,
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Set", "/test/retry-path", []byte("retry data"), int32(1)).Return((*zk.Stat)(nil), zk.ErrAPIError).Once()
+				mockClient.On("Set", "/test/retry-path", []byte("retry data"), int32(1)).Return((*zk.Stat)(nil), zk.ErrNoAuth).Once()
+				mockClient.On("Set", "/test/retry-path", []byte("retry data"), int32(1)).Return(sampleStat, nil).Once()
+			},
+			expectedStat: sampleStat,
+			retries:      3,
+		},
+		{
+			name:    "error: node not found",
+			path:    "/nonexistent/path",
+			value:   []byte("data"),
+			version: -1,
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Set", "/nonexistent/path", []byte("data"), int32(-1)).Return((*zk.Stat)(nil), zk.ErrNoNode).Once()
+			},
+			expectedStat:   nil,
+			expectedErrors: []error{zk.ErrNoNode},
+			retries:        1,
+		},
+		{
+			name:    "error: persistent connection error",
+			path:    "/test/path",
+			value:   []byte("data"),
+			version: 1,
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Set", "/test/path", []byte("data"), int32(1)).Return((*zk.Stat)(nil), zk.ErrConnectionClosed).Once()
+			},
+			expectedStat:   nil,
+			expectedErrors: []error{zk.ErrConnectionClosed},
+			retries:        1,
+		},
+		{
+			name:    "error: all retries fail with different errors",
+			path:    "/test/error-path",
+			value:   []byte("data"),
+			version: 1,
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Set", "/test/error-path", []byte("data"), int32(1)).Return((*zk.Stat)(nil), zk.ErrBadVersion).Once()
+				mockClient.On("Set", "/test/error-path", []byte("data"), int32(1)).Return((*zk.Stat)(nil), zk.ErrNoNode).Once()
+				mockClient.On("Set", "/test/error-path", []byte("data"), int32(1)).Return((*zk.Stat)(nil), zk.ErrAPIError).Once()
+			},
+			expectedStat:   nil,
+			expectedErrors: []error{zk.ErrBadVersion, zk.ErrNoNode, zk.ErrAPIError},
+			retries:        3,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create mock client
+			mockClient := new(MockZKClient)
+
+			// Setup mock expectations
+			tt.setupMock(mockClient)
+
+			// Create connection with mock client
+			conn := newTestConnection(api.ZookeeperNodes{}, mockClient, &ConnectionParams{MaxRetriesNum: tt.retries})
+
+			// Call the method under test
+			ctx := context.Background()
+			stat, err := conn.Set(ctx, tt.path, tt.value, tt.version)
+
+			// Assert expectations
+			if len(tt.expectedErrors) > 0 {
+				assert.Error(t, err)
+				// Verify that all expected errors are present in the aggregated error
+				for _, expectedError := range tt.expectedErrors {
+					assert.ErrorIs(t, err, expectedError, "Expected error %v should be present in aggregated error", expectedError)
+				}
+				assert.Nil(t, stat)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.expectedStat, stat)
+			}
+
+			// Verify all mock expectations were met
+			mockClient.AssertExpectations(t)
+		})
+	}
+}
+
+func TestConnection_Delete(t *testing.T) {
+	tests := []struct {
+		name           string
+		path           string
+		version        int32
+		setupMock      func(*MockZKClient)
+		expectedErrors []error
+		retries        int
+	}{
+		{
+			name:    "success: delete",
+			path:    "/test/path",
+			version: 1,
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Delete", "/test/path", int32(1)).Return(nil).Once()
+			},
+			retries: 1,
+		},
+		{
+			name:    "success: retry scenario - 2 errors then success",
+			path:    "/test/retry-path",
+			version: 1,
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Delete", "/test/retry-path", int32(1)).Return(zk.ErrAPIError).Once()
+				mockClient.On("Delete", "/test/retry-path", int32(1)).Return(zk.ErrNoAuth).Once()
+				mockClient.On("Delete", "/test/retry-path", int32(1)).Return(nil).Once()
+			},
+			retries: 3,
+		},
+		{
+			name:    "error: persistent connection error",
+			path:    "/test/path",
+			version: 1,
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Delete", "/test/path", int32(1)).Return(zk.ErrConnectionClosed).Once()
+			},
+			expectedErrors: []error{zk.ErrConnectionClosed},
+			retries:        1,
+		},
+		{
+			name:    "error: all retries fail with different errors",
+			path:    "/test/error-path",
+			version: 1,
+			setupMock: func(mockClient *MockZKClient) {
+				mockClient.On("Delete", "/test/error-path", int32(1)).Return(zk.ErrNotEmpty).Once()
+				mockClient.On("Delete", "/test/error-path", int32(1)).Return(zk.ErrBadVersion).Once()
+				mockClient.On("Delete", "/test/error-path", int32(1)).Return(zk.ErrAPIError).Once()
+			},
+			expectedErrors: []error{zk.ErrNotEmpty, zk.ErrBadVersion, zk.ErrAPIError},
+			retries:        3,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create mock client
+			mockClient := new(MockZKClient)
+
+			// Setup mock expectations
+			tt.setupMock(mockClient)
+
+			// Create connection with mock client
+			conn := newTestConnection(api.ZookeeperNodes{}, mockClient, &ConnectionParams{MaxRetriesNum: tt.retries})
+
+			// Call the method under test
+			ctx := context.Background()
+			err := conn.Delete(ctx, tt.path, tt.version)
+
+			// Assert expectations
+			if len(tt.expectedErrors) > 0 {
+				assert.Error(t, err)
+				// Verify that all expected errors are present in the aggregated error
+				for _, expectedError := range tt.expectedErrors {
+					assert.ErrorIs(t, err, expectedError, "Expected error %v should be present in aggregated error", expectedError)
+				}
+			} else {
+				assert.NoError(t, err)
+			}
+
+			// Verify all mock expectations were met
+			mockClient.AssertExpectations(t)
+		})
+	}
+}
+
+func TestConnection_Close(t *testing.T) {
+	tests := []struct {
+		name  string
+		setup func(*MockZKClient) *Connection
+	}{
+		{
+			name: "success: close connection",
+			setup: func(mockClient *MockZKClient) *Connection {
+				mockClient.On("Close").Return().Once()
+				return newTestConnection(api.ZookeeperNodes{}, mockClient, &ConnectionParams{MaxRetriesNum: 1})
+			},
+		},
+		{
+			name: "success: handle empty connection",
+			setup: func(mockClient *MockZKClient) *Connection {
+				return nil
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create mock client
+			mockClient := new(MockZKClient)
+
+			// Setup mock expectations
+			conn := tt.setup(mockClient)
+
+			// Call the method under test
+			err := conn.Close()
+
+			// Assert expectations - Close should always succeed
+			assert.NoError(t, err)
+
+			// Verify all mock expectations were met
+			mockClient.AssertExpectations(t)
+		})
+	}
+}
+
+// newTestConnection creates a Connection with a custom ZKClient for testing
+func newTestConnection(nodes api.ZookeeperNodes, client ZKClient, _params ...*ConnectionParams) *Connection {
+	conn := NewConnection(nodes, _params...)
+	conn.connection = client
+	conn.retryDelayFn = func(i int) {} // Disable retry delay for tests
+	return conn
+}
+
+// MockZKClient is a mock implementation of the ZKClient interface for testing
+type MockZKClient struct {
+	mock.Mock
+}
+
+func (m *MockZKClient) Get(path string) ([]byte, *zk.Stat, error) {
+	args := m.Called(path)
+	return args.Get(0).([]byte), args.Get(1).(*zk.Stat), args.Error(2)
+}
+
+func (m *MockZKClient) Set(path string, data []byte, version int32) (*zk.Stat, error) {
+	args := m.Called(path, data, version)
+	return args.Get(0).(*zk.Stat), args.Error(1)
+}
+
+func (m *MockZKClient) Create(path string, data []byte, flags int32, acl []zk.ACL) (string, error) {
+	args := m.Called(path, data, flags, acl)
+	return args.String(0), args.Error(1)
+}
+
+func (m *MockZKClient) Delete(path string, version int32) error {
+	args := m.Called(path, version)
+	return args.Error(0)
+}
+
+func (m *MockZKClient) Exists(path string) (bool, *zk.Stat, error) {
+	args := m.Called(path)
+	return args.Bool(0), args.Get(1).(*zk.Stat), args.Error(2)
+}
+
+func (m *MockZKClient) AddAuth(scheme string, auth []byte) error {
+	args := m.Called(scheme, auth)
+	return args.Error(0)
+}
+
+func (m *MockZKClient) Close() {
+	m.Called()
+}
diff --git a/tests/e2e/kubectl.py b/tests/e2e/kubectl.py
index 6e0a907a9..fc4a07ece 100644
--- a/tests/e2e/kubectl.py
+++ b/tests/e2e/kubectl.py
@@ -214,7 +214,19 @@ def create_and_check(manifest, check, kind="chi", ns=None, shell=None, timeout=1
 
 def get(kind, name, label="", ns=None, ok_to_fail=False, shell=None):
     out = launch(f"get {kind} {name} {label} -o json", ns=ns, ok_to_fail=ok_to_fail, shell=shell)
-    return json.loads(out.strip())
+    stripped = out.strip()
+
+    if not stripped or stripped.startswith("Error"):
+        if ok_to_fail:
+            return None
+        raise ValueError(f"kubectl returned error: {stripped}")
+
+    try:
+        return json.loads(stripped)
+    except json.JSONDecodeError as e:
+        if ok_to_fail:
+            return None
+        raise ValueError(f"Failed to parse JSON from: {stripped}") from e
 
 
 def get_chi_normalizedCompleted(chi, ns=None, shell=None):
@@ -647,13 +659,26 @@ def check_configmap(cfg_name, values, ns=None, shell=None):
 def check_pdb(chi, clusters, ns=None, shell=None):
     for c in clusters.keys():
         with Then(f"PDB is configured for cluster {c}"):
-            pdb = get("pdb", chi + "-" + c, shell=shell)
+            is_managed = True
+            if isinstance(clusters[c], dict):
+                is_managed = clusters[c].get("is_managed", True)
+                max_unavailable = clusters[c].get("max_`unavailable`", 1)
+            else:
+                # Treat simple integer as maxUnavailable to ensure backward compatibility.
+                max_unavailable = clusters[c]
+
+            pdb = get("pdb", chi + "-" + c, ok_to_fail=is_managed is False, shell=shell)
+
+            if not is_managed:
+                assert pdb is None
+                continue
+
             labels = pdb["spec"]["selector"]["matchLabels"]
             assert labels["clickhouse.altinity.com/app"] == "chop"
             assert labels["clickhouse.altinity.com/chi"] == chi
             assert labels["clickhouse.altinity.com/cluster"] == c
             assert labels["clickhouse.altinity.com/namespace"] == current().context.test_namespace
-            assert pdb["spec"]["maxUnavailable"] == clusters[c]
+            assert pdb["spec"]["maxUnavailable"] == max_unavailable
 
 def force_reconcile(chi, taskID="reconcile", ns=None, shell=None):
     with Then(f"Trigger CHI reconcile with taskID:\"{taskID}\""):
diff --git a/tests/e2e/manifests/chi/test-060-pdb-management-disabled.yaml b/tests/e2e/manifests/chi/test-060-pdb-management-disabled.yaml
new file mode 100644
index 000000000..fe942f006
--- /dev/null
+++ b/tests/e2e/manifests/chi/test-060-pdb-management-disabled.yaml
@@ -0,0 +1,11 @@
+apiVersion: "clickhouse.altinity.com/v1"
+kind: "ClickHouseInstallation"
+metadata:
+  name: "test-060-pdb-management-disabled"
+spec:
+  configuration:
+    clusters:
+      - name: single
+        pdbManaged: "no"
+        secret:
+          auto: "yes"
diff --git a/tests/e2e/test_operator.py b/tests/e2e/test_operator.py
index 93eec11b8..e724c5d4d 100644
--- a/tests/e2e/test_operator.py
+++ b/tests/e2e/test_operator.py
@@ -5632,6 +5632,36 @@ def test_059(self):
     with Finally("I clean up"):
         delete_test_namespace()
 
+@TestScenario
+@Name("test_060. pdb management disabled")
+@Requirements(RQ_SRS_026_ClickHouseOperator_Create("1.0"))
+def test_060(self):
+    create_shell_namespace_clickhouse_template()
+
+    chi = "test-060-pdb-management-disabled"
+    kubectl.create_and_check(
+        manifest="manifests/chi/test-060-pdb-management-disabled.yaml",
+        check={
+            "object_counts": {
+                "statefulset": 1,
+                "pod": 1,
+                "service": 2,
+            },
+            "configmaps": 1,
+            "pdb": {"single": {"is_managed": False}},
+            "do_not_delete": 1,
+        },
+    )
+
+    created_objects = kubectl.get_obj_names_grepped("pod,service,sts,pvc,cm,pdb,secret", grep=chi)
+    print("Created objects:")
+    for o in created_objects:
+        print(o)
+
+    kubectl.delete_chi(chi)
+
+    with Finally("I clean up"):
+        delete_test_namespace()
 
 def cleanup_chis(self):
     with Given("Cleanup CHIs"):