package integration

import (
	"context"
	"fmt"
	"math/rand"
	"sync"
	"time"

	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
	asappsv1 "github.com/pingcap/advanced-statefulset/client/apis/apps/v1"
	oplabel "github.com/pingcap/tidb-operator/pkg/apis/label"
	operatorv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/eks-provider/controllers/configchange"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	config2 "github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/namer"
)

var _ = Describe("ConfigChange controller", func() {
	var (
		ctx           context.Context
		testNamespace string

		network               *cloudv1.Network
		cluster2SQLExecutor   = map[string]*fakeSQLExecutor{}
		cluster2SQLExecutorMu = sync.Mutex{}
	)
	getSQLExecutor := func(clusterName string) *fakeSQLExecutor {
		cluster2SQLExecutorMu.Lock()
		defer cluster2SQLExecutorMu.Unlock()

		_, ok := cluster2SQLExecutor[clusterName]
		if !ok {
			cluster2SQLExecutor[clusterName] = newFakeSQLExecutor()
		}
		return cluster2SQLExecutor[clusterName]
	}

	BeforeEach(func() {
		configchange.RequeueAfterWhenConditionNotMet = time.Second
		configchange.RequeueAfterWhenCCNotDone = time.Second

		configchange.SQLExecutorGetter = func(ctx context.Context, cluster *cloudv1.Cluster, cloudAdminPasswdGetter configchange.CloudAdminPasswdGetter, conf config2.RootConfig) (configchange.SQLExecutor, error) {
			return getSQLExecutor(cluster.Name), nil
		}

		ctx = context.Background()
		testNamespace = genNamespace()
		network, _, _ = prepareNetwork(testNamespace, ctx)

	})

	AfterEach(func() {
		//// cleanup clusters
		//cs := &cloudv1.ClusterList{}
		//Expect(k8sClient.List(ctx, cs, client.InNamespace(testNamespace))).To(Succeed())
		//for _, c := range cs.Items {
		//	Expect(k8sClient.Delete(context.TODO(), &c)).To(Succeed())
		//}
		//
		//// cleanup configchange
		//ccs := &cloudv1.ConfigChangeList{}
		//Expect(k8sClient.List(ctx, ccs, client.InNamespace(testNamespace))).To(Succeed())
		//for _, c := range ccs.Items {
		//	Expect(k8sClient.Delete(context.TODO(), &c)).To(Succeed())
		//}
	})

	Context("when configchangs are created", func() {
		It("Should add labels to configchange by webhook when create a new configchange", func(ctx context.Context) {
			By("Create a cluster")
			cluster := newLegacyCluster(network, "1", "11111", "i3.2xlarge")
			Expect(k8sClient.Create(ctx, cluster)).To(Succeed())
			Expect(cluster.Spec.ConfigRevisionRef).ToNot(BeNil())
			rev, err := getClusterRevision(k8sClient, client.ObjectKeyFromObject(cluster))
			Expect(err).To(BeNil())

			cc := newConfigChange(cluster, rev.Name, cloudv1.ConfigChangeStrategyHotReload)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())
			_, existt := rev.Labels[cloudv1.TenantTagKey]
			_, existc := rev.Labels[cloudv1.ClusterTagKey]

			Expect(existt && existc).
				To(BeTrue(), "%s should have both tenant and cluster label", rev.Labels)
		}, NodeTimeout(time.Second*5))
		It("Should return error when creating a new configchange with NON-latest ConfigRevision", func(ctx context.Context) {
			By("Create a cluster")
			cluster := newLegacyCluster(network, "1", "11111", "i3.2xlarge")
			Expect(k8sClient.Create(ctx, cluster)).To(Succeed())
			Expect(cluster.Spec.ConfigRevisionRef).ToNot(BeNil())
			rev1 := newClusterRevision(cluster)
			Expect(k8sClient.Create(ctx, rev1)).To(Succeed())
			time.Sleep(time.Second) // ensure rev2 is created after rev1
			rev2 := newClusterRevision(cluster)
			Expect(k8sClient.Create(ctx, rev2)).To(Succeed())

			cc := newConfigChange(cluster, rev1.Name, cloudv1.ConfigChangeStrategyHotReload)
			Expect(k8sClient.Create(ctx, cc)).To(MatchError(ContainSubstring("is NOT latest")))
		}, NodeTimeout(time.Second*5))
		It("Should return error if strategy is HotReload but contains image changes", func(ctx context.Context) {
			By("Create a cluster")
			cluster := newLegacyCluster(network, "1", "11111", "i3.2xlarge")
			Expect(k8sClient.Create(ctx, cluster)).To(Succeed())
			rev, err := getClusterRevision(k8sClient, client.ObjectKeyFromObject(cluster))
			Expect(err).To(BeNil())

			By("Create ClusterRevision")
			time.Sleep(time.Second) // ensure rev1 is created after rev
			rev1 := newClusterRevision(cluster)
			rev1.Spec = rev.Spec // override spec
			rev1.Spec.TiDB.BaseImage = pointer.String("image1")
			Expect(k8sClient.Create(ctx, rev1)).To(Succeed())

			cc := newConfigChange(cluster, rev1.Name, cloudv1.ConfigChangeStrategyHotReload)
			Expect(k8sClient.Create(ctx, cc)).To(MatchError(ContainSubstring("image change")))
		}, NodeTimeout(time.Second*5))
		It("Should return error if there is another non-done configchange", func(ctx context.Context) {
			By("Create a cluster")
			cluster := newLegacyCluster(network, "1", "11111", "i3.2xlarge")
			Expect(k8sClient.Create(ctx, cluster)).To(Succeed())
			rev1, err := getClusterRevision(k8sClient, client.ObjectKeyFromObject(cluster))
			Expect(err).To(BeNil())

			By("Create ConfigChange1")
			cc := newConfigChange(cluster, rev1.Name, cloudv1.ConfigChangeStrategyHotReload)
			Expect(k8sClient.Create(ctx, cc)).To(BeNil())

			By("Create ClusterRevision")
			time.Sleep(time.Second) // ensure rev2 is created after rev1
			rev2 := newClusterRevision(cluster)
			rev2.Spec = rev1.Spec // override spec
			rev2.Spec.TiDB.Config = cloudv1.NewTomlConfig(map[string]interface{}{"a": "b"})
			Expect(k8sClient.Create(ctx, rev2)).To(Succeed())

			By("Create ConfigChange2")
			cc2 := newConfigChange(cluster, rev2.Name, cloudv1.ConfigChangeStrategyHotReload)
			Expect(k8sClient.Create(ctx, cc2)).To(MatchError(ContainSubstring("is under config changing")))
		}, NodeTimeout(time.Second*5))
	})

	Context("controller", func() {
		It("tc is paused", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				cr2.Spec.TiKV = &cloudv1.Configuration{}
				cr1.Spec.TiKV.DeepCopyInto(cr2.Spec.TiKV) // copy config from cr1 into cr2
				cr2.Spec.TiKV.Config.MP["a"] = "b"        // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli

			// pause tc
			objects.tc.Spec.Paused = true
			Expect(remoteCli.Update(ctx, objects.tc)).To(Succeed())

			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyRollingRestart)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())
			go func() {
				time.Sleep(time.Second)
				simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
			}()

			Eventually(func() error {
				// wait cluster to be config changing
				actualCluster := &cloudv1.Cluster{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(objects.cluster), actualCluster)).To(Succeed())
				if !actualCluster.ConfigChanging() {
					return fmt.Errorf("cluster is not under config changing")
				}

				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if step, ok := actualCC.Step(ctx, cloudv1.CCStepUpdateConfig2CM); ok && step.Status == cloudv1.CCStepStatusRunning {
					return nil
				} else {
					return fmt.Errorf("wait cc to be pending on tc")
				}
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentTiKV}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Steps).To(HaveLen(4))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseRunning))
			step, _ := actualCC.Step(ctx, cloudv1.CCStepUpdateConfig2CM)
			Expect(step.Reason).To(Equal(fmt.Sprintf("tc(%s) is paused, retry next time", objects.tc.Name)))
		}, NodeTimeout(time.Second*10))

		It("gc when the cluster doesn't exist", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				cr2.Spec.TiKV = &cloudv1.Configuration{}
				cr1.Spec.TiKV.DeepCopyInto(cr2.Spec.TiKV) // copy config from cr1 into cr2
				cr2.Spec.TiKV.Config.MP["a"] = "b"        // add a new configuration item into cr2
			})

			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyRollingRestart)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())
			// mark CC as Completed
			cc.Status.Phase = cloudv1.ConfigChangePhaseCompleted
			cc.Status.EndTime = &metav1.Time{Time: time.Now().Add(-2 * configchange.DefaultTTL)}
			Expect(k8sClient.Status().Update(ctx, cc)).To(Succeed())
			// delete the owner cluster
			Expect(k8sClient.Delete(ctx, objects.cluster)).To(Succeed())

			Eventually(func() error {
				// wait the ConfigChange to be deleted
				err := k8sClient.Get(ctx, client.ObjectKeyFromObject(objects.cluster), &cloudv1.ConfigChange{})
				if apierrors.IsNotFound(err) {
					return nil
				}
				return err
			}, timeout, interval).Should(Succeed())
		}, NodeTimeout(time.Second*10))
	})

	Context("RollingRestart", func() {
		It("change nothing", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				// change nothing
			})
			remoteCli := objects.remoteCli

			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyRollingRestart)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())
			go func() {
				time.Sleep(time.Second)
				simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
			}()

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if !actualCC.Done() {
					return fmt.Errorf("ConfigChange is not done")
				}
				actualCluster := &cloudv1.Cluster{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(objects.cluster), actualCluster)).To(Succeed())
				if actualCluster.ConfigChanging() {
					return fmt.Errorf("cluster is config changing")
				}

				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(BeNil())
			Expect(actualCC.Status.ChangedCategories).To(BeNil())
			Expect(actualCC.Status.Steps).To(Equal([]*cloudv1.ConfigChangeStep{
				{
					Name:   "UpdateCR2Cluster",
					Status: "Completed",
					Reason: fmt.Sprintf("updated %s to cluster(%s)", objects.cr2.Name, objects.cluster.Name),
				},
				{
					Name:   "UpdateConfig2CM",
					Status: "Completed",
					Reason: "config of components have been written to cm",
				},
				{
					Name:   "EnsureTCReady",
					Status: "Completed",
					Reason: "nonReadyTCNames: []",
				},
			}))
			Expect(actualCC.Status.DB.TiKV).To(BeNil())
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))

			// check cluster
			actualCluster := &cloudv1.Cluster{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(objects.cluster), actualCluster)).To(Succeed())
			Expect(actualCluster.Spec.ConfigRevisionRef.Name).To(Equal(objects.cr2.Name))

			// check tc
			actualTC := &operatorv1alpha1.TidbCluster{}
			Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.tc), actualTC)).To(Succeed())
			// both tikv and tidb are NOT restarted
			Expect(actualTC.Spec.TiKV.Annotations).ToNot(HaveKeyWithValue(configchange.RestartByConfigChangeAnnotation, cc.Name))
			Expect(actualTC.Spec.TiDB.Annotations).ToNot(HaveKeyWithValue(configchange.RestartByConfigChangeAnnotation, cc.Name))
		}, NodeTimeout(time.Second*10))

		It("version changed but all component images are NOT changed", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				cr2.Spec.Version = "v7.5.0"
			})
			remoteCli := objects.remoteCli

			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyRollingRestart)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())
			go func() {
				time.Sleep(time.Second)
				simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
			}()

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if !actualCC.Done() {
					return fmt.Errorf("ConfigChange is not done")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(BeNil())
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryVersion}))
			Expect(actualCC.Status.Steps).To(Equal([]*cloudv1.ConfigChangeStep{
				{
					Name:   "UpdateCR2Cluster",
					Status: "Completed",
					Reason: fmt.Sprintf("updated %s to cluster(%s)", objects.cr2.Name, objects.cluster.Name),
				},
				{
					Name:   "UpdateConfig2CM",
					Status: "Completed",
					Reason: "config of components have been written to cm",
				},
				{
					Name:   "EnsureTCReady",
					Status: "Completed",
					Reason: "nonReadyTCNames: []",
				},
			}))
			Expect(actualCC.Status.DB.TiKV).To(BeNil())
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))

			// check tc
			actualTC := &operatorv1alpha1.TidbCluster{}
			Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.tc), actualTC)).To(Succeed())
			// both tikv and tidb are NOT restarted
			Expect(actualTC.Spec.TiKV.Annotations).ToNot(HaveKeyWithValue(configchange.RestartByConfigChangeAnnotation, cc.Name))
			Expect(actualTC.Spec.TiDB.Annotations).ToNot(HaveKeyWithValue(configchange.RestartByConfigChangeAnnotation, cc.Name))
		}, NodeTimeout(time.Second*10))

		It("version changed but only tikv's image is changed", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				cr2.Spec.Version = "v7.5.0"
				cr2.Spec.TiKV.BaseImage = pointer.String("v7.5.0")
			})
			remoteCli := objects.remoteCli

			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyRollingRestart)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())
			go func() {
				time.Sleep(time.Second)
				simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
			}()

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if !actualCC.Done() {
					return fmt.Errorf("ConfigChange is not done")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentTiKV}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryVersion, cloudv1.ConfigChangeCategoryImage}))
			Expect(actualCC.Status.Steps).To(Equal([]*cloudv1.ConfigChangeStep{
				{
					Name:   "UpdateCR2Cluster",
					Status: "Completed",
					Reason: fmt.Sprintf("updated %s to cluster(%s)", objects.cr2.Name, objects.cluster.Name),
				},
				{
					Name:   "UpdateConfig2CM",
					Status: "Completed",
					Reason: "config of components have been written to cm",
				},
				{
					Name:   "RestartComponent/tikv",
					Status: "Completed",
					Reason: fmt.Sprintf("updatedReplicas: %d, readyReplicas: 0, phase: Normal", objects.tikvSTS.Spec.Replicas),
				},
				{
					Name:   "EnsureTCReady",
					Status: "Completed",
					Reason: "nonReadyTCNames: []",
				},
			}))
			Expect(actualCC.Status.DB.TiKV).To(Equal(&cloudv1.ConfigChangeComponentStatus{
				UpdatedReplicas:   objects.tikvSTS.Status.UpdatedReplicas,
				Done:              true,
				ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryImage},
			}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))

			// check tc
			actualTC := &operatorv1alpha1.TidbCluster{}
			Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.tc), actualTC)).To(Succeed())
			// Because the image of tikv is changed, so we don't need to manually restart tikv by the following annotation.
			Expect(actualTC.Spec.TiKV.Annotations).ToNot(HaveKeyWithValue(configchange.RestartByConfigChangeAnnotation, cc.Name))
			Expect(actualTC.Spec.TiDB.Annotations).ToNot(HaveKeyWithValue(configchange.RestartByConfigChangeAnnotation, cc.Name))
		}, NodeTimeout(time.Second*10))

		It("only tikv's config changed", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				cr2.Spec.TiKV = &cloudv1.Configuration{}
				cr1.Spec.TiKV.DeepCopyInto(cr2.Spec.TiKV) // copy config from cr1 into cr2
				cr2.Spec.TiKV.Config.MP["a"] = "b"        // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli

			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyRollingRestart)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())
			go func() {
				time.Sleep(time.Second)
				simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
			}()

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if !actualCC.Done() {
					return fmt.Errorf("ConfigChange is not done")
				}

				actualCluster := &cloudv1.Cluster{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(objects.cluster), actualCluster)).To(Succeed())
				if actualCluster.ConfigChanging() {
					return fmt.Errorf("cluster is config changing")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentTiKV}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Steps).To(HaveLen(4))
			Expect(actualCC.Status.DB.TiKV).To(Equal(&cloudv1.ConfigChangeComponentStatus{
				UpdatedReplicas:   objects.tikvSTS.Status.UpdatedReplicas,
				Done:              true,
				ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
			}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))

			// check tc
			// only tikv have been annotated with RestartByConfigChangeAnnotation
			actualTC := &operatorv1alpha1.TidbCluster{}
			Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.tc), actualTC)).To(Succeed())
			Expect(actualTC.Spec.TiKV.Annotations).To(HaveKeyWithValue(configchange.RestartByConfigChangeAnnotation, cc.Name))
			Expect(actualTC.Spec.TiDB.Annotations).ToNot(HaveKeyWithValue(configchange.RestartByConfigChangeAnnotation, cc.Name))
		}, NodeTimeout(time.Second*10))

		It("only pd's config changed", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				cr2.Spec.PD = &cloudv1.Configuration{}
				cr1.Spec.PD.DeepCopyInto(cr2.Spec.PD) // copy config from cr1 into cr2
				cr2.Spec.PD.Config.MP["a"] = "b"      // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli

			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyRollingRestart)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if !actualCC.Done() {
					simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
					return fmt.Errorf("ConfigChange is not done")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentPD}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Steps).To(Equal([]*cloudv1.ConfigChangeStep{
				{
					Name:   "UpdateCR2Cluster",
					Status: "Completed",
					Reason: fmt.Sprintf("updated %s to cluster(%s)", objects.cr2.Name, objects.cluster.Name),
				},
				{
					Name:   "UpdateConfig2CM",
					Status: "Completed",
					Reason: "config of components have been written to cm",
				},
				{
					Name:   "RestartComponent/pd",
					Status: "Completed",
					Reason: fmt.Sprintf("updatedReplicas: %d, readyReplicas: 0, phase: Normal", objects.pdSTS.Spec.Replicas),
				},
				{
					Name:   "EnsureTCReady",
					Status: "Completed",
					Reason: "nonReadyTCNames: []",
				},
			}))
			Expect(actualCC.Status.DB.PD).To(Equal(&cloudv1.ConfigChangeComponentStatus{
				UpdatedReplicas:   objects.pdSTS.Status.UpdatedReplicas,
				Done:              true,
				ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
			}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))

			// check tc
			// only pd have been annotated with RestartByConfigChangeAnnotation
			actualTC := &operatorv1alpha1.TidbCluster{}
			Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.tc), actualTC)).To(Succeed())
			Expect(actualTC.Spec.PD.Annotations).To(HaveKeyWithValue(configchange.RestartByConfigChangeAnnotation, cc.Name))
			Expect(actualTC.Spec.TiDB.Annotations).ToNot(HaveKeyWithValue(configchange.RestartByConfigChangeAnnotation, cc.Name))
		}, NodeTimeout(time.Second*20))

		It("tiproxy always use hotReload", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				// change tiproxy config only
				cr2.Spec.TiProxy = &cloudv1.Configuration{Config: cloudv1.NewTomlConfig(map[string]interface{}{})}
				cr2.Spec.TiProxy.Config.MP["a"] = "b" // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli

			By("Create CC")
			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyRollingRestart)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())
			go func() {
				time.Sleep(time.Second)
				simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
			}()

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}

				if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC); err != nil {
					return err
				}
				if !actualCC.Done() {
					return fmt.Errorf("ConfigChange is not done")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentTiProxy}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Steps).To(Equal([]*cloudv1.ConfigChangeStep{
				{
					Name:   "UpdateCR2Cluster",
					Status: "Completed",
					Reason: fmt.Sprintf("updated %s to cluster(%s)", objects.cr2.Name, objects.cluster.Name),
				},
				{
					Name:   "UpdateConfig2CM",
					Status: "Completed",
					Reason: "config of components have been written to cm",
				},
				{
					Name:   "HotReloadComponent/tiproxy",
					Status: "Completed",
					Reason: "TiProxy config have been auto reloaded by itself",
				},
				{
					Name:   "EnsureTCReady",
					Status: "Completed",
					Reason: "nonReadyTCNames: []",
				},
			}))
			Expect(actualCC.Status.DB.TiProxy).To(Equal(&cloudv1.ConfigChangeComponentStatus{
				UpdatedReplicas:   int32(objects.cluster.Spec.Cluster.TiProxy.Replicas),
				Done:              true,
				ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
			}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))

			// check tc
			// tiproxy shouldn't be annotated with RestartByConfigChangeAnnotation
			actualTC := &operatorv1alpha1.TidbCluster{}
			Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.tc), actualTC)).To(Succeed())
			Expect(actualTC.Spec.TiProxy).To(BeNil())
		}, NodeTimeout(time.Second*10))

		It("change both tikv, pd and tiproxy", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				// change tiproxy config
				cr2.Spec.TiProxy = &cloudv1.Configuration{Config: cloudv1.NewTomlConfig(map[string]interface{}{})}
				cr2.Spec.TiProxy.Config.MP["a"] = "b" // add a new configuration item into cr2

				cr2.Spec.PD = &cloudv1.Configuration{}
				cr1.Spec.PD.DeepCopyInto(cr2.Spec.PD) // copy config from cr1 into cr2
				cr2.Spec.PD.Config.MP["a"] = "b"      // add a new configuration item into cr2

				cr2.Spec.TiKV = &cloudv1.Configuration{}
				cr1.Spec.TiKV.DeepCopyInto(cr2.Spec.TiKV) // copy config from cr1 into cr2
				cr2.Spec.TiKV.Config.MP["a"] = "b"        // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli

			By("Create CC")
			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyRollingRestart)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())
			go func() {
				time.Sleep(time.Second)
			}()

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}

				if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC); err != nil {
					return err
				}
				if !actualCC.Done() {
					simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
					return fmt.Errorf("ConfigChange is not done")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentPD, cloudv1.ComponentTiKV, cloudv1.ComponentTiProxy}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Steps).To(Equal([]*cloudv1.ConfigChangeStep{
				{
					Name:   "UpdateCR2Cluster",
					Status: "Completed",
					Reason: fmt.Sprintf("updated %s to cluster(%s)", objects.cr2.Name, objects.cluster.Name),
				},
				{
					Name:   "UpdateConfig2CM",
					Status: "Completed",
					Reason: "config of components have been written to cm",
				},
				{
					Name:   "RestartComponent/pd",
					Status: "Completed",
					Reason: fmt.Sprintf("updatedReplicas: %d, readyReplicas: 0, phase: Normal", objects.pdSTS.Spec.Replicas),
				},
				{
					Name:   "HotReloadComponent/tiproxy",
					Status: "Completed",
					Reason: "TiProxy config have been auto reloaded by itself",
				},
				{
					Name:   "RestartComponent/tikv",
					Status: "Completed",
					Reason: fmt.Sprintf("updatedReplicas: %d, readyReplicas: 0, phase: Normal", objects.tikvSTS.Spec.Replicas),
				},
				{
					Name:   "EnsureTCReady",
					Status: "Completed",
					Reason: "nonReadyTCNames: []",
				},
			}))
			Expect(actualCC.Status.DB.TiProxy).To(Equal(&cloudv1.ConfigChangeComponentStatus{
				UpdatedReplicas:   int32(objects.cluster.Spec.Cluster.TiProxy.Replicas),
				Done:              true,
				ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
			}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))

			// check tc
			// tiproxy shouldn't be annotated with RestartByConfigChangeAnnotation
			actualTC := &operatorv1alpha1.TidbCluster{}
			Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.tc), actualTC)).To(Succeed())
			Expect(actualTC.Spec.TiProxy).To(BeNil())
		}, NodeTimeout(time.Second*20))

		It("tiflashLearner work fine and should use tiflash to restart", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				// change tiflashLearner config only
				cr2.Spec.TiFlashLearner.Config.MP["a"] = "b" // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli

			By("Create CC")
			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyRollingRestart)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())
			go func() {
				time.Sleep(time.Second)
				simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
			}()

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}

				if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC); err != nil {
					return err
				}
				if !actualCC.Done() {
					return fmt.Errorf("ConfigChange is not done")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentTiFlashLearner}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Steps).To(Equal([]*cloudv1.ConfigChangeStep{
				{
					Name:   "UpdateCR2Cluster",
					Status: "Completed",
					Reason: fmt.Sprintf("updated %s to cluster(%s)", objects.cr2.Name, objects.cluster.Name),
				},
				{
					Name:   "UpdateConfig2CM",
					Status: "Completed",
					Reason: "config of components have been written to cm",
				},
				{
					Name:   "RestartComponent/tiflash-learner",
					Status: "Completed",
					Reason: fmt.Sprintf("updatedReplicas: %d, readyReplicas: 0, phase: Normal", objects.tiflashSTS.Spec.Replicas),
				},
				{
					Name:   "EnsureTCReady",
					Status: "Completed",
					Reason: "nonReadyTCNames: []",
				},
			}))
			Expect(actualCC.Status.DB.TiFlashLearner).To(Equal(&cloudv1.ConfigChangeComponentStatus{
				UpdatedReplicas:   objects.tikvSTS.Status.UpdatedReplicas,
				Done:              true,
				ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
			}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))

			// check tc
			// only tiflash have been annotated with RestartByConfigChangeAnnotation
			// RestartByConfigChangeAnnotation should be annotated on tiflash component
			actualTC := &operatorv1alpha1.TidbCluster{}
			Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.tc), actualTC)).To(Succeed())
			Expect(actualTC.Spec.TiFlash.Annotations).To(HaveKeyWithValue(configchange.RestartByConfigChangeAnnotation, cc.Name))
		}, NodeTimeout(time.Second*10))

		It("change config for tiflashLearner but tiflash don't exists should work fine", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				// change tiflash config only
				cr2.Spec.TiFlashLearner.Config.MP["a"] = "b" // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli
			// delete tiflash from Cluster and TC
			Eventually(func() error {
				objects.cluster.Spec.Cluster.TiFlash = cloudv1.TiFlash{}
				if err := k8sClient.Update(ctx, objects.cluster); err != nil {
					return err
				}

				objects.tc.Spec.TiFlash = nil
				if err := remoteCli.Update(ctx, objects.tc); err != nil {
					return err
				}
				return nil
			}, timeout, interval).Should(Succeed())

			By("Create CC")
			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyRollingRestart)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())

			sqlExecutor := getSQLExecutor(objects.cluster.Name)

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if !actualCC.Done() {
					simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
					return fmt.Errorf("ConfigChange is not done")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentTiFlashLearner}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Steps).To(Equal([]*cloudv1.ConfigChangeStep{
				{
					Name:   "UpdateCR2Cluster",
					Status: "Completed",
					Reason: fmt.Sprintf("updated %s to cluster(%s)", objects.cr2.Name, objects.cluster.Name),
				},
				{
					Name:   "UpdateConfig2CM",
					Status: "Completed",
					Reason: "config of components have been written to cm",
				},
				{
					Name:   "RestartComponent/tiflash-learner",
					Status: "Completed",
					Reason: "tiflash-learner is not deployed",
				},
				{
					Name:   "EnsureTCReady",
					Status: "Completed",
					Reason: "nonReadyTCNames: []",
				},
			}))
			Expect(actualCC.Status.DB.TiFlashLearner).To(Equal(&cloudv1.ConfigChangeComponentStatus{
				UpdatedReplicas:   int32(objects.cluster.Spec.Cluster.TiFlash.Replicas),
				Done:              true,
				ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
			}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))

			// check executed SQLs, tiflash shouldn't execute any sql to apply config
			Expect(sqlExecutor.sqlSet).To(BeEmpty())
		}, NodeTimeout(time.Second*1000))

		It("change ticdc config but ticdc is inexistent should work fine", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				// change ticdc config only
				cr2.Spec.TiCDC = &cloudv1.Configuration{Config: cloudv1.NewTomlConfig(map[string]interface{}{})}
				cr2.Spec.TiCDC.Config.MP["a"] = "b" // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli

			By("Create CC")
			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyRollingRestart)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())
			go func() {
				time.Sleep(time.Second)
				simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
			}()

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}

				if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC); err != nil {
					return err
				}
				if !actualCC.Done() {
					return fmt.Errorf("ConfigChange is not done")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentTiCDC}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Steps).To(Equal([]*cloudv1.ConfigChangeStep{
				{
					Name:   "UpdateCR2Cluster",
					Status: "Completed",
					Reason: fmt.Sprintf("updated %s to cluster(%s)", objects.cr2.Name, objects.cluster.Name),
				},
				{
					Name:   "UpdateConfig2CM",
					Status: "Completed",
					Reason: "config of components have been written to cm",
				},
				{
					Name:   "RestartComponent/ticdc",
					Status: "Completed",
					Reason: "ticdc is not deployed",
				},
				{
					Name:   "EnsureTCReady",
					Status: "Completed",
					Reason: "nonReadyTCNames: []",
				},
			}))
			// ticdc componentStatus should be done
			Expect(actualCC.Status.DB.TiCDC).To(Equal(&cloudv1.ConfigChangeComponentStatus{
				UpdatedReplicas:   objects.tikvSTS.Status.UpdatedReplicas,
				Done:              true,
				ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
			}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))

			// check tc
			// ticdc haven't been annotated with RestartByConfigChangeAnnotation
			actualTC := &operatorv1alpha1.TidbCluster{}
			Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.tc), actualTC)).To(Succeed())
			// we don't have ticdc in tc
			Expect(actualTC.Spec.TiCDC).To(BeNil())
		}, NodeTimeout(time.Second*10))

		It("with extra group", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				cr2.Spec.TiDB = &cloudv1.Configuration{}
				cr1.Spec.TiDB.DeepCopyInto(cr2.Spec.TiDB) // copy config from cr1 into cr2
				cr2.Spec.TiDB.Config.MP["a"] = "b"        // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli

			// create tc, asts and update cluster status for creating extra group
			var extraGroupTC *operatorv1alpha1.TidbCluster
			Eventually(func() error {
				// create extra group tc
				extraGroupNamer := namer.NewNamer().
					WithTiDBGroupName("group1").
					WithComponent(string(cloudv1.ComponentTiDB))
				extraGroupTC = newTC(objects.cluster)
				extraGroupTC.Spec.PD = nil
				extraGroupTC.Spec.TiKV = nil
				extraGroupTC.Spec.TiFlash = nil
				extraGroupTC.Name = extraGroupNamer.GetTiDBName()
				if err := remoteCli.Create(ctx, extraGroupTC); err != nil {
					if !apierrors.IsAlreadyExists(err) {
						return err
					}
				}

				extraGroupSTS := newSTS(objects.cluster, cloudv1.ComponentTiDB, []corev1.Volume{
					{
						Name: "db-tidb", // the name should have prefix like '{tc.Name}-{componentName}'
						VolumeSource: corev1.VolumeSource{
							ConfigMap: &corev1.ConfigMapVolumeSource{
								LocalObjectReference: corev1.LocalObjectReference{Name: objects.tidbCM.Name},
							},
						},
					},
				})
				extraGroupSTS.Name = extraGroupTC.Name + "-" + string(cloudv1.ComponentTiDB)
				Expect(remoteCli.Create(ctx, extraGroupSTS)).To(Succeed())

				// set extra group status in cluster
				objects.cluster.Spec.ExtraGroups = []cloudv1.ExtraGroupSpec{
					{
						Name:         "group1",
						TiDB:         objects.cluster.Spec.Cluster.TiDB,
						SqlEndpoints: objects.cluster.Spec.SqlEndpoints,
					},
					{
						Name: "deleted-group", // deleted group hasn't TC in dataplane
						TiDB: cloudv1.TiDB{
							Replicas:      0, // represent that the group is deleted
							InstanceType:  "m6g.xlarge",
							Audit:         nil,
							PrecheckAudit: nil,
							Security:      nil,
							SuspendAction: nil,
						},
						SqlEndpoints: objects.cluster.Spec.SqlEndpoints,
					},
				}
				if err := k8sClient.Update(ctx, objects.cluster); err != nil {
					return err
				}
				objects.cluster.Status.ExtraEndpoints = []cloudv1.ExtraGroupStatus{
					{
						Name: "group1",
					},
				}
				return k8sClient.Status().Update(ctx, objects.cluster)
			}, timeout, interval).Should(Succeed())

			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyRollingRestart)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if !actualCC.Done() {
					simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
					return fmt.Errorf("ConfigChange is not done")
				}

				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentTiDB}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Steps).To(Equal([]*cloudv1.ConfigChangeStep{
				{
					Name:   "UpdateCR2Cluster",
					Status: "Completed",
					Reason: fmt.Sprintf("updated %s to cluster(%s)", objects.cr2.Name, objects.cluster.Name),
				},
				{
					Name:   "UpdateConfig2CM",
					Status: "Completed",
					Reason: "config of components have been written to cm",
				},
				{
					Name:   "RestartComponent/tidb",
					Status: "Completed",
					Reason: fmt.Sprintf("updatedReplicas: %d, readyReplicas: 0, phase: Normal", objects.tidbSTS.Spec.Replicas),
				},
				{
					Name:   "EnsureTCReady",
					Status: "Completed",
					Reason: "nonReadyTCNames: []",
				},
			}))
			Expect(actualCC.Status.DB.TiDB).To(Equal(&cloudv1.ConfigChangeComponentStatus{
				UpdatedReplicas:   objects.tidbSTS.Status.UpdatedReplicas,
				Done:              true,
				ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
			}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))

			// check tc
			// only tikv have been annotated with RestartByConfigChangeAnnotation
			actualTC := &operatorv1alpha1.TidbCluster{}
			Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.tc), actualTC)).To(Succeed())
			Expect(actualTC.Spec.ConfigUpdateStrategy).To(Equal(operatorv1alpha1.ConfigUpdateStrategyInPlace))
			Expect(actualTC.Spec.TiDB.Annotations).To(HaveKeyWithValue(configchange.RestartByConfigChangeAnnotation, cc.Name))

			actualExtraGroupTC := &operatorv1alpha1.TidbCluster{}
			Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(extraGroupTC), actualExtraGroupTC)).To(Succeed())
			Expect(actualExtraGroupTC.Spec.ConfigUpdateStrategy).To(Equal(operatorv1alpha1.ConfigUpdateStrategyInPlace))
			Expect(actualExtraGroupTC.Spec.TiDB.Annotations).To(HaveKeyWithValue(configchange.RestartByConfigChangeAnnotation, cc.Name))
		}, NodeTimeout(time.Second*100))
	})

	Context("HotReload", func() {
		It("change config for pd should work fine", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				// change pd config only
				cr2.Spec.PD = &cloudv1.Configuration{}
				cr1.Spec.PD.DeepCopyInto(cr2.Spec.PD) // copy config from cr1 into cr2
				cr2.Spec.PD.Config.MP["a"] = "b"      // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli

			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyHotReload)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())
			go func() {
				time.Sleep(time.Second)
				simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
			}()
			sqlExecutor := getSQLExecutor(objects.cluster.Name)

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if !actualCC.Done() {
					return fmt.Errorf("ConfigChange is not done")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentPD}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Steps).To(Equal([]*cloudv1.ConfigChangeStep{
				{
					Name:   "UpdateCR2Cluster",
					Status: "Completed",
					Reason: fmt.Sprintf("updated %s to cluster(%s)", objects.cr2.Name, objects.cluster.Name),
				},
				{
					Name:   "UpdateConfig2CM",
					Status: "Completed",
					Reason: "config of components have been written to cm",
				},
				{
					Name:   "HotReloadComponent/pd",
					Status: "Completed",
					Reason: "PD config have been updated by cluster-controller via HTTP API",
				},
			}))
			Expect(actualCC.Status.DB.PD).To(Equal(&cloudv1.ConfigChangeComponentStatus{
				UpdatedReplicas:   int32(objects.cluster.Spec.Cluster.PD.Replicas),
				Done:              true,
				ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
			}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))

			// check executed SQLs, pd shouldn't execute any sql to apply config
			Expect(sqlExecutor.sqlSet).To(BeEmpty())
		}, NodeTimeout(time.Second*10))

		It("change config for tiproxy should work fine", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				// change tiproxy config only
				cr2.Spec.TiProxy = &cloudv1.Configuration{Config: cloudv1.NewTomlConfig(map[string]interface{}{})}
				cr2.Spec.TiProxy.Config.MP["a"] = "b" // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli

			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyHotReload)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())
			go func() {
				time.Sleep(time.Second)
				simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
			}()
			sqlExecutor := getSQLExecutor(objects.cluster.Name)

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if !actualCC.Done() {
					return fmt.Errorf("ConfigChange is not done")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentTiProxy}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Steps).To(Equal([]*cloudv1.ConfigChangeStep{
				{
					Name:   "UpdateCR2Cluster",
					Status: "Completed",
					Reason: fmt.Sprintf("updated %s to cluster(%s)", objects.cr2.Name, objects.cluster.Name),
				},
				{
					Name:   "UpdateConfig2CM",
					Status: "Completed",
					Reason: "config of components have been written to cm",
				},
				{
					Name:   "HotReloadComponent/tiproxy",
					Status: "Completed",
					Reason: "TiProxy config have been auto reloaded by itself",
				},
			}))
			Expect(actualCC.Status.DB.TiProxy).To(Equal(&cloudv1.ConfigChangeComponentStatus{
				UpdatedReplicas:   int32(objects.cluster.Spec.Cluster.TiProxy.Replicas),
				Done:              true,
				ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
			}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))

			// check executed SQLs, tiproxy shouldn't execute any sql to apply config
			Expect(sqlExecutor.sqlSet).To(BeEmpty())
		}, NodeTimeout(time.Second*10))

		It("change config for tiflashLearner should work fine", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				// change tiflashLearner config only
				cr2.Spec.TiFlashLearner.Config.MP["raftstore.region-split-check-diff"] = "10MiB" // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli

			By("Create CC")
			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyHotReload)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())

			go func() {
				time.Sleep(time.Second)
				simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
			}()
			sqlExecutor := getSQLExecutor(objects.cluster.Name)
			// init sql query result
			sqlExecutor.sqlQueryResult["show config where type='tiflash' and name='raftstore-proxy.raftstore.region-split-check-diff'"] =
				`{"Instance":["aTE=","aTM="],"Value":["MTBNaUI=","MTBNaUI="]}`

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if !actualCC.Done() {
					return fmt.Errorf("ConfigChange is not done")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentTiFlashLearner}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Steps).To(Equal([]*cloudv1.ConfigChangeStep{
				{
					Name:   "UpdateCR2Cluster",
					Status: "Completed",
					Reason: fmt.Sprintf("updated %s to cluster(%s)", objects.cr2.Name, objects.cluster.Name),
				},
				{
					Name:   "UpdateConfig2CM",
					Status: "Completed",
					Reason: "config of components have been written to cm",
				},
				{
					Name:   "HotReloadComponent/tiflash-learner",
					Status: "Completed",
					Reason: "all config changes are applied via SQL",
				},
			}))
			Expect(actualCC.Status.DB.TiFlashLearner).To(Equal(&cloudv1.ConfigChangeComponentStatus{
				UpdatedReplicas:   int32(objects.cluster.Spec.Cluster.TiFlash.Replicas),
				Done:              true,
				ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
			}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))

			// check executed SQLs
			Expect(sqlExecutor.sqlSet).To(HaveKey("set config tiflash `raftstore-proxy.raftstore.region-split-check-diff`='10MiB'"))
		}, NodeTimeout(time.Second*10))

		It("change config for tiflash should work fine", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				// change tiflash config only
				cr2.Spec.TiFlash = &cloudv1.Configuration{}
				cr1.Spec.TiFlash.DeepCopyInto(cr2.Spec.TiFlash) // copy config from cr1 into cr2
				cr2.Spec.TiFlash.Config.MP["a"] = "b"           // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli

			By("Set annotation for tiflash")
			if objects.tc.Spec.TiFlash.Annotations == nil {
				objects.tc.Spec.TiFlash.Annotations = make(map[string]string)
			}
			objects.tc.Spec.TiFlash.Annotations[oplabel.AnnTiflashMountCMInTiflashContainer] = "true"
			Expect(remoteCli.Update(ctx, objects.tc)).To(Succeed())

			By("Create CC")
			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyHotReload)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())

			go func() {
				time.Sleep(time.Second)
				simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
			}()
			sqlExecutor := getSQLExecutor(objects.cluster.Name)

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if !actualCC.Done() {
					return fmt.Errorf("ConfigChange is not done")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentTiFlash}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))
			Expect(actualCC.Status.Steps).To(Equal([]*cloudv1.ConfigChangeStep{
				{
					Name:   "UpdateCR2Cluster",
					Status: "Completed",
					Reason: fmt.Sprintf("updated %s to cluster(%s)", objects.cr2.Name, objects.cluster.Name),
				},
				{
					Name:   "UpdateConfig2CM",
					Status: "Completed",
					Reason: "config of components have been written to cm",
				},
				{
					Name:   "HotReloadComponent/tiflash",
					Status: "Completed",
					Reason: "TiFlash config have been auto reloaded by itself",
				},
			}))
			Expect(actualCC.Status.DB.TiFlash).To(Equal(&cloudv1.ConfigChangeComponentStatus{
				UpdatedReplicas:   int32(objects.cluster.Spec.Cluster.TiFlash.Replicas),
				Done:              true,
				ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
			}))

			// check executed SQLs, tiflash shouldn't execute any sql to apply config
			Expect(sqlExecutor.sqlSet).To(BeEmpty())
		}, NodeTimeout(time.Second*10))

		It("change config for tiflashLearner but tiflash don't exists should work fine", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				// change tiflash config only
				cr2.Spec.TiFlashLearner.Config.MP["a"] = "b" // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli
			// delete tiflash from Cluster and TC
			Eventually(func() error {
				objects.cluster.Spec.Cluster.TiFlash = cloudv1.TiFlash{}
				if err := k8sClient.Update(ctx, objects.cluster); err != nil {
					return err
				}

				objects.tc.Spec.TiFlash = nil
				if err := remoteCli.Update(ctx, objects.tc); err != nil {
					return err
				}
				return nil
			}, timeout, interval).Should(Succeed())

			By("Create CC")
			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyHotReload)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())

			sqlExecutor := getSQLExecutor(objects.cluster.Name)

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if !actualCC.Done() {
					simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
					return fmt.Errorf("ConfigChange is not done")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentTiFlashLearner}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Steps).To(Equal([]*cloudv1.ConfigChangeStep{
				{
					Name:   "UpdateCR2Cluster",
					Status: "Completed",
					Reason: fmt.Sprintf("updated %s to cluster(%s)", objects.cr2.Name, objects.cluster.Name),
				},
				{
					Name:   "UpdateConfig2CM",
					Status: "Completed",
					Reason: "config of components have been written to cm",
				},
				{
					Name:   "HotReloadComponent/tiflash-learner",
					Status: "Completed",
					Reason: "tiflash-learner is not deployed",
				},
			}))
			Expect(actualCC.Status.DB.TiFlashLearner).To(Equal(&cloudv1.ConfigChangeComponentStatus{
				UpdatedReplicas:   int32(objects.cluster.Spec.Cluster.TiFlash.Replicas),
				Done:              true,
				ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
			}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))

			// check executed SQLs, tiflash shouldn't execute any sql to apply config
			Expect(sqlExecutor.sqlSet).To(BeEmpty())
		}, NodeTimeout(time.Second*1000))

		It("should fail if tiflash haven't AnnTiflashMountCMInTiflashContainer", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				// change tiflash config only
				cr2.Spec.TiFlash = &cloudv1.Configuration{}
				cr1.Spec.TiFlash.DeepCopyInto(cr2.Spec.TiFlash) // copy config from cr1 into cr2
				cr2.Spec.TiFlash.Config.MP["a"] = "b"           // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli

			By("Create CC")
			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyHotReload)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())

			go func() {
				time.Sleep(time.Second)
				simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
			}()
			sqlExecutor := getSQLExecutor(objects.cluster.Name)

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if !actualCC.Done() {
					return fmt.Errorf("ConfigChange is not done")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentTiFlash}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseFailed))
			Expect(actualCC.Status.Reason).To(ContainSubstring("AnnTiflashMountCMInTiflashContainer"))

			// check executed SQLs, tiflash shouldn't execute any sql to apply config
			Expect(sqlExecutor.sqlSet).To(BeEmpty())
		}, NodeTimeout(time.Second*10))

		It("change config for tikv should work fine", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				cr2.Spec.TiKV = &cloudv1.Configuration{}
				cr1.Spec.TiKV.DeepCopyInto(cr2.Spec.TiKV) // copy config from cr1 into cr2
				cr2.Spec.TiKV.Config.MP["a"] = "b"        // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli

			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyHotReload)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())
			go func() {
				time.Sleep(time.Second)
				simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
			}()
			sqlExecutor := getSQLExecutor(objects.cluster.Name)
			// init sql query result
			sqlExecutor.sqlQueryResult["show config where type='tikv' and name='a'"] =
				`{"Instance":["aTE=","aTI=","aTM="],"Value":["Yg==","Yg==","Yg=="]}`

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if !actualCC.Done() {
					return fmt.Errorf("ConfigChange is not done")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentTiKV}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Steps).To(HaveLen(3))
			Expect(actualCC.Status.DB.TiKV).To(Equal(&cloudv1.ConfigChangeComponentStatus{
				UpdatedReplicas:   int32(objects.cluster.Spec.Cluster.TiKV.Replicas),
				Done:              true,
				ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
			}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))

			// check executed SQLs
			Expect(sqlExecutor.sqlSet).To(HaveKey("set config tikv `a`='b'"))
		}, NodeTimeout(time.Second*10))

		It("change config with misused unit configuration for tikv should work fine", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				cr2.Spec.TiKV = &cloudv1.Configuration{}
				cr1.Spec.TiKV.DeepCopyInto(cr2.Spec.TiKV) // copy config from cr1 into cr2
				cr2.Spec.TiKV.Config.MP["a"] = "10MB"     // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli

			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyHotReload)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())
			go func() {
				time.Sleep(time.Second)
				simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
			}()
			sqlExecutor := getSQLExecutor(objects.cluster.Name)
			// init sql query result
			sqlExecutor.sqlQueryResult["show config where type='tikv' and name='a'"] =
				`{"Instance":["aTE=","aTI=","aTM="],"Value":["MTBNaUI=","MTBNaUI=","MTBNaUI="]}` // 'MTBNaUI=' is '10MiB'

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if !actualCC.Done() {
					return fmt.Errorf("ConfigChange is not done")
				}
				return nil
			}, timeout, interval).Should(Succeed())
		})

		It("change config for both tikv, tiflashLearner and tiproxy at the same time should work fine", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				// change config for tikv
				cr2.Spec.TiKV = &cloudv1.Configuration{}
				cr1.Spec.TiKV.DeepCopyInto(cr2.Spec.TiKV) // copy config from cr1 into cr2
				cr2.Spec.TiKV.Config.MP["a"] = "b"        // add a new configuration item into cr2

				// change config for tiflashLearner
				cr2.Spec.TiFlashLearner.Config.MP["c"] = "d" // add a new configuration item into cr2

				// change config for tiproxy
				cr2.Spec.TiProxy = &cloudv1.Configuration{Config: cloudv1.NewTomlConfig(map[string]interface{}{})}
				cr2.Spec.TiProxy.Config.MP["x"] = "y" // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli

			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyHotReload)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())
			go func() {
				time.Sleep(time.Second)
				simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
			}()
			sqlExecutor := getSQLExecutor(objects.cluster.Name)
			// init sql query result
			sqlExecutor.sqlQueryResult["show config where type='tikv' and name='a'"] =
				`{"Instance":["aTE=","aTI=","aTM="],"Value":["Yg==","Yg==","Yg=="]}`
			sqlExecutor.sqlQueryResult["show config where type='tiflash' and name='raftstore-proxy.c'"] =
				`{"Instance":["aTE=","aTI="],"Value":["ZA==","ZA=="]}`

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if !actualCC.Done() {
					return fmt.Errorf("ConfigChange is not done")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentTiKV, cloudv1.ComponentTiProxy, cloudv1.ComponentTiFlashLearner}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Steps).To(Equal([]*cloudv1.ConfigChangeStep{
				{
					Name:   "UpdateCR2Cluster",
					Status: "Completed",
					Reason: fmt.Sprintf("updated %s to cluster(%s)", objects.cr2.Name, objects.cluster.Name),
				},
				{
					Name:   "UpdateConfig2CM",
					Status: "Completed",
					Reason: "config of components have been written to cm",
				},
				{
					Name:   "HotReloadComponent/tiproxy",
					Status: "Completed",
					Reason: "TiProxy config have been auto reloaded by itself",
				},
				{
					Name:   "HotReloadComponent/tiflash-learner",
					Status: "Completed",
					Reason: "all config changes are applied via SQL",
				},
				{
					Name:   "HotReloadComponent/tikv",
					Status: "Completed",
					Reason: "all config changes are applied via SQL",
				},
			}))
			Expect(actualCC.Status.DB).To(Equal(cloudv1.ConfigChangeDBStatus{
				TiProxy: &cloudv1.ConfigChangeComponentStatus{
					UpdatedReplicas:   int32(objects.cluster.Spec.Cluster.TiProxy.Replicas),
					Done:              true,
					ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
				},
				TiFlashLearner: &cloudv1.ConfigChangeComponentStatus{
					UpdatedReplicas:   int32(objects.cluster.Spec.Cluster.TiFlash.Replicas),
					Done:              true,
					ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
				},
				TiKV: &cloudv1.ConfigChangeComponentStatus{
					UpdatedReplicas:   int32(objects.cluster.Spec.Cluster.TiKV.Replicas),
					Done:              true,
					ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
				},
			}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))

			// check executed SQLs
			Expect(sqlExecutor.sqlSet).To(HaveKey("set config tikv `a`='b'"))
			Expect(sqlExecutor.sqlSet).To(HaveKey("set config tiflash `raftstore-proxy.c`='d'"))
		}, NodeTimeout(time.Second*10))
	})

	Context("ChangeCMOnly", func() {
		It("only tikv's config changed", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				cr2.Spec.TiKV = &cloudv1.Configuration{}
				cr1.Spec.TiKV.DeepCopyInto(cr2.Spec.TiKV) // copy config from cr1 into cr2
				cr2.Spec.TiKV.Config.MP["a"] = "b"        // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli

			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyChangeCMOnly)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())
			go func() {
				time.Sleep(time.Second)
				simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
			}()

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if !actualCC.Done() {
					return fmt.Errorf("ConfigChange is not done")
				}

				actualCluster := &cloudv1.Cluster{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(objects.cluster), actualCluster)).To(Succeed())
				if actualCluster.ConfigChanging() {
					return fmt.Errorf("cluster is config changing")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentTiKV}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Steps).To(Equal([]*cloudv1.ConfigChangeStep{
				{
					Name:   "UpdateCR2Cluster",
					Status: "Completed",
					Reason: fmt.Sprintf("updated %s to cluster(%s)", objects.cr2.Name, objects.cluster.Name),
				},
				{
					Name:   "UpdateConfig2CM",
					Status: "Completed",
					Reason: "config of components have been written to cm",
				},
				{
					Name:   "ChangeCMOnly/mark-all-done",
					Status: "Completed",
					Reason: "mark all components done",
				},
			}))
			Expect(actualCC.Status.DB.TiKV).To(Equal(&cloudv1.ConfigChangeComponentStatus{
				UpdatedReplicas:   objects.tikvSTS.Status.UpdatedReplicas,
				Done:              true,
				Reason:            "skipped, because the strategy is ChangeCMOnly.",
				ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
			}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))

			// check tc
			// all component haven't been annotated with RestartByConfigChangeAnnotation
			actualTC := &operatorv1alpha1.TidbCluster{}
			Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.tc), actualTC)).To(Succeed())
			Expect(actualTC.Spec.TiKV.Annotations).ToNot(HaveKeyWithValue(configchange.RestartByConfigChangeAnnotation, cc.Name))
			Expect(actualTC.Spec.TiDB.Annotations).ToNot(HaveKeyWithValue(configchange.RestartByConfigChangeAnnotation, cc.Name))
		}, NodeTimeout(time.Second*10))
	})

	Context("EnsureUpdateToTC", func() {
		It("only tikv's config changed", func(ctx context.Context) {
			objects := initObjects4ConfigChange(ctx, network, func(cr1, cr2 *cloudv1.ConfigRevision) {
				cr2.Spec.TiKV = &cloudv1.Configuration{}
				cr1.Spec.TiKV.DeepCopyInto(cr2.Spec.TiKV) // copy config from cr1 into cr2
				cr2.Spec.TiKV.Config.MP["a"] = "b"        // add a new configuration item into cr2
			})
			remoteCli := objects.remoteCli

			cc := newReconcilableConfigChange(objects.cluster, objects.cr2.Name, cloudv1.ConfigChangeStrategyEnsureUpdateToTC)
			Expect(k8sClient.Create(ctx, cc)).To(Succeed())

			Eventually(func() error {
				actualCC := &cloudv1.ConfigChange{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
				if !actualCC.Done() {
					simulateOtherControllerBehavior(ctx, remoteCli, cc, objects)
					return fmt.Errorf("ConfigChange is not done")
				}

				actualCluster := &cloudv1.Cluster{}
				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(objects.cluster), actualCluster)).To(Succeed())
				if actualCluster.ConfigChanging() {
					return fmt.Errorf("cluster is config changing")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			// check configchange
			actualCC := &cloudv1.ConfigChange{}
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), actualCC)).To(Succeed())
			Expect(actualCC.Status.ChangedComponents).To(Equal([]cloudv1.ComponentType{cloudv1.ComponentTiKV}))
			Expect(actualCC.Status.ChangedCategories).To(Equal([]cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig}))
			Expect(actualCC.Status.Phase).To(Equal(cloudv1.ConfigChangePhaseCompleted))
			Expect(actualCC.Status.Steps).To(Equal([]*cloudv1.ConfigChangeStep{
				{
					Name:   "UpdateCR2Cluster",
					Status: "Completed",
					Reason: fmt.Sprintf("updated %s to cluster(%s)", objects.cr2.Name, objects.cluster.Name),
				},
				{
					Name:   "UpdateCR2TC",
					Status: "Completed",
					Reason: "ConfigRevision has been updated to TC",
				},
				{
					Name:   "EnsureUpdateToTC/mark-all-done",
					Status: "Completed",
					Reason: "mark all components done",
				},
			}))
			Expect(actualCC.Status.DB.TiKV).To(Equal(&cloudv1.ConfigChangeComponentStatus{
				UpdatedReplicas:   objects.tikvSTS.Status.UpdatedReplicas,
				Done:              true,
				Reason:            "skipped, because the strategy is EnsureUpdateToTC.",
				ChangedCategories: []cloudv1.ConfigChangeCategory{cloudv1.ConfigChangeCategoryConfig},
			}))

			// check tc
			// all component haven't been annotated with RestartByConfigChangeAnnotation
			actualTC := &operatorv1alpha1.TidbCluster{}
			Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.tc), actualTC)).To(Succeed())
			Expect(actualTC.Spec.TiKV.Annotations).ToNot(HaveKeyWithValue(configchange.RestartByConfigChangeAnnotation, cc.Name))
			Expect(actualTC.Spec.TiDB.Annotations).ToNot(HaveKeyWithValue(configchange.RestartByConfigChangeAnnotation, cc.Name))
		}, NodeTimeout(time.Second*10))
	})
})

type objects4ConfigChange struct {
	cluster    *cloudv1.Cluster
	cr1        *cloudv1.ConfigRevision
	cr2        *cloudv1.ConfigRevision
	pdCM       *corev1.ConfigMap
	tiproxyCM  *corev1.ConfigMap
	pdSTS      *asappsv1.StatefulSet
	tidbCM     *corev1.ConfigMap
	tidbSTS    *asappsv1.StatefulSet
	tikvCM     *corev1.ConfigMap
	tikvSTS    *asappsv1.StatefulSet
	tiflashCM  *corev1.ConfigMap
	tiflashSTS *asappsv1.StatefulSet
	tc         *operatorv1alpha1.TidbCluster
	remoteCli  client.Client
}

// initObjects4ConfigChange inits objects using for testing ConfigChange
func initObjects4ConfigChange(ctx context.Context, network *cloudv1.Network, crMutator func(cr1, cr2 *cloudv1.ConfigRevision)) *objects4ConfigChange {
	out := &objects4ConfigChange{}
	var err error
	By("Create a cluster")
	out.cluster = createRunningClusterHaving(ctx, network, "1", "11111", "i3.2xlarge")
	Expect(out.cluster.Spec.ConfigRevisionRef).ToNot(BeNil())
	out.cr1, err = getClusterRevision(k8sClient, client.ObjectKeyFromObject(out.cluster))
	Expect(err).To(BeNil())

	By("Create ClusterRevision cr2")
	time.Sleep(time.Second * 2) // ensure cr2 is created after cr1
	out.cr2 = newClusterRevision(out.cluster)
	out.cr2.Spec = out.cr1.Spec // override spec
	// pd,tidb,tiflash inherit config from cr1
	crMutator(out.cr1, out.cr2)
	Expect(k8sClient.Create(ctx, out.cr2)).To(Succeed())

	remoteCli, err := cachedK8sProvider.GetClient(ctx, out.cluster)
	Expect(err).To(BeNil())
	out.remoteCli = remoteCli

	By("Create pd")
	pdConfigBytes, _ := out.cr2.Spec.PD.Config.MarshalTOML()
	out.pdCM = newConfigMap(out.cluster, cloudv1.ComponentPD, map[string]string{
		"config-file": string(pdConfigBytes),
	})
	Expect(remoteCli.Create(ctx, out.pdCM)).To(Succeed())
	out.pdSTS = newSTS(out.cluster, cloudv1.ComponentPD, []corev1.Volume{
		{
			Name: "db-pd", // the name should have prefix like '{tc.Name}-{componentName}'
			VolumeSource: corev1.VolumeSource{
				ConfigMap: &corev1.ConfigMapVolumeSource{
					LocalObjectReference: corev1.LocalObjectReference{Name: out.pdCM.Name},
				},
			},
		},
	})
	Expect(remoteCli.Create(ctx, out.pdSTS)).To(Succeed())

	By("Create tiproxy")
	tiproxyConf := cloudv1.NewTomlConfig(map[string]interface{}{})
	if out.cr2.Spec.TiProxy != nil && out.cr2.Spec.TiProxy.Config != nil {
		tiproxyConf = out.cr2.Spec.TiProxy.Config
	}
	tiproxyConfigBytes, _ := tiproxyConf.MarshalTOML()
	out.tiproxyCM = newConfigMap(out.cluster, cloudv1.ComponentTiProxy, map[string]string{
		"config-file": string(tiproxyConfigBytes),
	})
	Expect(remoteCli.Create(ctx, out.tiproxyCM)).To(Succeed())
	//out.tiproxySTS = newSTS(out.cluster, cloudv1.ComponentTiProxy, []corev1.Volume{
	//	{
	//		Name: "db-tiproxy", // the name should have prefix like '{tc.Name}-{componentName}'
	//		VolumeSource: corev1.VolumeSource{
	//			ConfigMap: &corev1.ConfigMapVolumeSource{
	//				LocalObjectReference: corev1.LocalObjectReference{Name: out.tiproxyCM.Name},
	//			},
	//		},
	//	},
	//})
	//Expect(remoteCli.Create(ctx, out.tiproxySTS)).To(Succeed())

	By("Create tidb")
	tidbConfigBytes, _ := out.cr2.Spec.TiDB.Config.MarshalTOML()
	out.tidbCM = newConfigMap(out.cluster, cloudv1.ComponentTiDB, map[string]string{
		"config-file": string(tidbConfigBytes),
	})
	Expect(remoteCli.Create(ctx, out.tidbCM)).To(Succeed())
	out.tidbSTS = newSTS(out.cluster, cloudv1.ComponentTiDB, []corev1.Volume{
		{
			Name: "db-tidb", // the name should have prefix like '{tc.Name}-{componentName}'
			VolumeSource: corev1.VolumeSource{
				ConfigMap: &corev1.ConfigMapVolumeSource{
					LocalObjectReference: corev1.LocalObjectReference{Name: out.tidbCM.Name},
				},
			},
		},
	})
	Expect(remoteCli.Create(ctx, out.tidbSTS)).To(Succeed())

	By("Create tikv")
	tikvConfigBytes, _ := out.cr2.Spec.TiKV.Config.MarshalTOML()
	out.tikvCM = newConfigMap(out.cluster, cloudv1.ComponentTiKV, map[string]string{
		"config-file": string(tikvConfigBytes),
	})
	Expect(remoteCli.Create(ctx, out.tikvCM)).To(Succeed())
	out.tikvSTS = newSTS(out.cluster, cloudv1.ComponentTiKV, []corev1.Volume{
		{
			Name: "db-tikv", // the name should have prefix like '{tc.Name}-{componentName}'
			VolumeSource: corev1.VolumeSource{
				ConfigMap: &corev1.ConfigMapVolumeSource{
					LocalObjectReference: corev1.LocalObjectReference{Name: out.tikvCM.Name},
				},
			},
		},
	})
	Expect(remoteCli.Create(ctx, out.tikvSTS)).To(Succeed())

	By("Create tiflash")
	tiflashConfigBytes, _ := out.cr2.Spec.TiFlash.Config.MarshalTOML()
	tiflashLearnerConfigBytes, _ := out.cr2.Spec.TiFlashLearner.Config.MarshalTOML()
	out.tiflashCM = newConfigMap(out.cluster, cloudv1.ComponentTiFlash, map[string]string{
		"config_templ.toml": string(tiflashConfigBytes),
		"proxy_templ.toml":  string(tiflashLearnerConfigBytes),
	})
	Expect(remoteCli.Create(ctx, out.tiflashCM)).To(Succeed())
	out.tiflashSTS = newSTS(out.cluster, cloudv1.ComponentTiFlash, []corev1.Volume{
		{
			Name: "db-tiflash", // the name should have prefix like '{tc.Name}-{componentName}'
			VolumeSource: corev1.VolumeSource{
				ConfigMap: &corev1.ConfigMapVolumeSource{
					LocalObjectReference: corev1.LocalObjectReference{Name: out.tiflashCM.Name},
				},
			},
		},
	})
	Expect(remoteCli.Create(ctx, out.tiflashSTS)).To(Succeed())

	By("Create TC")
	out.tc = newTC(out.cluster)
	Expect(remoteCli.Create(ctx, out.tc)).To(Succeed())

	return out
}

func simulateOtherControllerBehavior(ctx context.Context, remoteCli client.Client, cc *cloudv1.ConfigChange, objects *objects4ConfigChange) {
	actualCluster := &cloudv1.Cluster{}
	Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(objects.cluster), actualCluster)).To(Succeed())

	// simulate cluster-controller update TC
	actualTC := &operatorv1alpha1.TidbCluster{}
	Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.tc), actualTC)).To(Succeed())
	actualTC.Annotations[cloudv1.ConfigRevisionIdKey] = cc.Spec.ConfigRevisionName
	Expect(remoteCli.Update(ctx, actualTC)).To(Succeed())

	// simulate tidb-operator sync config to CM
	actualTikvCM := &corev1.ConfigMap{}
	Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.tikvCM), actualTikvCM)).To(Succeed())
	tikvDataBytes, _ := objects.cr2.Spec.TiKV.Config.MarshalTOML()
	actualTikvCM.Data["config-file"] = string(tikvDataBytes)
	Expect(remoteCli.Update(ctx, actualTikvCM)).To(Succeed())

	// sync config for pd
	actualPDCM := &corev1.ConfigMap{}
	Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.pdCM), actualPDCM)).To(Succeed())
	pdDataBytes, _ := objects.cr2.Spec.PD.Config.MarshalTOML()
	actualPDCM.Data["config-file"] = string(pdDataBytes)
	Expect(remoteCli.Update(ctx, actualPDCM)).To(Succeed())

	// sync config for tiproxy
	if objects.cr2.Spec.TiProxy != nil && objects.cr2.Spec.TiProxy.Config != nil {
		actualTiProxyCM := &corev1.ConfigMap{}
		Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.tiproxyCM), actualTiProxyCM)).To(Succeed())
		tiproxyDataBytes, _ := objects.cr2.Spec.TiProxy.Config.MarshalTOML()
		actualTiProxyCM.Data["config-file"] = string(tiproxyDataBytes)
		Expect(remoteCli.Update(ctx, actualTiProxyCM)).To(Succeed())
	}

	// simulate tidb-operator sync annotation from TC to sts
	actualPDSTS := &asappsv1.StatefulSet{}
	Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.pdSTS), actualPDSTS)).To(Succeed())
	actualPDSTS.Spec.Template.Annotations[configchange.RestartByConfigChangeAnnotation] = actualTC.Spec.PD.Annotations[configchange.RestartByConfigChangeAnnotation]
	Expect(remoteCli.Update(ctx, actualPDSTS)).To(Succeed())

	// tidb
	actualTidbSTS := &asappsv1.StatefulSet{}
	Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.tidbSTS), actualTidbSTS)).To(Succeed())
	actualTidbSTS.Spec.Template.Annotations[configchange.RestartByConfigChangeAnnotation] = actualTC.Spec.TiDB.Annotations[configchange.RestartByConfigChangeAnnotation]
	Expect(remoteCli.Update(ctx, actualTidbSTS)).To(Succeed())

	// simulate tidb-operator sync annotation from TC to sts for tidb extra group
	if len(actualCluster.Status.ExtraEndpoints) != 0 {
		for _, extraGroup := range actualCluster.Status.ExtraEndpoints {
			extraGroupNamer := namer.NewNamer().
				WithTiDBGroupName(extraGroup.Name).
				WithComponent(string(cloudv1.ComponentTiDB))
			extraGroupTCName := extraGroupNamer.GetTiDBName()
			extraGroupTC := &operatorv1alpha1.TidbCluster{}
			Expect(remoteCli.Get(ctx, types.NamespacedName{Namespace: actualCluster.PrefixedId(), Name: extraGroupTCName}, extraGroupTC)).To(Succeed())

			actualSTS := &asappsv1.StatefulSet{}
			stsName := fmt.Sprintf("%s-%s", extraGroupTCName, cloudv1.ComponentTiDB)
			Expect(remoteCli.Get(ctx, types.NamespacedName{Namespace: actualCluster.PrefixedId(), Name: stsName}, actualSTS)).To(Succeed())
			actualSTS.Spec.Template.Annotations[configchange.RestartByConfigChangeAnnotation] = extraGroupTC.Spec.TiDB.Annotations[configchange.RestartByConfigChangeAnnotation]
			Expect(remoteCli.Update(ctx, actualSTS)).To(Succeed())
		}
	}

	actualTikvSTS := &asappsv1.StatefulSet{}
	Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.tikvSTS), actualTikvSTS)).To(Succeed())
	actualTikvSTS.Spec.Template.Annotations[configchange.RestartByConfigChangeAnnotation] = cc.Name
	Expect(remoteCli.Update(ctx, actualTikvSTS)).To(Succeed())

	actualTiflashSTS := &asappsv1.StatefulSet{}
	Expect(remoteCli.Get(ctx, client.ObjectKeyFromObject(objects.tiflashSTS), actualTiflashSTS)).To(Succeed())
	actualTiflashSTS.Spec.Template.Annotations[configchange.RestartByConfigChangeAnnotation] = cc.Name
	Expect(remoteCli.Update(ctx, actualTiflashSTS)).To(Succeed())
}

func newConfigChange(cluster *cloudv1.Cluster, configRevisionName string, strategy cloudv1.ConfigChangeStrategy) *cloudv1.ConfigChange {
	cc := &cloudv1.ConfigChange{
		ObjectMeta: metav1.ObjectMeta{
			Name:      fmt.Sprintf("configchange-%d", rand.Int()),
			Namespace: cluster.Namespace,
			Labels: map[string]string{
				cloudv1.ReconcilerLabelKey: "none",
			},
			Annotations: map[string]string{},
		},
		Spec: cloudv1.ConfigChangeSpec{
			ClusterName:        cluster.Name,
			ConfigRevisionName: configRevisionName,
			ChangeStrategy:     strategy,
			Paused:             nil,
		},
	}
	return cc
}

func newReconcilableConfigChange(cluster *cloudv1.Cluster, configRevisionName string, strategy cloudv1.ConfigChangeStrategy) *cloudv1.ConfigChange {
	cc := newConfigChange(cluster, configRevisionName, strategy)
	cc.Labels[cloudv1.ReconcilerLabelKey] = configChangeReconcilerLabel
	return cc
}

func createRunningClusterHaving(ctx context.Context, network *cloudv1.Network, tenantID, clusterID, instanceType string) *cloudv1.Cluster {
	cluster := newLegacyCluster(network, tenantID, clusterID, instanceType)
	Expect(k8sClient.Create(ctx, cluster)).To(Succeed())
	cluster.Status.Phase = cloudv1.ClusterPhaseRunning
	Expect(k8sClient.Status().Update(ctx, cluster)).To(Succeed())

	return cluster
}

func newTC(cluster *cloudv1.Cluster) *operatorv1alpha1.TidbCluster {
	return &operatorv1alpha1.TidbCluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:        "db",
			Namespace:   fmt.Sprintf("tidb%s", cluster.Name),
			Annotations: map[string]string{"placeholder": "dummy"},
			Labels:      map[string]string{"placeholder": "dummy"},
		},
		Spec: operatorv1alpha1.TidbClusterSpec{
			PD:      &operatorv1alpha1.PDSpec{Replicas: int32(cluster.Spec.Cluster.PD.Replicas)},
			TiDB:    &operatorv1alpha1.TiDBSpec{Replicas: int32(cluster.Spec.Cluster.TiDB.Replicas)},
			TiKV:    &operatorv1alpha1.TiKVSpec{Replicas: int32(cluster.Spec.Cluster.TiKV.Replicas)},
			TiFlash: &operatorv1alpha1.TiFlashSpec{Replicas: int32(cluster.Spec.Cluster.TiFlash.Replicas)},
		},
		Status: operatorv1alpha1.TidbClusterStatus{
			PD:      operatorv1alpha1.PDStatus{Phase: operatorv1alpha1.NormalPhase},
			TiDB:    operatorv1alpha1.TiDBStatus{Phase: operatorv1alpha1.NormalPhase},
			TiKV:    operatorv1alpha1.TiKVStatus{Phase: operatorv1alpha1.NormalPhase},
			TiFlash: operatorv1alpha1.TiFlashStatus{Phase: operatorv1alpha1.NormalPhase},
			TiProxy: operatorv1alpha1.TiProxyStatus{Phase: operatorv1alpha1.NormalPhase},
			Conditions: []operatorv1alpha1.TidbClusterCondition{
				{
					Type:   operatorv1alpha1.TidbClusterReady,
					Status: corev1.ConditionTrue,
				},
			},
		},
	}
}

func newConfigMap(cluster *cloudv1.Cluster, componentType cloudv1.ComponentType, data map[string]string) *corev1.ConfigMap {
	return &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      fmt.Sprintf("db-%s", componentType),
			Namespace: fmt.Sprintf("tidb%s", cluster.Name),
		},
		Data: data,
	}
}

func newSTS(cluster *cloudv1.Cluster, componentType cloudv1.ComponentType, volumes []corev1.Volume) *asappsv1.StatefulSet {
	return &asappsv1.StatefulSet{
		TypeMeta: metav1.TypeMeta{
			Kind:       "StatefulSet",
			APIVersion: "apps.pingcap.com/v1",
		},
		ObjectMeta: metav1.ObjectMeta{
			Name:        fmt.Sprintf("db-%s", componentType),
			Namespace:   fmt.Sprintf("tidb%s", cluster.Name),
			Annotations: map[string]string{"placeholder": "dummy"},
			Labels:      map[string]string{"placeholder": "dummy"},
		},
		Spec: asappsv1.StatefulSetSpec{
			Replicas: nil,
			Selector: nil,
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Annotations: map[string]string{"placeholder": "dummy"},
					Labels:      map[string]string{"placeholder": "dummy"},
				},
				Spec: corev1.PodSpec{
					Volumes: volumes,
				},
			},
			VolumeClaimTemplates: nil,
			ServiceName:          "",
			PodManagementPolicy:  "",
			RevisionHistoryLimit: nil,
		},
	}
}