fix: Issues with paused rollouts and test cases

internal/pkg/controller/configmap_reconciler.go

@@ -30,6 +30,7 @@ type ConfigMapReconciler struct {
 	EventRecorder *events.Recorder
 	WebhookClient *webhook.Client
 	Alerter       alerting.Alerter
+	PauseHandler  *reload.PauseHandler
 
 	handler     *ReloadHandler
 	initialized bool
@@ -100,6 +101,7 @@ func (r *ConfigMapReconciler) reloadHandler() *ReloadHandler {
 			Collectors:    r.Collectors,
 			EventRecorder: r.EventRecorder,
 			Alerter:       r.Alerter,
+			PauseHandler:  r.PauseHandler,
 		}
 	}
 	return r.handler

internal/pkg/controller/deployment_reconciler.go

@@ -54,11 +54,18 @@ func (r *DeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		return ctrl.Result{RequeueAfter: remainingTime}, nil
 	}
 
-	// Pause period has expired - unpause the deployment
 	log.Info("Unpausing deployment after pause period expired")
-	r.PauseHandler.ClearPause(&deploy)
+	err = UpdateObjectWithRetry(
+		ctx, r.Client, &deploy, func() (bool, error) {
+			if !r.PauseHandler.IsPausedByReloader(&deploy) {
+				return false, nil
+			}
+			r.PauseHandler.ClearPause(&deploy)
+			return true, nil
+		},
+	)
 
-	if err := r.Update(ctx, &deploy, client.FieldOwner(FieldManager)); err != nil {
+	if err != nil {
 		log.Error(err, "Failed to unpause deployment")
 		return ctrl.Result{}, err
 	}

internal/pkg/controller/handler.go

@@ -24,6 +24,7 @@ type ReloadHandler struct {
 	Collectors    *metrics.Collectors
 	EventRecorder *events.Recorder
 	Alerter       alerting.Alerter
+	PauseHandler  *reload.PauseHandler
 }
 
 // Process handles the reload workflow: list workloads, get decisions, webhook or apply.
@@ -112,6 +113,7 @@ func (h *ReloadHandler) applyReloads(
 			ctx,
 			h.Client,
 			h.ReloadService,
+			h.PauseHandler,
 			decision.Workload,
 			resourceName,
 			resourceType,

internal/pkg/controller/manager.go

@@ -150,6 +150,7 @@ func SetupReconcilers(mgr ctrl.Manager, cfg *config.Config, log logr.Logger, col
 			EventRecorder: eventRecorder,
 			WebhookClient: webhookClient,
 			Alerter:       alerter,
+			PauseHandler:  pauseHandler,
 		}).SetupWithManager(mgr); err != nil {
 			return fmt.Errorf("setting up configmap reconciler: %w", err)
 		}
@@ -167,6 +168,7 @@ func SetupReconcilers(mgr ctrl.Manager, cfg *config.Config, log logr.Logger, col
 			EventRecorder: eventRecorder,
 			WebhookClient: webhookClient,
 			Alerter:       alerter,
+			PauseHandler:  pauseHandler,
 		}).SetupWithManager(mgr); err != nil {
 			return fmt.Errorf("setting up secret reconciler: %w", err)
 		}

internal/pkg/controller/retry.go

@@ -13,6 +13,39 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
+// UpdateObjectWithRetry updates a Kubernetes object with retry on conflict.
+// It re-fetches the object on each retry attempt and calls modifyFn to apply changes.
+// The modifyFn receives the latest version of the object and should modify it in place.
+// If modifyFn returns false, the update is skipped (e.g., if the condition no longer applies).
+func UpdateObjectWithRetry(
+	ctx context.Context,
+	c client.Client,
+	obj client.Object,
+	modifyFn func() (shouldUpdate bool, err error),
+) error {
+	return retry.RetryOnConflict(
+		retry.DefaultBackoff, func() error {
+			if err := c.Get(ctx, client.ObjectKeyFromObject(obj), obj); err != nil {
+				if errors.IsNotFound(err) {
+					return nil
+				}
+				return err
+			}
+
+			shouldUpdate, err := modifyFn()
+			if err != nil {
+				return err
+			}
+
+			if !shouldUpdate {
+				return nil
+			}
+
+			return c.Update(ctx, obj, client.FieldOwner(FieldManager))
+		},
+	)
+}
+
 // UpdateWorkloadWithRetry updates a workload with exponential backoff on conflict.
 // On conflict, it re-fetches the object, re-applies the reload changes, and retries.
 // For Jobs and CronJobs, special handling is applied:
@@ -23,6 +56,7 @@ func UpdateWorkloadWithRetry(
 	ctx context.Context,
 	c client.Client,
 	reloadService *reload.Service,
+	pauseHandler *reload.PauseHandler,
 	wl workload.WorkloadAccessor,
 	resourceName string,
 	resourceType reload.ResourceType,
@@ -38,6 +72,8 @@ func UpdateWorkloadWithRetry(
 		return updateCronJobWithNewJob(ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload)
 	case workload.KindArgoRollout:
 		return updateArgoRollout(ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload)
+	case workload.KindDeployment:
+		return updateDeploymentWithPause(ctx, c, reloadService, pauseHandler, wl, resourceName, resourceType, namespace, hash, autoReload)
 	default:
 		return updateStandardWorkload(ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload)
 	}
@@ -60,36 +96,38 @@ func retryWithReload(
 	var updated bool
 	isFirstAttempt := true
 
-	err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
-		if !isFirstAttempt {
-			obj := wl.GetObject()
-			key := client.ObjectKeyFromObject(obj)
-			if err := c.Get(ctx, key, obj); err != nil {
-				if errors.IsNotFound(err) {
-					return nil
+	err := retry.RetryOnConflict(
+		retry.DefaultBackoff, func() error {
+			if !isFirstAttempt {
+				obj := wl.GetObject()
+				key := client.ObjectKeyFromObject(obj)
+				if err := c.Get(ctx, key, obj); err != nil {
+					if errors.IsNotFound(err) {
+						return nil
+					}
+					return err
 				}
-				return err
 			}
-		}
-		isFirstAttempt = false
+			isFirstAttempt = false
 
-		var applyErr error
-		updated, applyErr = reloadService.ApplyReload(ctx, wl, resourceName, resourceType, namespace, hash, autoReload)
-		if applyErr != nil {
-			return applyErr
-		}
+			var applyErr error
+			updated, applyErr = reloadService.ApplyReload(ctx, wl, resourceName, resourceType, namespace, hash, autoReload)
+			if applyErr != nil {
+				return applyErr
+			}
 
-		if !updated {
-			return nil
-		}
+			if !updated {
+				return nil
+			}
 
-		return updateFn()
-	})
+			return updateFn()
+		},
+	)
 
 	return updated, err
 }
 
-// updateStandardWorkload updates Deployments, DaemonSets, StatefulSets, etc.
+// updateStandardWorkload updates DaemonSets, StatefulSets, etc.
 func updateStandardWorkload(
 	ctx context.Context,
 	c client.Client,
@@ -101,10 +139,40 @@ func updateStandardWorkload(
 	hash string,
 	autoReload bool,
 ) (bool, error) {
-	return retryWithReload(ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload,
+	return retryWithReload(
+		ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload,
 		func() error {
 			return c.Update(ctx, wl.GetObject(), client.FieldOwner(FieldManager))
-		})
+		},
+	)
+}
+
+// updateDeploymentWithPause updates a Deployment and applies pause if configured.
+func updateDeploymentWithPause(
+	ctx context.Context,
+	c client.Client,
+	reloadService *reload.Service,
+	pauseHandler *reload.PauseHandler,
+	wl workload.WorkloadAccessor,
+	resourceName string,
+	resourceType reload.ResourceType,
+	namespace string,
+	hash string,
+	autoReload bool,
+) (bool, error) {
+	shouldPause := pauseHandler != nil && pauseHandler.ShouldPause(wl)
+
+	return retryWithReload(
+		ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload,
+		func() error {
+			if shouldPause {
+				if err := pauseHandler.ApplyPause(wl); err != nil {
+					return err
+				}
+			}
+			return c.Update(ctx, wl.GetObject(), client.FieldOwner(FieldManager))
+		},
+	)
 }
 
 // updateJobWithRecreate deletes the Job and recreates it with the updated spec.
@@ -148,9 +216,11 @@ func updateJobWithRecreate(
 
 	// Delete the old job with background propagation
 	policy := metav1.DeletePropagationBackground
-	if err := c.Delete(ctx, oldJob, &client.DeleteOptions{
-		PropagationPolicy: &policy,
-	}); err != nil {
+	if err := c.Delete(
+		ctx, oldJob, &client.DeleteOptions{
+			PropagationPolicy: &policy,
+		},
+	); err != nil {
 		if !errors.IsNotFound(err) {
 			return false, err
 		}
@@ -217,12 +287,10 @@ func updateCronJobWithNewJob(
 
 	cronJob := cronJobWl.GetCronJob()
 
-	// Build annotations for the new Job
 	annotations := make(map[string]string)
 	annotations["cronjob.kubernetes.io/instantiate"] = "manual"
 	maps.Copy(annotations, cronJob.Spec.JobTemplate.Annotations)
 
-	// Create a new Job from the CronJob template
 	job := &batchv1.Job{
 		ObjectMeta: metav1.ObjectMeta{
 			GenerateName: cronJob.Name + "-",
@@ -240,6 +308,22 @@ func updateCronJobWithNewJob(
 		return false, err
 	}
 
+	savedAnnotations := maps.Clone(cronJob.Spec.JobTemplate.Spec.Template.Annotations)
+
+	err = UpdateObjectWithRetry(
+		ctx, c, cronJob, func() (bool, error) {
+			if cronJob.Spec.JobTemplate.Spec.Template.Annotations == nil {
+				cronJob.Spec.JobTemplate.Spec.Template.Annotations = make(map[string]string)
+			}
+			maps.Copy(cronJob.Spec.JobTemplate.Spec.Template.Annotations, savedAnnotations)
+			return true, nil
+		},
+	)
+
+	if err != nil {
+		return false, err
+	}
+
 	return true, nil
 }
 
@@ -262,8 +346,10 @@ func updateArgoRollout(
 		return false, nil
 	}
 
-	return retryWithReload(ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload,
+	return retryWithReload(
+		ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload,
 		func() error {
 			return rolloutWl.Update(ctx, c)
-		})
+		},
+	)
 }

internal/pkg/controller/retry_test.go

@@ -133,6 +133,7 @@ func TestUpdateWorkloadWithRetry_WorkloadTypes(t *testing.T) {
 					context.Background(),
 					fakeClient,
 					reloadService,
+					nil, // no pause handler
 					wl,
 					"test-resource",
 					tt.resourceType,
@@ -214,6 +215,7 @@ func TestUpdateWorkloadWithRetry_Strategies(t *testing.T) {
 					context.Background(),
 					fakeClient,
 					reloadService,
+					nil, // no pause handler for this test
 					wl,
 					"test-cm",
 					reload.ResourceTypeConfigMap,
@@ -265,6 +267,7 @@ func TestUpdateWorkloadWithRetry_NoUpdate(t *testing.T) {
 		context.Background(),
 		fakeClient,
 		reloadService,
+		nil, // no pause handler
 		wl,
 		"test-cm",
 		reload.ResourceTypeConfigMap,
@@ -300,3 +303,283 @@ func TestResourceTypeKind(t *testing.T) {
 		)
 	}
 }
+
+func TestUpdateWorkloadWithRetry_PauseDeployment(t *testing.T) {
+	cfg := config.NewDefault()
+	reloadService := reload.NewService(cfg)
+	pauseHandler := reload.NewPauseHandler(cfg)
+
+	deployment := testutil.NewDeployment(
+		"test-deployment", "default", map[string]string{
+			"reloader.stakater.com/auto":                    "true",
+			"deployment.reloader.stakater.com/pause-period": "5m",
+		},
+	)
+
+	fakeClient := fake.NewClientBuilder().
+		WithScheme(testutil.NewScheme()).
+		WithObjects(deployment).
+		Build()
+
+	wl := workload.NewDeploymentWorkload(deployment)
+
+	updated, err := controller.UpdateWorkloadWithRetry(
+		context.Background(),
+		fakeClient,
+		reloadService,
+		pauseHandler,
+		wl,
+		"test-cm",
+		reload.ResourceTypeConfigMap,
+		"default",
+		"abc123",
+		true,
+	)
+
+	if err != nil {
+		t.Fatalf("UpdateWorkloadWithRetry failed: %v", err)
+	}
+	if !updated {
+		t.Error("Expected workload to be updated")
+	}
+
+	var result appsv1.Deployment
+	if err := fakeClient.Get(
+		context.Background(), types.NamespacedName{Name: "test-deployment", Namespace: "default"}, &result,
+	); err != nil {
+		t.Fatalf("Failed to get deployment: %v", err)
+	}
+
+	if result.Spec.Template.Annotations == nil {
+		t.Fatal("Expected pod template annotations to be set")
+	}
+
+	if !result.Spec.Paused {
+		t.Error("Expected deployment to be paused (spec.Paused=true)")
+	}
+
+	pausedAt := result.Annotations[cfg.Annotations.PausedAt]
+	if pausedAt == "" {
+		t.Error("Expected paused-at annotation to be set")
+	}
+}
+
+// TestUpdateWorkloadWithRetry_PauseWithExplicitAnnotation tests pause with explicit configmap annotation (no auto).
+func TestUpdateWorkloadWithRetry_PauseWithExplicitAnnotation(t *testing.T) {
+	cfg := config.NewDefault()
+	reloadService := reload.NewService(cfg)
+	pauseHandler := reload.NewPauseHandler(cfg)
+
+	deployment := testutil.NewDeployment(
+		"test-deployment", "default", map[string]string{
+			cfg.Annotations.ConfigmapReload: "test-cm", // explicit, not auto
+			cfg.Annotations.PausePeriod:     "5m",
+		},
+	)
+
+	fakeClient := fake.NewClientBuilder().
+		WithScheme(testutil.NewScheme()).
+		WithObjects(deployment).
+		Build()
+
+	wl := workload.NewDeploymentWorkload(deployment)
+
+	updated, err := controller.UpdateWorkloadWithRetry(
+		context.Background(),
+		fakeClient,
+		reloadService,
+		pauseHandler,
+		wl,
+		"test-cm",
+		reload.ResourceTypeConfigMap,
+		"default",
+		"abc123",
+		false, // NOT auto reload
+	)
+
+	if err != nil {
+		t.Fatalf("UpdateWorkloadWithRetry failed: %v", err)
+	}
+	if !updated {
+		t.Error("Expected workload to be updated")
+	}
+
+	var result appsv1.Deployment
+	if err := fakeClient.Get(
+		context.Background(), types.NamespacedName{Name: "test-deployment", Namespace: "default"}, &result,
+	); err != nil {
+		t.Fatalf("Failed to get deployment: %v", err)
+	}
+
+	if result.Spec.Template.Annotations == nil {
+		t.Fatal("Expected pod template annotations to be set")
+	}
+
+	if !result.Spec.Paused {
+		t.Error("Expected deployment to be paused (spec.Paused=true)")
+	}
+
+	pausedAt := result.Annotations[cfg.Annotations.PausedAt]
+	if pausedAt == "" {
+		t.Error("Expected paused-at annotation to be set")
+	}
+}
+
+// TestUpdateWorkloadWithRetry_PauseWithSecretReload tests pause with Secret-triggered reload.
+func TestUpdateWorkloadWithRetry_PauseWithSecretReload(t *testing.T) {
+	cfg := config.NewDefault()
+	reloadService := reload.NewService(cfg)
+	pauseHandler := reload.NewPauseHandler(cfg)
+
+	deployment := testutil.NewDeployment(
+		"test-deployment", "default", map[string]string{
+			cfg.Annotations.SecretReload: "test-secret", // explicit secret, not auto
+			cfg.Annotations.PausePeriod:  "5m",
+		},
+	)
+
+	fakeClient := fake.NewClientBuilder().
+		WithScheme(testutil.NewScheme()).
+		WithObjects(deployment).
+		Build()
+
+	wl := workload.NewDeploymentWorkload(deployment)
+
+	updated, err := controller.UpdateWorkloadWithRetry(
+		context.Background(),
+		fakeClient,
+		reloadService,
+		pauseHandler,
+		wl,
+		"test-secret",
+		reload.ResourceTypeSecret,
+		"default",
+		"abc123",
+		false,
+	)
+
+	if err != nil {
+		t.Fatalf("UpdateWorkloadWithRetry failed: %v", err)
+	}
+	if !updated {
+		t.Error("Expected workload to be updated")
+	}
+
+	var result appsv1.Deployment
+	if err := fakeClient.Get(
+		context.Background(), types.NamespacedName{Name: "test-deployment", Namespace: "default"}, &result,
+	); err != nil {
+		t.Fatalf("Failed to get deployment: %v", err)
+	}
+
+	if !result.Spec.Paused {
+		t.Error("Expected deployment to be paused (spec.Paused=true)")
+	}
+
+	pausedAt := result.Annotations[cfg.Annotations.PausedAt]
+	if pausedAt == "" {
+		t.Error("Expected paused-at annotation to be set")
+	}
+}
+
+// TestUpdateWorkloadWithRetry_PauseWithAutoSecret tests pause with auto annotation + Secret change.
+func TestUpdateWorkloadWithRetry_PauseWithAutoSecret(t *testing.T) {
+	cfg := config.NewDefault()
+	reloadService := reload.NewService(cfg)
+	pauseHandler := reload.NewPauseHandler(cfg)
+
+	deployment := testutil.NewDeployment(
+		"test-deployment", "default", map[string]string{
+			cfg.Annotations.Auto:        "true",
+			cfg.Annotations.PausePeriod: "5m",
+		},
+	)
+
+	fakeClient := fake.NewClientBuilder().
+		WithScheme(testutil.NewScheme()).
+		WithObjects(deployment).
+		Build()
+
+	wl := workload.NewDeploymentWorkload(deployment)
+
+	updated, err := controller.UpdateWorkloadWithRetry(
+		context.Background(),
+		fakeClient,
+		reloadService,
+		pauseHandler,
+		wl,
+		"test-secret",
+		reload.ResourceTypeSecret,
+		"default",
+		"abc123",
+		true,
+	)
+
+	if err != nil {
+		t.Fatalf("UpdateWorkloadWithRetry failed: %v", err)
+	}
+	if !updated {
+		t.Error("Expected workload to be updated")
+	}
+
+	var result appsv1.Deployment
+	if err := fakeClient.Get(
+		context.Background(), types.NamespacedName{Name: "test-deployment", Namespace: "default"}, &result,
+	); err != nil {
+		t.Fatalf("Failed to get deployment: %v", err)
+	}
+
+	if !result.Spec.Paused {
+		t.Error("Expected deployment to be paused (spec.Paused=true)")
+	}
+}
+
+func TestUpdateWorkloadWithRetry_NoPauseWithoutAnnotation(t *testing.T) {
+	cfg := config.NewDefault()
+	reloadService := reload.NewService(cfg)
+	pauseHandler := reload.NewPauseHandler(cfg)
+
+	deployment := testutil.NewDeployment(
+		"test-deployment", "default", map[string]string{
+			"reloader.stakater.com/auto": "true",
+		},
+	)
+
+	fakeClient := fake.NewClientBuilder().
+		WithScheme(testutil.NewScheme()).
+		WithObjects(deployment).
+		Build()
+
+	wl := workload.NewDeploymentWorkload(deployment)
+
+	updated, err := controller.UpdateWorkloadWithRetry(
+		context.Background(),
+		fakeClient,
+		reloadService,
+		pauseHandler,
+		wl,
+		"test-cm",
+		reload.ResourceTypeConfigMap,
+		"default",
+		"abc123",
+		true,
+	)
+
+	if err != nil {
+		t.Fatalf("UpdateWorkloadWithRetry failed: %v", err)
+	}
+	if !updated {
+		t.Error("Expected workload to be updated")
+	}
+
+	var result appsv1.Deployment
+	if err := fakeClient.Get(
+		context.Background(), types.NamespacedName{Name: "test-deployment", Namespace: "default"}, &result,
+	); err != nil {
+		t.Fatalf("Failed to get deployment: %v", err)
+	}
+
+	if result.Spec.Paused {
+		t.Error("Expected deployment NOT to be paused (no pause-period annotation)")
+	}
+}

internal/pkg/controller/secret_reconciler.go

@@ -30,6 +30,7 @@ type SecretReconciler struct {
 	EventRecorder *events.Recorder
 	WebhookClient *webhook.Client
 	Alerter       alerting.Alerter
+	PauseHandler  *reload.PauseHandler
 
 	handler     *ReloadHandler
 	initialized bool
@@ -97,6 +98,7 @@ func (r *SecretReconciler) reloadHandler() *ReloadHandler {
 			Collectors:    r.Collectors,
 			EventRecorder: r.EventRecorder,
 			Alerter:       r.Alerter,
+			PauseHandler:  r.PauseHandler,
 		}
 	}
 	return r.handler

internal/pkg/testutil/testutil.go

@@ -430,3 +430,51 @@ func WaitForDeploymentConfigReloadedAnnotation(client openshiftclient.Interface,
 	}
 	return found, err
 }
+
+// WaitForDeploymentPaused waits for a deployment to be paused (spec.Paused=true).
+func WaitForDeploymentPaused(client kubernetes.Interface, namespace, name string, timeout time.Duration) (bool, error) {
+	var paused bool
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	err := wait.PollUntilContextTimeout(
+		ctx, time.Second, timeout, true, func(ctx context.Context) (bool, error) {
+			deployment, err := client.AppsV1().Deployments(namespace).Get(ctx, name, metav1.GetOptions{})
+			if err != nil {
+				return false, nil // Keep waiting
+			}
+			if deployment.Spec.Paused {
+				paused = true
+				return true, nil
+			}
+			return false, nil
+		},
+	)
+	if wait.Interrupted(err) {
+		return paused, nil
+	}
+	return paused, err
+}
+
+// WaitForDeploymentUnpaused waits for a deployment to be unpaused (spec.Paused=false).
+func WaitForDeploymentUnpaused(client kubernetes.Interface, namespace, name string, timeout time.Duration) (bool, error) {
+	var unpaused bool
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	err := wait.PollUntilContextTimeout(
+		ctx, time.Second, timeout, true, func(ctx context.Context) (bool, error) {
+			deployment, err := client.AppsV1().Deployments(namespace).Get(ctx, name, metav1.GetOptions{})
+			if err != nil {
+				return false, nil // Keep waiting
+			}
+			if !deployment.Spec.Paused {
+				unpaused = true
+				return true, nil
+			}
+			return false, nil
+		},
+	)
+	if wait.Interrupted(err) {
+		return unpaused, nil
+	}
+	return unpaused, err
+}

test/e2e/e2e_test.go

@@ -228,6 +228,46 @@ func (f *testFixture) assertDeploymentConfigReloaded(name string) {
 	}
 }
 
+// assertDeploymentPaused asserts that a deployment is paused (spec.Paused=true).
+func (f *testFixture) assertDeploymentPaused(name string) {
+	f.t.Helper()
+	paused, err := testutil.WaitForDeploymentPaused(k8sClient, namespace, name, waitTimeout)
+	if err != nil {
+		f.t.Fatalf("Error waiting for deployment %s to be paused: %v", name, err)
+	}
+	if !paused {
+		f.t.Errorf("Deployment %s was not paused after reload", name)
+	}
+}
+
+// assertDeploymentUnpaused asserts that a deployment is unpaused (spec.Paused=false).
+func (f *testFixture) assertDeploymentUnpaused(name string, timeout time.Duration) {
+	f.t.Helper()
+	unpaused, err := testutil.WaitForDeploymentUnpaused(k8sClient, namespace, name, timeout)
+	if err != nil {
+		f.t.Fatalf("Error waiting for deployment %s to be unpaused: %v", name, err)
+	}
+	if !unpaused {
+		f.t.Errorf("Deployment %s was not unpaused after pause period", name)
+	}
+}
+
+// assertDeploymentHasPausedAtAnnotation asserts that a deployment has the paused-at annotation.
+func (f *testFixture) assertDeploymentHasPausedAtAnnotation(name string) {
+	f.t.Helper()
+	deploy, err := k8sClient.AppsV1().Deployments(namespace).Get(context.Background(), name, metav1.GetOptions{})
+	if err != nil {
+		f.t.Fatalf("Failed to get deployment %s: %v", name, err)
+	}
+	if deploy.Annotations == nil {
+		f.t.Errorf("Deployment %s has no annotations", name)
+		return
+	}
+	if _, ok := deploy.Annotations[cfg.Annotations.PausedAt]; !ok {
+		f.t.Errorf("Deployment %s does not have paused-at annotation", name)
+	}
+}
+
 // cleanup removes all created resources.
 func (f *testFixture) cleanup() {
 	for _, w := range f.workloads {
@@ -615,6 +655,77 @@ func TestDeploymentConfigAutoReload(t *testing.T) {
 	f.assertDeploymentConfigReloaded(f.name)
 }
 
+// TestDeploymentPausePeriod tests the pause-period annotation on Deployment.
+// It verifies that after a reload, the deployment is paused and then unpaused after the period expires.
+func TestDeploymentPausePeriod(t *testing.T) {
+	f := newFixture(t, "pause-period")
+	defer f.cleanup()
+
+	pausePeriod := "10s"
+
+	f.createConfigMap(f.name, "initial-data")
+	f.createDeployment(
+		f.name, true, map[string]string{
+			cfg.Annotations.ConfigmapReload: f.name,
+			cfg.Annotations.PausePeriod:     pausePeriod,
+		},
+	)
+	f.waitForReady()
+	f.updateConfigMap(f.name, "updated-data")
+	f.assertDeploymentReloaded(f.name, nil)
+	f.assertDeploymentPaused(f.name)
+	f.assertDeploymentHasPausedAtAnnotation(f.name)
+	t.Log("Waiting for pause period to expire...")
+	f.assertDeploymentUnpaused(f.name, 20*time.Second)
+}
+
+// TestDeploymentPausePeriodWithAutoReload tests pause-period with auto reload annotation.
+func TestDeploymentPausePeriodWithAutoReload(t *testing.T) {
+	f := newFixture(t, "pause-auto")
+	defer f.cleanup()
+
+	pausePeriod := "10s"
+
+	f.createConfigMap(f.name, "initial-data")
+	f.createDeployment(
+		f.name, true, map[string]string{
+			cfg.Annotations.Auto:        "true",
+			cfg.Annotations.PausePeriod: pausePeriod,
+		},
+	)
+	f.waitForReady()
+	f.updateConfigMap(f.name, "updated-data")
+	f.assertDeploymentReloaded(f.name, nil)
+	f.assertDeploymentPaused(f.name)
+	t.Log("Waiting for pause period to expire...")
+	f.assertDeploymentUnpaused(f.name, 20*time.Second)
+}
+
+// TestDeploymentNoPauseWithoutAnnotation tests that deployments without pause-period are not paused.
+func TestDeploymentNoPauseWithoutAnnotation(t *testing.T) {
+	f := newFixture(t, "no-pause")
+	defer f.cleanup()
+
+	f.createConfigMap(f.name, "initial-data")
+	f.createDeployment(
+		f.name, true, map[string]string{
+			cfg.Annotations.ConfigmapReload: f.name,
+		},
+	)
+	f.waitForReady()
+	f.updateConfigMap(f.name, "updated-data")
+	f.assertDeploymentReloaded(f.name, nil)
+
+	time.Sleep(3 * time.Second)
+	deploy, err := k8sClient.AppsV1().Deployments(namespace).Get(context.Background(), f.name, metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("Failed to get deployment: %v", err)
+	}
+	if deploy.Spec.Paused {
+		t.Errorf("Deployment should NOT be paused without pause-period annotation")
+	}
+}
+
 // startManagerWithConfig creates and starts a controller-runtime manager for e2e testing.
 func startManagerWithConfig(cfg *config.Config, restConfig *rest.Config) (manager.Manager, context.CancelFunc) {
 	collectors := metrics.NewCollectors()