New operator spec: upgrade strategy

Closes #36520

Signed-off-by: Pedro Ruivo <pruivo@redhat.com>
Signed-off-by: Alexander Schwartz <aschwart@redhat.com>
Co-authored-by: Alexander Schwartz <aschwart@redhat.com>

docs/guides/operator/advanced-configuration.adoc

@@ -439,4 +439,53 @@ They need to access {project_name} to scrape the available metrics.
 
 Check the https://kubernetes.io/docs/concepts/services-networking/network-policies/[Kubernetes Network Policies documentation] for more information about NetworkPolicies.
 
+=== Managing Keycloak Operator Updates (Preview)
+
+The Keycloak Operator offers updates strategies to control how the Operator handles changes to the Keycloak CR.
+
+**Supported Updates Types:**
+
+Rolling Updates:: Update the StatefulSet in a rolling fashion, minimizing downtime (requires multiple replicas).
+Recreate Updates:: Scale down the StatefulSet before applying updates, causing temporary downtime.
+
+==== Configuring the Update Strategy
+
+The update strategy is specified within the `spec` section of the Keycloak CR YAML definition.
+
+[NOTE]
+====
+During this preview stage, the update strategy defaults to mimicking Keycloak 26.1 or older behavior:
+When the Keycloak CR's image field changes, the Operator scales down the StatefulSet before applying the new image, resulting in downtime.
+Any configuration change will be a rolling update.
+====
+
+[source,yaml]
+----
+apiVersion: k8s.keycloak.org/v2alpha1
+kind: Keycloak
+metadata:
+  name: example-kc
+spec:
+  update:
+    strategy: Recreate|<not set> # <1>
+----
+
+<1> Set the desired update strategy here (Recreate in this example).
+
+[%autowidth]
+.Possible field values
+|===
+|Value |Downtime? |Description
+
+|`<not set>` (default)
+|On image name or tag change
+|Mimics Keycloak 26.1 or older behavior.
+When the image field changes, the StatefulSet is scaled down before applying the new image.
+
+|`Recreate`
+|On any configuration or image change
+|The StatefulSet is scaled down before applying the new configuration or image.
+
+|===
+
 </@tmpl.guide>

operator/src/main/java/org/keycloak/operator/ContextUtils.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2025 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.keycloak.operator;
+
+import java.util.Optional;
+
+import io.fabric8.kubernetes.api.model.apps.StatefulSet;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import org.keycloak.operator.crds.v2alpha1.deployment.Keycloak;
+import org.keycloak.operator.upgrade.UpgradeType;
+
+public final class ContextUtils {
+
+    // context keys
+    private static final String OLD_DEPLOYMENT_KEY = "current_stateful_set";
+    private static final String NEW_DEPLOYMENT_KEY = "desired_new_stateful_set";
+    private static final String UPGRADE_TYPE_KEY = "upgrade_type";
+
+    private ContextUtils() {}
+
+
+    public static void storeCurrentStatefulSet(Context<Keycloak> context, StatefulSet statefulSet) {
+        context.managedDependentResourceContext().put(OLD_DEPLOYMENT_KEY, statefulSet);
+    }
+
+    public static StatefulSet getCurrentStatefulSet(Context<Keycloak> context) {
+        return context.managedDependentResourceContext().getMandatory(OLD_DEPLOYMENT_KEY, StatefulSet.class);
+    }
+
+    public static void storeDesiredStatefulSet(Context<Keycloak> context, StatefulSet statefulSet) {
+        context.managedDependentResourceContext().put(NEW_DEPLOYMENT_KEY, statefulSet);
+    }
+
+    public static StatefulSet getDesiredStatefulSet(Context<Keycloak> context) {
+        return context.managedDependentResourceContext().getMandatory(NEW_DEPLOYMENT_KEY, StatefulSet.class);
+    }
+
+    public static void storeUpgradeType(Context<Keycloak> context, UpgradeType upgradeType) {
+        context.managedDependentResourceContext().put(UPGRADE_TYPE_KEY, upgradeType);
+    }
+
+    public static Optional<UpgradeType> getUpgradeType(Context<Keycloak> context) {
+        return context.managedDependentResourceContext().get(UPGRADE_TYPE_KEY, UpgradeType.class);
+    }
+}

operator/src/main/java/org/keycloak/operator/controllers/KeycloakController.java

@@ -18,6 +18,7 @@ package org.keycloak.operator.controllers;
 
 import io.fabric8.kubernetes.api.model.ContainerState;
 import io.fabric8.kubernetes.api.model.ContainerStateWaiting;
+import io.fabric8.kubernetes.api.model.ContainerStatus;
 import io.fabric8.kubernetes.api.model.PodSpec;
 import io.fabric8.kubernetes.api.model.PodStatus;
 import io.fabric8.kubernetes.api.model.Service;
@@ -50,6 +51,7 @@ import org.keycloak.operator.crds.v2alpha1.deployment.KeycloakStatusAggregator;
 import org.keycloak.operator.crds.v2alpha1.deployment.spec.HostnameSpec;
 import org.keycloak.operator.crds.v2alpha1.deployment.spec.HostnameSpecBuilder;
 
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -57,6 +59,7 @@ import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 
 import jakarta.inject.Inject;
+import org.keycloak.operator.upgrade.UpgradeLogicFactory;
 
 @ControllerConfiguration(
     dependents = {
@@ -79,6 +82,9 @@ public class KeycloakController implements Reconciler<Keycloak>, EventSourceInit
     @Inject
     KeycloakDistConfigurator distConfigurator;
 
+    @Inject
+    UpgradeLogicFactory upgradeLogicFactory;
+
     volatile KeycloakDeploymentDependentResource deploymentDependentResource;
 
     @Override
@@ -134,6 +140,13 @@ public class KeycloakController implements Reconciler<Keycloak>, EventSourceInit
             return UpdateControl.updateResource(kc);
         }
 
+        var upgradeLogicControl = upgradeLogicFactory.create(kc, context, deploymentDependentResource)
+                .decideUpgrade();
+        if (upgradeLogicControl.isPresent()) {
+            Log.debug("--- Reconciliation interrupted due to upgrade logic");
+            return upgradeLogicControl.get();
+        }
+
         // after the spec has possibly been updated, reconcile the StatefulSet
         this.deploymentDependentResource.reconcile(kc, context);
 
@@ -263,11 +276,12 @@ public class KeycloakController implements Reconciler<Keycloak>, EventSourceInit
                 .list().getItems().stream()
                 .filter(p -> !Readiness.isPodReady(p)
                         && Optional.ofNullable(p.getStatus()).map(PodStatus::getContainerStatuses).isPresent())
-                .sorted((p1, p2) -> p1.getMetadata().getName().compareTo(p2.getMetadata().getName()))
+                .sorted(Comparator.comparing(p -> p.getMetadata().getName()))
                 .forEachOrdered(p -> {
-                    Optional.of(p.getStatus()).map(s -> s.getContainerStatuses()).stream().flatMap(List::stream)
+                    Optional.of(p.getStatus()).map(PodStatus::getContainerStatuses).stream().flatMap(List::stream)
                             .filter(cs -> !Boolean.TRUE.equals(cs.getReady()))
-                            .sorted((cs1, cs2) -> cs1.getName().compareTo(cs2.getName())).forEachOrdered(cs -> {
+                            .sorted(Comparator.comparing(ContainerStatus::getName))
+                            .forEachOrdered(cs -> {
                                 if (Optional.ofNullable(cs.getState()).map(ContainerState::getWaiting)
                                         .map(ContainerStateWaiting::getReason).map(String::toLowerCase)
                                         .filter(s -> s.contains("err") || s.equals("crashloopbackoff")).isPresent()) {

operator/src/main/java/org/keycloak/operator/controllers/KeycloakDeploymentDependentResource.java

@@ -22,7 +22,6 @@ import io.fabric8.kubernetes.api.model.EnvVar;
 import io.fabric8.kubernetes.api.model.EnvVarBuilder;
 import io.fabric8.kubernetes.api.model.EnvVarSource;
 import io.fabric8.kubernetes.api.model.EnvVarSourceBuilder;
-import io.fabric8.kubernetes.api.model.PodSpec;
 import io.fabric8.kubernetes.api.model.PodSpecFluent;
 import io.fabric8.kubernetes.api.model.PodTemplateSpec;
 import io.fabric8.kubernetes.api.model.Secret;
@@ -31,7 +30,6 @@ import io.fabric8.kubernetes.api.model.VolumeBuilder;
 import io.fabric8.kubernetes.api.model.VolumeMountBuilder;
 import io.fabric8.kubernetes.api.model.apps.StatefulSet;
 import io.fabric8.kubernetes.api.model.apps.StatefulSetBuilder;
-import io.fabric8.kubernetes.api.model.apps.StatefulSetSpec;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.CRUDKubernetesDependentResource;
@@ -40,6 +38,7 @@ import io.quarkus.logging.Log;
 
 import org.keycloak.operator.Config;
 import org.keycloak.operator.Constants;
+import org.keycloak.operator.ContextUtils;
 import org.keycloak.operator.Utils;
 import org.keycloak.operator.crds.v2alpha1.deployment.Keycloak;
 import org.keycloak.operator.crds.v2alpha1.deployment.KeycloakSpec;
@@ -131,32 +130,33 @@ public class KeycloakDeploymentDependentResource extends CRUDKubernetesDependent
         addEnvVars(baseDeployment, primary, allSecrets);
         addResources(primary.getSpec().getResourceRequirements(), operatorConfig, kcContainer);
         Optional.ofNullable(primary.getSpec().getCacheSpec())
-                .ifPresent(c -> configureCache(primary, baseDeployment, kcContainer, c, context.getClient()));
+                .ifPresent(c -> configureCache(baseDeployment, kcContainer, c, context.getClient()));
 
         if (!allSecrets.isEmpty()) {
             watchedResources.annotateDeployment(new ArrayList<>(allSecrets), Secret.class, baseDeployment, context.getClient());
         }
 
-        StatefulSet existingDeployment = context.getSecondaryResource(StatefulSet.class).orElse(null);
-        if (existingDeployment == null) {
-            Log.debug("No existing Deployment found, using the default");
-        }
-        else {
-            Log.debug("Existing Deployment found, handling migration");
-
-            // version 22 changed the match labels, account for older versions
-            if (!existingDeployment.isMarkedForDeletion() && !hasExpectedMatchLabels(existingDeployment, primary)) {
-                context.getClient().resource(existingDeployment).lockResourceVersion().delete();
-                Log.info("Existing Deployment found with old label selector, it will be recreated");
-            }
-
-            migrateDeployment(existingDeployment, baseDeployment, context);
+        var upgradeType = ContextUtils.getUpgradeType(context);
+        // empty means no existing stateful set.
+        if (upgradeType.isEmpty()) {
+            return baseDeployment;
         }
 
-        return baseDeployment;
+        var existingDeployment = ContextUtils.getCurrentStatefulSet(context);
+
+        // version 22 changed the match labels, account for older versions
+        if (!existingDeployment.isMarkedForDeletion() && !hasExpectedMatchLabels(existingDeployment, primary)) {
+            context.getClient().resource(existingDeployment).lockResourceVersion().delete();
+            Log.info("Existing Deployment found with old label selector, it will be recreated");
+        }
+
+        return switch (upgradeType.get()) {
+            case ROLLING -> handleRollingUpdate(baseDeployment);
+            case RECREATE -> handleRecreateUpdate(existingDeployment, baseDeployment);
+        };
     }
 
-    private void configureCache(Keycloak keycloakCR, StatefulSet deployment, Container kcContainer, CacheSpec spec, KubernetesClient client) {
+    private void configureCache(StatefulSet deployment, Container kcContainer, CacheSpec spec, KubernetesClient client) {
         Optional.ofNullable(spec.getConfigMapFile()).ifPresent(configFile -> {
             if (configFile.getName() == null || configFile.getKey() == null) {
                 throw new IllegalStateException("Cache file ConfigMap requires both a name and a key");
@@ -345,7 +345,7 @@ public class KeycloakDeploymentDependentResource extends CRUDKubernetesDependent
         }
 
         // add in ports - there's no merging being done here
-        final StatefulSet baseDeployment = containerBuilder
+        return containerBuilder
             .addNewPort()
                 .withName(Constants.KEYCLOAK_HTTPS_PORT_NAME)
                 .withContainerPort(Constants.KEYCLOAK_HTTPS_PORT)
@@ -362,8 +362,6 @@ public class KeycloakDeploymentDependentResource extends CRUDKubernetesDependent
                 .withProtocol(Constants.KEYCLOAK_SERVICE_PROTOCOL)
             .endPort()
             .endContainer().endSpec().endTemplate().endSpec().build();
-
-        return baseDeployment;
     }
 
     private void handleScheduling(Keycloak keycloakCR, Map<String, String> labels, PodSpecFluent<?> specBuilder) {
@@ -477,7 +475,7 @@ public class KeycloakDeploymentDependentResource extends CRUDKubernetesDependent
 
         // merge with the CR; the values in CR take precedence
         if (keycloakCR.getSpec().getAdditionalOptions() != null) {
-            Set<String> inCr = keycloakCR.getSpec().getAdditionalOptions().stream().map(v -> v.getName()).collect(Collectors.toSet());
+            Set<String> inCr = keycloakCR.getSpec().getAdditionalOptions().stream().map(ValueOrSecret::getName).collect(Collectors.toSet());
             serverConfigsList.removeIf(v -> inCr.contains(v.getName()));
             serverConfigsList.addAll(keycloakCR.getSpec().getAdditionalOptions());
         }
@@ -514,28 +512,6 @@ public class KeycloakDeploymentDependentResource extends CRUDKubernetesDependent
         return keycloak.getMetadata().getName();
     }
 
-    public void migrateDeployment(StatefulSet previousDeployment, StatefulSet reconciledDeployment, Context<Keycloak> context) {
-        var previousContainer = Optional.ofNullable(previousDeployment).map(StatefulSet::getSpec)
-                .map(StatefulSetSpec::getTemplate).map(PodTemplateSpec::getSpec).map(PodSpec::getContainers)
-                .flatMap(c -> c.stream().findFirst()).orElse(null);
-        if (previousContainer == null) {
-            return;
-        }
-        var reconciledContainer = reconciledDeployment.getSpec().getTemplate().getSpec().getContainers().get(0);
-
-        if (!previousContainer.getImage().equals(reconciledContainer.getImage())
-                && previousDeployment.getStatus().getReplicas() > 0) {
-            // TODO Check if migration is really needed (e.g. based on actual KC version); https://github.com/keycloak/keycloak/issues/10441
-            Log.info("Detected changed Keycloak image, assuming Keycloak upgrade. Scaling down the deployment to one instance to perform a safe database migration");
-            Log.infof("original image: %s; new image: %s", previousContainer.getImage(), reconciledContainer.getImage());
-
-            reconciledContainer.setImage(previousContainer.getImage());
-            reconciledDeployment.getSpec().setReplicas(0);
-
-            reconciledDeployment.getMetadata().getAnnotations().put(Constants.KEYCLOAK_MIGRATING_ANNOTATION, Boolean.TRUE.toString());
-        }
-    }
-
     protected Optional<String> readConfigurationValue(String key, Keycloak keycloakCR, Context<Keycloak> context) {
         return Optional.ofNullable(keycloakCR.getSpec()).map(KeycloakSpec::getAdditionalOptions)
                 .flatMap(l -> l.stream().filter(sc -> sc.getName().equals(key)).findFirst().map(serverConfigValue -> {
@@ -557,4 +533,29 @@ public class KeycloakDeploymentDependentResource extends CRUDKubernetesDependent
         }));
     }
 
+    private static StatefulSet handleRollingUpdate(StatefulSet desired) {
+        // return the desired stateful set since Kubernetes does a rolling in-place upgrade by default.
+        Log.debug("Performing a rolling upgrade");
+        return desired;
+    }
+
+    private static StatefulSet  handleRecreateUpdate(StatefulSet actual, StatefulSet desired) {
+        if (actual.getStatus().getReplicas() == 0) {
+            Log.debug("Performing a recreate upgrade - scaling up the stateful set");
+            return desired;
+        }
+        Log.debug("Performing a recreate upgrade - scaling down the stateful set");
+        // return the existing stateful set, but set replicas to zero
+        var builder = actual.toBuilder();
+        builder.editSpec()
+                .withReplicas(0)
+                .endSpec();
+        // update metadata from the new stateful set, it is safe to do so.
+        builder.withMetadata(desired.getMetadata());
+        builder.editMetadata()
+                .addToAnnotations(Constants.KEYCLOAK_MIGRATING_ANNOTATION, Boolean.TRUE.toString())
+                .endMetadata();
+        return builder.build();
+    }
+
 }

operator/src/main/java/org/keycloak/operator/crds/v2alpha1/CRDUtils.java

@@ -19,11 +19,20 @@ package org.keycloak.operator.crds.v2alpha1;
 
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.function.Predicate;
 import java.util.stream.Stream;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.PodSpec;
+import io.fabric8.kubernetes.api.model.PodTemplateSpec;
+import io.fabric8.kubernetes.api.model.apps.StatefulSet;
+import io.fabric8.kubernetes.api.model.apps.StatefulSetSpec;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
 import org.keycloak.operator.Constants;
 import org.keycloak.operator.crds.v2alpha1.deployment.Keycloak;
 import org.keycloak.operator.crds.v2alpha1.deployment.KeycloakSpec;
@@ -81,4 +90,18 @@ public final class CRDUtils {
                 .map(Keycloak::getSpec);
     }
 
+    public static Optional<Container> firstContainerOf(StatefulSet statefulSet) {
+        return Optional.ofNullable(statefulSet)
+                .map(StatefulSet::getSpec)
+                .map(StatefulSetSpec::getTemplate)
+                .map(PodTemplateSpec::getSpec)
+                .map(PodSpec::getContainers)
+                .filter(Predicate.not(List::isEmpty))
+                .map(containers -> containers.get(0));
+    }
+
+    public static <T> JsonNode toJsonNode(T value, Context<Keycloak> context) {
+        final var kubernetesSerialization = context.getClient().getKubernetesSerialization();
+        return kubernetesSerialization.convertValue(value, JsonNode.class);
+    }
 }

operator/src/main/java/org/keycloak/operator/crds/v2alpha1/deployment/KeycloakSpec.java

@@ -44,6 +44,7 @@ import java.util.Map;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonPropertyDescription;
+import org.keycloak.operator.crds.v2alpha1.deployment.spec.UpdateSpec;
 
 @JsonInclude(JsonInclude.Include.NON_NULL)
 public class KeycloakSpec {
@@ -130,6 +131,10 @@ public class KeycloakSpec {
     @JsonPropertyDescription("In this section you can configure OpenTelemetry Tracing for Keycloak.")
     private TracingSpec tracingSpec;
 
+    @JsonProperty("update")
+    @JsonPropertyDescription("Configuration related to Keycloak deployment upgrades.")
+    private UpdateSpec updateSpec;
+
     public HttpSpec getHttpSpec() {
         return httpSpec;
     }
@@ -303,4 +308,12 @@ public class KeycloakSpec {
     public void setTracingSpec(TracingSpec tracingSpec) {
         this.tracingSpec = tracingSpec;
     }
+
+    public UpdateSpec getUpdateSpec() {
+        return updateSpec;
+    }
+
+    public void setUpdateSpec(UpdateSpec updateSpec) {
+        this.updateSpec = updateSpec;
+    }
 }

operator/src/main/java/org/keycloak/operator/crds/v2alpha1/deployment/spec/UnsupportedSpec.java

@@ -31,9 +31,10 @@ import io.sundr.builder.annotations.BuildableReference;
 })
 public class UnsupportedSpec {
 
-    @JsonPropertyDescription("You can configure that will be merged with the one configured by default by the operator.\n" +
-            "Use at your own risk, we reserve the possibility to remove/change the way any field gets merged in future releases without notice.\n" +
-            "Reference: https://kubernetes.io/docs/concepts/workloads/pods/#pod-templates")
+    @JsonPropertyDescription("""
+            You can configure that will be merged with the one configured by default by the operator.
+            Use at your own risk, we reserve the possibility to remove/change the way any field gets merged in future releases without notice.
+            Reference: https://kubernetes.io/docs/concepts/workloads/pods/#pod-templates""")
     private PodTemplateSpec podTemplate;
 
     public UnsupportedSpec() {}
@@ -46,7 +47,7 @@ public class UnsupportedSpec {
         return podTemplate;
     }
 
-    public void setPodTeplate(PodTemplateSpec podTemplate) {
+    public void setPodTemplate(PodTemplateSpec podTemplate) {
         this.podTemplate = podTemplate;
     }
 

operator/src/main/java/org/keycloak/operator/crds/v2alpha1/deployment/spec/UpdateSpec.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2025 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.keycloak.operator.crds.v2alpha1.deployment.spec;
+
+import java.util.Optional;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonPropertyDescription;
+import io.sundr.builder.annotations.Buildable;
+import org.keycloak.operator.crds.v2alpha1.CRDUtils;
+import org.keycloak.operator.crds.v2alpha1.deployment.Keycloak;
+import org.keycloak.operator.crds.v2alpha1.deployment.KeycloakSpec;
+import org.keycloak.operator.upgrade.UpdateStrategy;
+
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@Buildable(editableEnabled = false, builderPackage = "io.fabric8.kubernetes.api.builder")
+public class UpdateSpec {
+
+    @JsonPropertyDescription("Sets the upgrade strategy to use.")
+    private UpdateStrategy strategy;
+
+    public UpdateStrategy getStrategy() {
+        return strategy;
+    }
+
+    public void setStrategy(UpdateStrategy strategy) {
+        this.strategy = strategy;
+    }
+
+    public static Optional<UpdateStrategy> findUpdateStrategy(Keycloak keycloak) {
+        return CRDUtils.keycloakSpecOf(keycloak)
+                .map(KeycloakSpec::getUpdateSpec)
+                .map(UpdateSpec::getStrategy);
+    }
+}

operator/src/main/java/org/keycloak/operator/upgrade/UpdateStrategy.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2025 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.keycloak.operator.upgrade;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyDescription;
+
+public enum UpdateStrategy {
+    @JsonPropertyDescription("Shutdown the Keycloak cluster before applying the new changes.")
+    @JsonProperty("Recreate")
+    RECREATE
+}

operator/src/main/java/org/keycloak/operator/upgrade/UpgradeLogic.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2025 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.keycloak.operator.upgrade;
+
+import java.util.Optional;
+
+import io.fabric8.kubernetes.api.model.apps.StatefulSet;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import org.keycloak.operator.crds.v2alpha1.deployment.Keycloak;
+
+/**
+ * An API to implement to handle Keycloak CR updates.
+ * <p>
+ * This interface is invoked all the time before creating the {@link StatefulSet} and it can manipulate the
+ * reconciliation to perform or check other tasks required before the {@link StatefulSet} is created or updated.
+ */
+public interface UpgradeLogic {
+
+    /**
+     * It must check is an existing {@link StatefulSet} exists and decided on the {@link UpgradeType} to update the
+     * {@link StatefulSet}.
+     * <p>
+     * The method should use {@link org.keycloak.operator.ContextUtils#storeUpgradeType(Context, UpgradeType)} to store
+     * its decision. If no prior {@link StatefulSet} is present, no decision is required and
+     * {@link org.keycloak.operator.ContextUtils#storeUpgradeType(Context, UpgradeType)} must not be invoked.
+     * <p>
+     * Return a non-empty {@link Optional} to interrupt the reconciliation until the next event. The interrupted
+     * prevents the {@link StatefulSet} from being updated.
+     *
+     * @return The {@link UpdateControl} if the reconciliation needs to be interrupted or an empty {@link Optional} if
+     * it can proceed.
+     */
+    Optional<UpdateControl<Keycloak>> decideUpgrade();
+
+}

operator/src/main/java/org/keycloak/operator/upgrade/UpgradeLogicFactory.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2025 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.keycloak.operator.upgrade;
+
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import jakarta.enterprise.context.ApplicationScoped;
+import org.keycloak.operator.controllers.KeycloakDeploymentDependentResource;
+import org.keycloak.operator.crds.v2alpha1.deployment.Keycloak;
+import org.keycloak.operator.crds.v2alpha1.deployment.spec.UpdateSpec;
+import org.keycloak.operator.upgrade.impl.AlwaysRecreateUpgradeLogic;
+import org.keycloak.operator.upgrade.impl.RecreateOnImageChangeUpgradeLogic;
+
+/**
+ * The {@link UpgradeLogic} factory. It returns an implementation based on the {@link Keycloak} configuration.
+ */
+@ApplicationScoped
+public class UpgradeLogicFactory {
+
+    @SuppressWarnings("removal")
+    public UpgradeLogic create(Keycloak keycloak, Context<Keycloak> context, KeycloakDeploymentDependentResource dependentResource) {
+        var strategy = UpdateSpec.findUpdateStrategy(keycloak);
+        if (strategy.isEmpty()) {
+            return new RecreateOnImageChangeUpgradeLogic(context, keycloak, dependentResource);
+        }
+        return switch (strategy.get()) {
+            case RECREATE -> new AlwaysRecreateUpgradeLogic(context, keycloak, dependentResource);
+        };
+    }
+
+}

operator/src/main/java/org/keycloak/operator/upgrade/UpgradeType.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2025 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.keycloak.operator.upgrade;
+
+import io.fabric8.kubernetes.api.model.apps.StatefulSet;
+import org.keycloak.operator.controllers.KeycloakDeploymentDependentResource;
+
+/**
+ * Supported upgrade type by {@link KeycloakDeploymentDependentResource}.
+ */
+public enum UpgradeType {
+    /**
+     * Shutdown the existing cluster before updating the {@link StatefulSet}.
+     */
+    RECREATE,
+    /**
+     * Updates the {@link StatefulSet} and does a rolling upgrade.
+     */
+    ROLLING
+}

operator/src/main/java/org/keycloak/operator/upgrade/impl/AlwaysRecreateUpgradeLogic.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2025 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.keycloak.operator.upgrade.impl;
+
+import java.util.Optional;
+
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import org.keycloak.operator.controllers.KeycloakDeploymentDependentResource;
+import org.keycloak.operator.crds.v2alpha1.deployment.Keycloak;
+import org.keycloak.operator.upgrade.UpgradeLogic;
+import org.keycloak.operator.upgrade.UpgradeType;
+
+/**
+ * An {@link UpgradeLogic} implementation that forces a {@link UpgradeType#RECREATE} on every configuration or image
+ * change.
+ */
+public class AlwaysRecreateUpgradeLogic extends BaseUpgradeLogic {
+
+    public AlwaysRecreateUpgradeLogic(Context<Keycloak> context, Keycloak keycloak, KeycloakDeploymentDependentResource statefulSetResource) {
+        super(context, keycloak, statefulSetResource);
+    }
+
+    @Override
+    Optional<UpdateControl<Keycloak>> onUpgrade() {
+        decideRecreateUpgrade();
+        return Optional.empty();
+    }
+}

operator/src/main/java/org/keycloak/operator/upgrade/impl/BaseUpgradeLogic.java

@@ -0,0 +1,159 @@
+/*
+ * Copyright 2025 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.keycloak.operator.upgrade.impl;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.EnvVar;
+import io.fabric8.kubernetes.api.model.apps.StatefulSet;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.quarkus.logging.Log;
+import org.keycloak.operator.ContextUtils;
+import org.keycloak.operator.controllers.KeycloakDeploymentDependentResource;
+import org.keycloak.operator.crds.v2alpha1.CRDUtils;
+import org.keycloak.operator.crds.v2alpha1.deployment.Keycloak;
+import org.keycloak.operator.upgrade.UpgradeLogic;
+import org.keycloak.operator.upgrade.UpgradeType;
+
+/**
+ * Common {@link UpgradeLogic} implementation that checks if the upgrade logic needs to be run.
+ * <p>
+ * The upgrade logic can be skipped if it is the first deployment or if the change is not relevance (like, updating the
+ * number of replicas or annotations).
+ */
+abstract class BaseUpgradeLogic implements UpgradeLogic {
+
+    protected final Context<Keycloak> context;
+    protected final Keycloak keycloak;
+    protected final KeycloakDeploymentDependentResource statefulSetResource;
+
+    BaseUpgradeLogic(Context<Keycloak> context, Keycloak keycloak, KeycloakDeploymentDependentResource statefulSetResource) {
+        this.context = context;
+        this.keycloak = keycloak;
+        this.statefulSetResource = statefulSetResource;
+    }
+
+    @Override
+    public final Optional<UpdateControl<Keycloak>> decideUpgrade() {
+        var existing = context.getSecondaryResource(StatefulSet.class);
+        if (existing.isEmpty()) {
+            // new deployment, no upgrade needed
+            Log.debug("New deployment - skipping upgrade logic");
+            return Optional.empty();
+        }
+        var desiredStatefulSet = statefulSetResource.desired(keycloak, context);
+        var desiredContainer = CRDUtils.firstContainerOf(desiredStatefulSet).orElseThrow(BaseUpgradeLogic::containerNotFound);
+        var actualContainer = CRDUtils.firstContainerOf(existing.get()).orElseThrow(BaseUpgradeLogic::containerNotFound);
+
+        if (isContainerEquals(actualContainer, desiredContainer)) {
+            // container is equals, no upgrade required
+            Log.debug("No changes detected in the container - skipping upgrade logic");
+            return Optional.empty();
+        }
+
+        // store in context the current and desired stateful set for easy access.
+        ContextUtils.storeCurrentStatefulSet(context, existing.get());
+        ContextUtils.storeDesiredStatefulSet(context, desiredStatefulSet);
+        return onUpgrade();
+    }
+
+    /**
+     * Concrete upgrade logic should be implemented here.
+     * <p>
+     * Use {@link ContextUtils#getCurrentStatefulSet(Context)} and/or
+     * {@link ContextUtils#getDesiredStatefulSet(Context)} to get the current and the desired {@link StatefulSet},
+     * respectively.
+     * <p>
+     * Use the methods {@link #decideRecreateUpgrade()} or {@link #decideRollingUpgrade()} to use one of the available
+     * upgrade logics.
+     *
+     * @return An {@link UpdateControl} if the reconciliation must be interrupted before updating the
+     * {@link StatefulSet}.
+     */
+    abstract Optional<UpdateControl<Keycloak>> onUpgrade();
+
+    void decideRollingUpgrade() {
+        Log.debug("Decided rolling upgrade type.");
+        ContextUtils.storeUpgradeType(context, UpgradeType.ROLLING);
+    }
+
+    void decideRecreateUpgrade() {
+        Log.debug("Decided recreate upgrade type.");
+        ContextUtils.storeUpgradeType(context, UpgradeType.RECREATE);
+    }
+
+    static IllegalStateException containerNotFound() {
+        return new IllegalStateException("Container not found in stateful set.");
+    }
+
+    private static boolean isContainerEquals(Container actual, Container desired) {
+        return isImageEquals(actual, desired) &&
+                isArgsEquals(actual, desired) &&
+                isEnvEquals(actual, desired);
+    }
+
+    private static boolean isImageEquals(Container actual, Container desired) {
+        return isEquals("image", actual.getImage(), desired.getImage());
+    }
+
+    private static boolean isArgsEquals(Container actual, Container desired) {
+        var actualArgs = actual.getArgs().stream().sorted().toList();
+        var desiredArgs = desired.getArgs().stream().sorted().toList();
+        return isEquals("args", actualArgs, desiredArgs);
+    }
+
+    private static boolean isEnvEquals(Container actual, Container desired) {
+        var actualEnv = envVars(actual);
+        var desiredEnv = envVars(desired);
+        return isEquals("env", actualEnv, desiredEnv);
+    }
+
+    private static Map<String, String> envVars(Container container) {
+        // The operator only sets value or secrets. Any other combination is from unsupported pod template.
+        //noinspection DataFlowIssue
+        return container.getEnv().stream()
+                .filter(envVar -> !envVar.getName().equals(KeycloakDeploymentDependentResource.POD_IP))
+                .filter(envVar -> Objects.nonNull(valueOrSecret(envVar)))
+                .collect(Collectors.toMap(EnvVar::getName, BaseUpgradeLogic::valueOrSecret));
+    }
+
+    private static String valueOrSecret(EnvVar envVar) {
+        var value = envVar.getValue();
+        if (value != null) {
+            return value;
+        }
+        var secret = envVar.getValueFrom().getSecretKeyRef();
+        if (secret != null) {
+            return secret.getName();
+        }
+        return null;
+    }
+
+    private static <T> boolean isEquals(String key, T actual, T desired) {
+        var isEquals = Objects.equals(actual, desired);
+        if (!isEquals) {
+            Log.debugf("Found difference in container's %s:%nactual:%s%ndesired:%s", key, actual, desired);
+        }
+        return isEquals;
+    }
+}

operator/src/main/java/org/keycloak/operator/upgrade/impl/RecreateOnImageChangeUpgradeLogic.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2025 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.keycloak.operator.upgrade.impl;
+
+import java.util.Objects;
+import java.util.Optional;
+
+import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.apps.StatefulSet;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import org.keycloak.operator.ContextUtils;
+import org.keycloak.operator.controllers.KeycloakDeploymentDependentResource;
+import org.keycloak.operator.crds.v2alpha1.CRDUtils;
+import org.keycloak.operator.crds.v2alpha1.deployment.Keycloak;
+import org.keycloak.operator.upgrade.UpgradeType;
+
+/**
+ * Implements Keycloak 26.0 logic.
+ * <p>
+ * It uses a {@link UpgradeType#RECREATE} if the image changes; otherwise uses {@link UpgradeType#ROLLING}.
+ *
+ * @deprecated To be removed when the new zero-downtime feature is completed.
+ */
+@SuppressWarnings("ALL")
+@Deprecated(forRemoval = true)
+public class RecreateOnImageChangeUpgradeLogic extends BaseUpgradeLogic {
+
+    public RecreateOnImageChangeUpgradeLogic(Context<Keycloak> context, Keycloak keycloak, KeycloakDeploymentDependentResource dependentResource) {
+        super(context, keycloak, dependentResource);
+    }
+
+    @Override
+    Optional<UpdateControl<Keycloak>> onUpgrade() {
+        var currentImage = extractImage(ContextUtils.getCurrentStatefulSet(context));
+        var desiredImage = extractImage(ContextUtils.getDesiredStatefulSet(context));
+
+        if (Objects.equals(currentImage, desiredImage)) {
+            decideRollingUpgrade();
+        } else {
+            decideRecreateUpgrade();
+        }
+        return Optional.empty();
+    }
+
+    private static String extractImage(StatefulSet statefulSet) {
+        return CRDUtils.firstContainerOf(statefulSet)
+                .map(Container::getImage)
+                .orElseThrow(BaseUpgradeLogic::containerNotFound);
+    }
+
+}

operator/src/test/java/org/keycloak/operator/testsuite/integration/PodTemplateTest.java

@@ -92,7 +92,7 @@ public class PodTemplateTest extends BaseOperatorTest {
                 .withName("foo")
                 .endMetadata()
                 .build();
-        plainKc.getSpec().getUnsupported().setPodTeplate(podTemplate);
+        plainKc.getSpec().getUnsupported().setPodTemplate(podTemplate);
 
         // Act
         K8sUtils.set(k8sclient, plainKc);
@@ -120,7 +120,7 @@ public class PodTemplateTest extends BaseOperatorTest {
                     .withNamespace(wrongNamespace)
                     .endMetadata()
                     .build();
-            plainKc.getSpec().getUnsupported().setPodTeplate(podTemplate);
+            plainKc.getSpec().getUnsupported().setPodTemplate(podTemplate);
 
             // Act
             K8sUtils.set(k8sclient, plainKc);
@@ -151,7 +151,7 @@ public class PodTemplateTest extends BaseOperatorTest {
                 .endContainer()
                 .endSpec()
                 .build();
-        plainKc.getSpec().getUnsupported().setPodTeplate(podTemplate);
+        plainKc.getSpec().getUnsupported().setPodTemplate(podTemplate);
 
         // Act
         K8sUtils.set(k8sclient, plainKc);
@@ -177,7 +177,7 @@ public class PodTemplateTest extends BaseOperatorTest {
                 .endContainer()
                 .endSpec()
                 .build();
-        plainKc.getSpec().getUnsupported().setPodTeplate(podTemplate);
+        plainKc.getSpec().getUnsupported().setPodTemplate(podTemplate);
 
         // Act
         K8sUtils.set(k8sclient, plainKc);
@@ -211,7 +211,7 @@ public class PodTemplateTest extends BaseOperatorTest {
                 .build();
 
         var plainKc = getEmptyPodTemplateKeycloak();
-        plainKc.getSpec().getUnsupported().setPodTeplate(podTemplate);
+        plainKc.getSpec().getUnsupported().setPodTemplate(podTemplate);
 
         // Act
         K8sUtils.set(k8sclient, plainKc);

operator/src/test/java/org/keycloak/operator/testsuite/integration/UpgradeTest.java

@@ -0,0 +1,123 @@
+/*
+ * Copyright 2025 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.keycloak.operator.testsuite.integration;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Stream;
+
+import io.quarkus.test.junit.QuarkusTest;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.keycloak.operator.crds.v2alpha1.deployment.Keycloak;
+import org.keycloak.operator.crds.v2alpha1.deployment.KeycloakStatusCondition;
+import org.keycloak.operator.crds.v2alpha1.deployment.ValueOrSecret;
+import org.keycloak.operator.crds.v2alpha1.deployment.spec.UnsupportedSpec;
+import org.keycloak.operator.crds.v2alpha1.deployment.spec.UpdateSpec;
+import org.keycloak.operator.upgrade.UpdateStrategy;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.keycloak.operator.testsuite.utils.CRAssert.assertKeycloakStatusCondition;
+import static org.keycloak.operator.testsuite.utils.CRAssert.eventuallyRecreateUpgradeStatus;
+import static org.keycloak.operator.testsuite.utils.CRAssert.eventuallyRollingUpgradeStatus;
+import static org.keycloak.operator.testsuite.utils.K8sUtils.deployKeycloak;
+
+@QuarkusTest
+public class UpgradeTest extends BaseOperatorTest {
+
+    private static Stream<UpdateStrategy> upgradeStrategy() {
+        return Stream.of(
+                null,
+                UpdateStrategy.RECREATE
+        );
+    }
+
+    @ParameterizedTest(name = "testImageChange-{0}")
+    @MethodSource("upgradeStrategy")
+    public void testImageChange(UpdateStrategy updateStrategy) throws InterruptedException {
+        var kc = createInitialDeployment(updateStrategy);
+        deployKeycloak(k8sclient, kc, true);
+
+        var stsGetter = k8sclient.apps().statefulSets().inNamespace(namespace).withName(kc.getMetadata().getName());
+        final String newImage = "quay.io/keycloak/non-existing-keycloak";
+
+        // changing the image to non-existing will always use the recreate upgrade type.
+        kc.getSpec().setImage(newImage);
+        var upgradeCondition = eventuallyRecreateUpgradeStatus(k8sclient, kc);
+
+        deployKeycloak(k8sclient, kc, false);
+        await(upgradeCondition);
+
+        Awaitility.await()
+                .ignoreExceptions()
+                .untilAsserted(() -> {
+                    var sts = stsGetter.get();
+                    assertEquals(kc.getSpec().getInstances(), sts.getSpec().getReplicas()); // just checking specs as we're using a non-existing image
+                    assertEquals(newImage, sts.getSpec().getTemplate().getSpec().getContainers().get(0).getImage());
+
+                    var currentKc = k8sclient.resources(Keycloak.class)
+                            .inNamespace(namespace).withName(kc.getMetadata().getName()).get();
+                    assertKeycloakStatusCondition(currentKc, KeycloakStatusCondition.READY, false, "Waiting for more replicas");
+                });
+    }
+
+    @ParameterizedTest(name = "testCacheMaxCount-{0}")
+    @MethodSource("upgradeStrategy")
+    public void testCacheMaxCount(UpdateStrategy updateStrategy) throws InterruptedException {
+        var kc = createInitialDeployment(updateStrategy);
+        deployKeycloak(k8sclient, kc, true);
+
+        // changing the local cache max-count should never use the recreate upgrade type
+        // except if forced by the Keycloak CR.
+        kc.getSpec().getAdditionalOptions().add(new ValueOrSecret("cache-embedded-authorization-max-count", "10"));
+        var upgradeCondition = updateStrategy == UpdateStrategy.RECREATE ?
+                eventuallyRecreateUpgradeStatus(k8sclient, kc) :
+                eventuallyRollingUpgradeStatus(k8sclient, kc);
+
+        deployKeycloak(k8sclient, kc, true);
+
+        await(upgradeCondition);
+    }
+
+    private static Keycloak createInitialDeployment(UpdateStrategy updateStrategy) {
+        var kc = getTestKeycloakDeployment(true);
+        kc.getSpec().setInstances(3);
+        if (updateStrategy == null) {
+            return kc;
+        }
+        var updateSpec = new UpdateSpec();
+        updateSpec.setStrategy(updateStrategy);
+
+        if (kc.getSpec().getUnsupported() == null) {
+            kc.getSpec().setUnsupported(new UnsupportedSpec());
+        }
+        kc.getSpec().setUpdateSpec(updateSpec);
+        return kc;
+    }
+
+    private static <T> void await(CompletableFuture<T> future) throws InterruptedException {
+        try {
+            future.get(2, TimeUnit.MINUTES);
+        } catch (ExecutionException | TimeoutException e) {
+            throw new AssertionError(e);
+        }
+    }
+}

operator/src/test/java/org/keycloak/operator/testsuite/unit/CRSerializationTest.java

@@ -37,6 +37,7 @@ import org.keycloak.operator.crds.v2alpha1.deployment.spec.TracingSpec;
 import org.keycloak.operator.crds.v2alpha1.deployment.spec.TransactionsSpec;
 import org.keycloak.operator.crds.v2alpha1.realmimport.KeycloakRealmImport;
 import org.keycloak.operator.testsuite.utils.K8sUtils;
+import org.keycloak.operator.upgrade.UpdateStrategy;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.emptyString;
@@ -49,6 +50,7 @@ import static org.hamcrest.Matchers.nullValue;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
@@ -253,6 +255,23 @@ public class CRSerializationTest {
         assertNetworkPolicyRules(networkPolicySpec.getManagementRules());
     }
 
+    @Test
+    public void testUpgradeStrategy() {
+        var keycloak = Serialization.unmarshal(this.getClass().getResourceAsStream("/test-serialization-keycloak-cr.yml"), Keycloak.class);
+        var updateSpec = keycloak.getSpec().getUpdateSpec();
+        assertNotNull(updateSpec);
+        var upgradeStrategy = updateSpec.getStrategy();
+        assertNotNull(upgradeStrategy);
+        assertEquals(UpdateStrategy.RECREATE, upgradeStrategy);
+    }
+
+    @Test
+    public void testInvalidUpgradeStrategy() {
+        var thrown = assertThrows(IllegalArgumentException.class,
+                () -> Serialization.unmarshal(this.getClass().getResourceAsStream("/test-serialization-keycloak-cr-invalid-update.yml"), Keycloak.class));
+        assertTrue(thrown.getMessage().contains("Cannot deserialize value of type `org.keycloak.operator.upgrade.UpdateStrategy` from String \"abc\""));
+    }
+
     private static void assertNetworkPolicyRules(Collection<NetworkPolicyPeer> rules) {
         assertNotNull(rules);
         assertEquals(3, rules.size());

operator/src/test/java/org/keycloak/operator/testsuite/unit/PodTemplateTest.java

@@ -34,6 +34,7 @@ import io.fabric8.kubernetes.api.model.apps.StatefulSet;
 import io.fabric8.kubernetes.api.model.apps.StatefulSetBuilder;
 import io.fabric8.kubernetes.client.utils.Serialization;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DefaultManagedDependentResourceContext;
 import io.quarkus.test.InjectMock;
 import io.quarkus.test.junit.QuarkusTest;
 
@@ -55,7 +56,7 @@ import org.keycloak.operator.crds.v2alpha1.deployment.spec.HttpSpecBuilder;
 import org.keycloak.operator.crds.v2alpha1.deployment.spec.UnsupportedSpec;
 import org.mockito.Mockito;
 
-import java.util.Arrays;
+import java.util.List;
 import java.util.Optional;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -107,7 +108,10 @@ public class PodTemplateTest {
 
         kc.setSpec(keycloakSpecBuilder.build());
 
+        var managedDependentResourceContext = new DefaultManagedDependentResourceContext();
+        //noinspection unchecked
         Context<Keycloak> context = Mockito.mock(Context.class);
+        Mockito.when(context.managedDependentResourceContext()).thenReturn(managedDependentResourceContext);
         Mockito.when(context.getSecondaryResource(StatefulSet.class)).thenReturn(Optional.ofNullable(existingDeployment));
 
         return deployment.desired(kc, context);
@@ -571,14 +575,14 @@ public class PodTemplateTest {
                 .getSpec().getTemplate();
 
         // Assert
-        assertThat(podTemplate.getSpec().getTolerations()).isEqualTo(Arrays.asList(toleration));
+        assertThat(podTemplate.getSpec().getTolerations()).isEqualTo(List.of(toleration));
 
         podTemplate = getDeployment(new PodTemplateSpecBuilder().withNewSpec().withTolerations(new Toleration()).endSpec().build(), null,
                 s -> s.withNewSchedulingSpec().addToTolerations(toleration).endSchedulingSpec())
                 .getSpec().getTemplate();
 
         // Assert
-        assertThat(podTemplate.getSpec().getTolerations()).isNotEqualTo(Arrays.asList(toleration));
+        assertThat(podTemplate.getSpec().getTolerations()).isNotEqualTo(List.of(toleration));
 
     }
 
@@ -595,14 +599,14 @@ public class PodTemplateTest {
                 .getSpec().getTemplate();
 
         // Assert
-        assertThat(podTemplate.getSpec().getTopologySpreadConstraints()).isEqualTo(Arrays.asList(tsc));
+        assertThat(podTemplate.getSpec().getTopologySpreadConstraints()).isEqualTo(List.of(tsc));
 
         podTemplate = getDeployment(new PodTemplateSpecBuilder().withNewSpec().withTopologySpreadConstraints(new TopologySpreadConstraint()).endSpec().build(), null,
                 s -> s.withNewSchedulingSpec().addToTopologySpreadConstraints(tsc).endSchedulingSpec())
                 .getSpec().getTemplate();
 
         // Assert
-        assertThat(podTemplate.getSpec().getTopologySpreadConstraints()).isNotEqualTo(Arrays.asList(tsc));
+        assertThat(podTemplate.getSpec().getTopologySpreadConstraints()).isNotEqualTo(List.of(tsc));
     }
 
     @Test

operator/src/test/java/org/keycloak/operator/testsuite/utils/CRAssert.java

@@ -22,6 +22,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -285,4 +286,26 @@ public final class CRAssert {
                 .filter(rule -> rule.getPorts().stream().anyMatch(port -> port.getPort().getIntVal() == rulePort))
                 .findFirst();
     }
+
+    public static CompletableFuture<List<Keycloak>> eventuallyRollingUpgradeStatus(KubernetesClient client, Keycloak keycloak) {
+        return client.resource(keycloak).informOnCondition(kcs -> {
+            try {
+                assertKeycloakStatusCondition(kcs.get(0), KeycloakStatusCondition.ROLLING_UPDATE, true, "Rolling out deployment update");
+                return true;
+            } catch (AssertionError e) {
+                return false;
+            }
+        });
+    }
+
+    public static CompletableFuture<List<Keycloak>> eventuallyRecreateUpgradeStatus(KubernetesClient client, Keycloak keycloak) {
+        return client.resource(keycloak).informOnCondition(kcs -> {
+            try {
+                assertKeycloakStatusCondition(kcs.get(0), KeycloakStatusCondition.READY, false, "Performing Keycloak upgrade");
+                return true;
+            } catch (AssertionError e) {
+                return false;
+            }
+        });
+    }
 }

operator/src/test/resources/test-serialization-keycloak-cr-invalid-update.yml

@@ -0,0 +1,9 @@
+apiVersion: k8s.keycloak.org/v2alpha1
+kind: Keycloak
+metadata:
+  name: test-serialization-kc
+spec:
+  instances: 3
+  image: my-image
+  update:
+    strategy: abc

operator/src/test/resources/test-serialization-keycloak-cr.yml

@@ -123,6 +123,8 @@ spec:
       secret: something
     service:
       secret: else
+  update:
+    strategy: Recreate
   unsupported:
     podTemplate:
       metadata: