Repository: kubernetes-sigs/external-dns

Issue:
ci: add check on test coverage regression
<!-- Please only use this template for submitting enhancement requests. This can be something like a new provider or a new gateway.  -->

**What would you like to be added**:

Add a step in the GitHub Actions that checks the code coverage both before and after the PR. Ensure the new percent code coverage for each file is equal to or higher than the previous one.

**Why is this needed**:

Enforce test-driven development best practices.

Hints from issue comments:
@ivankatliarchuk: I tried. Kubernetes org is quite burecratic https://github.com/kubernetes/org/issues/5512@TobyTheHutt: I read the referenced issue. I wrote a quick abstract here. It's detailed because I thought it might be helpful for future reference. Maybe it also brings inspiration for a new approach towards solving this ticket.

For the impatient: Just read "Further Steps" at the end.

**Need**  
A code-coverage reporting for the external-dns repo (and potentially others, too) to improve on test quality, early regression detection and to gain transparent metrics for contributors, users and gates.

**Requirements**  
|#|Title|Description|
|---|---|---|
|R01|Ownership|An app like Coveralls must be community owned and governed (no integrations under personal account governance).|
|R02|Audit|Ownership of the solution must be auditable.|
|R03|Per-Repo Permission Scopes|Scope of the solution to repos must be explicitly permitted for each appliance.|

**Constraints**
|#|Title|Description|
|---|---|---|
|C01|Org Access & CI Policy|Settings and access within the Kubernetes org are declaratively managed in `kubernetes/org`, reconciled by Peribolos and logged in the repo PR history.|
|C02|Change Restrictions|Changes to the org are proposed through a change PR to `kubernetes/org` and needs to be approved by an owner. If merged, Peribolos runs and applies the state.|
|C03|CI Governance|All CI automation jobs are defined as code in `kubernetes/test-infra` and Prow executes them as K8s pods via a `ProwJob` resource. To add or change a job or gate, a PR to `kubernetes/test-infra` adjusting job YAML must be reviewed and merged, so it gets deployed by Prow.|

**Solutions**  
|#|Title|Description|
|---|---|---|
|S01|Prow-Native|Utilising jobs under `kubernetes/test-infra` to publish the coverage to dashboards or PR checks would remove the need for a new third-party app.|
|S02|Org-Installed App|An org owner installs an App like Coveralls at the organisation level and restricts it to target specific repos (service account and access will be recorded in Peribolos).|
|S03|Self-Hosted Coverage|We operate our own coverage service under CNCF/K8s control (which of course leads to high overhead and is overall atypical)|

**Conclusion**  
A quick benchmark of possible solutions makes it clear that S02 using Coveralls would require an org-owned and Peribolos documented exception.  

S01 on the other hand does not conflict with any apparent rule or good practice within the organisation. In fact, the `kubernetes/kubernetes` repo already uses Prow-native coverage (e.g. the `ci-kubernetes-coverage-unit` job). Other repos also use the `ci-kubernetes-coverage-unit` and `kubernetes-coverage-e2e-gci-gce` Prow-jobs, as defined in the `kubernetes/test-infra` repo.

S03 is out of the question for several reasons, which is likely clear to anyone reading this. I only added it for completeness and reference.

**Further steps**  
We might start to adopt the Prow-native checks (S01) as a safe and accepted baseline for our CI jobs. For this, we can open a PR in `kubernetes/test-infra` and add the wanted coverage jobs and PR check for externalDNS.  

Further, if we still want Coveralls (S02), we will need to convince an org owner of its benefits for as, and install it scoped to our repo. This will require a new record for ownership, scope and rotation in `kubernetes/org`.  @ivankatliarchuk: The AI generated report misses the point and please make sure to validate output. We currently have Prow enabled. But is nowhere near industry standards, available dashboards here https://github.com/kubernetes/test-infra. So prow does not support it.

Another option is to find something that does not require any 3rd party services or write own. This may get an approval from owners.
@ivankatliarchuk: Related PR https://github.com/kubernetes-sigs/external-dns/pull/5218@AndrewCharlesHay: I feel like I hit on something @ivankatliarchuk feels very strongly about 😅 @TobyTheHutt: > The AI generated report misses the point and please make sure to validate output. We currently have Prow enabled. But is nowhere near industry standards, available dashboards here https://github.com/kubernetes/test-infra. So prow does not support it.

This gnawed on me for a few days, tbh. 😅 No need to reply to this post - or even credit it, if you don't feel like it. Snowflake that I am, I just had to take position on that, so I can forget about it.

To varying extents, AI is present in many things I do.  However, AI did not write my last post. The structure and content of chapters and tables is mine and it's how I work. Writing summaries like that belongs among the tools I regularly use and value. 

If the post was unwelcome, or even annoying to some, I apologize. Apparently, my conclusions went into the wrong direction and I value the feedback regarding this.@ivankatliarchuk: My apologies, I didn't intend to cause offense or upset you. Your post came across as something an AI might have generated mainly due to it structure, but I was mistaken.

Solution patch:
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 9d8e391a..91a6e613 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -11,12 +11,12 @@ permissions:
 
 jobs:
 
-  build:
+  test:
     permissions:
       contents: read  #  to fetch code (actions/checkout)
       checks: write  #  to create a new check based on the results (shogo82148/actions-goveralls)
 
-    name: Build
+    name: Test
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
@@ -38,21 +38,28 @@ jobs:
       run: |
         go get -v -t -d ./...
 
-    - name: Install additional CI for nektos/act
-      run: |
-            apt update
-            apt install -y make gcc libc-dev git
-      if: github.actor == 'nektos/act' && matrix.os == 'ubuntu-latest'
-
     - name: Test
       env:
-        GOMAXPROCS: 1
-        GOMEMLIMIT: 2048MiB
-      run: make test
+        GOMAXPROCS: 4
+        GOMEMLIMIT: 8192MiB
+      run: make go-test
 
     - name: Send coverage
-      uses: shogo82148/actions-goveralls@v1
+      uses: coverallsapp/github-action@v2
       with:
-        path-to-profile: profile.cov
-      if: github.actor != 'nektos/act' && matrix.os == 'ubuntu-latest'
+        file: profile.cov
+        format: golang
+        flag-name: run-${{ join(matrix.*, '-') }}
+        parallel: true
       continue-on-error: true
+
+  finish:
+    needs: test
+    if: ${{ always() }}
+    runs-on: ubuntu-latest
+    steps:
+    - name: Coveralls Finished
+      uses: coverallsapp/github-action@v2
+      with:
+        parallel-finished: true
+        carryforward: "run-ubuntu-latest,run-macos-latest"
diff --git a/.github/workflows/dependency-update.yaml b/.github/workflows/dependency-update.yaml
index 33d0cd26..8fb749a2 100644
--- a/.github/workflows/dependency-update.yaml
+++ b/.github/workflows/dependency-update.yaml
@@ -17,7 +17,7 @@ jobs:
         uses: actions/checkout@v5.0.0
       # https://github.com/renovatebot/github-action
       - name: self-hosted renovate
-        uses: renovatebot/github-action@v43.0.12
+        uses: renovatebot/github-action@v43.0.14
         with:
           # https://docs.github.com/en/actions/security-for-github-actions/security-guides/automatic-token-authentication
           token: ${{ secrets.GITHUB_TOKEN }}
diff --git a/Makefile b/Makefile
index ee8605fe..268d2ba6 100644
--- a/Makefile
+++ b/Makefile
@@ -74,7 +74,14 @@ crd: controller-gen-install
 #? test: The verify target runs tasks similar to the CI tasks, but without code coverage
 .PHONY: test
 test:
+	go test -race ./...
+
+
+.PHONY: test
+go-test:
 	go test -race -coverprofile=profile.cov ./...
+	go tool cover -func=profile.cov > coverage.summary
+	@tail -n 1 coverage.summary
 
 #? build: The build targets allow to build the binary and container image
 .PHONY: build
diff --git a/README.md b/README.md
index 4997156d..8eb29b4c 100644
--- a/README.md
+++ b/README.md
@@ -115,6 +115,7 @@ from the usage of any externally developed webhook.
 | STACKIT               | https://github.com/stackitcloud/external-dns-stackit-webhook         |
 | Unbound               | https://github.com/guillomep/external-dns-unbound-webhook            |
 | Unifi                 | https://github.com/kashalls/external-dns-unifi-webhook               |
+| Volcengine Cloud      | https://github.com/volcengine/external-dns-volcengine-webhook        |
 | Vultr                 | https://github.com/vultr/external-dns-vultr-webhook                  |
 | Yandex Cloud          | https://github.com/ismailbaskin/external-dns-yandex-webhook/         |
 
diff --git a/controller/controller.go b/controller/controller.go
index bdbddfb6..f0a3fbde 100644
--- a/controller/controller.go
+++ b/controller/controller.go
@@ -182,6 +182,8 @@ type Controller struct {
 	ExcludeRecordTypes []string
 	// MinEventSyncInterval is used as a window for batching events
 	MinEventSyncInterval time.Duration
+	// Old txt-owner value we need to migrate from
+	TXTOwnerOld string
 }
 
 // RunOnce runs a single iteration of a reconciliation loop.
@@ -236,6 +238,7 @@ func (c *Controller) RunOnce(ctx context.Context) error {
 		ManagedRecords: c.ManagedRecordTypes,
 		ExcludeRecords: c.ExcludeRecordTypes,
 		OwnerID:        c.Registry.OwnerID(),
+		OldOwnerId:     c.TXTOwnerOld,
 	}
 
 	plan = plan.Calculate()
diff --git a/controller/execute.go b/controller/execute.go
index 5782db8d..d2140c8d 100644
--- a/controller/execute.go
+++ b/controller/execute.go
@@ -382,6 +382,7 @@ func buildController(
 		ManagedRecordTypes:   cfg.ManagedDNSRecordTypes,
 		ExcludeRecordTypes:   cfg.ExcludeDNSRecordTypes,
 		MinEventSyncInterval: cfg.MinEventSyncInterval,
+		TXTOwnerOld:          cfg.TXTOwnerOld,
 		EventEmitter:         eventEmitter,
 	}, nil
 }
@@ -418,7 +419,7 @@ func selectRegistry(cfg *externaldns.Config, p provider.Provider) (registry.Regi
 	case "noop":
 		r, err = registry.NewNoopRegistry(p)
 	case "txt":
-		r, err = registry.NewTXTRegistry(p, cfg.TXTPrefix, cfg.TXTSuffix, cfg.TXTOwnerID, cfg.TXTCacheInterval, cfg.TXTWildcardReplacement, cfg.ManagedDNSRecordTypes, cfg.ExcludeDNSRecordTypes, cfg.TXTEncryptEnabled, []byte(cfg.TXTEncryptAESKey))
+		r, err = registry.NewTXTRegistry(p, cfg.TXTPrefix, cfg.TXTSuffix, cfg.TXTOwnerID, cfg.TXTCacheInterval, cfg.TXTWildcardReplacement, cfg.ManagedDNSRecordTypes, cfg.ExcludeDNSRecordTypes, cfg.TXTEncryptEnabled, []byte(cfg.TXTEncryptAESKey), cfg.TXTOwnerOld)
 	case "aws-sd":
 		r, err = registry.NewAWSSDRegistry(p, cfg.TXTOwnerID)
 	default:
diff --git a/docs/flags.md b/docs/flags.md
index 61ce8934..f03026c9 100644
--- a/docs/flags.md
+++ b/docs/flags.md
@@ -164,6 +164,7 @@
 | `--txt-wildcard-replacement=""` | When using the TXT registry, a custom string that's used instead of an asterisk for TXT records corresponding to wildcard DNS records (optional) |
 | `--[no-]txt-encrypt-enabled` | When using the TXT registry, set if TXT records should be encrypted before stored (default: disabled) |
 | `--txt-encrypt-aes-key=""` | When using the TXT registry, set TXT record decryption and encryption 32 byte aes key (required when --txt-encrypt=true) |
+| `--migrate-from-txt-owner=""` | Old txt-owner-id that needs to be overwritten (default: default) |
 | `--dynamodb-region=""` | When using the DynamoDB registry, the AWS region of the DynamoDB table (optional) |
 | `--dynamodb-table="external-dns"` | When using the DynamoDB registry, the name of the DynamoDB table (default: "external-dns") |
 | `--txt-cache-interval=0s` | The interval between cache synchronizations in duration format (default: disabled) |
diff --git a/docs/registry/txt.md b/docs/registry/txt.md
index 5c2b9857..46ca34cb 100644
--- a/docs/registry/txt.md
+++ b/docs/registry/txt.md
@@ -205,3 +205,91 @@ The TXT registry can optionally cache DNS records read from the provider. This c
 rate limits imposed by the provider.
 
 Caching is enabled by specifying a cache duration with the `--txt-cache-interval` flag.
+
+## OwnerID migration
+
+The owner ID of the TXT records managed by external-dns instance can be updated.
+
+When `--migrate-from-txt-owner` is set, it will enable the migration checks
+in the run loop using `--txt-owner-id=new-owner-id` and the value you defined for this flag.
+
+If you want to test the outputs of a migration beforehand, you can use the `--dry-run` flag
+along with `--migrate-from-txt-owner`.
+
+Example, if you had a standard deployment like so:
+
+```yaml
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: external-dns
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: external-dns
+  strategy:
+    type: Recreate
+  template:
+    metadata:
+      labels:
+        app: external-dns
+    spec:
+      serviceAccountName: external-dns
+      containers:
+      - name: external-dns
+        image: registry.k8s.io/external-dns/external-dns:v0.19.0
+        imagePullPolicy: Always
+        args:
+        - "--txt-prefix=%{record_type}-"
+        - "--txt-cache-interval=2m"
+        - "--log-level=debug"
+        - "--log-format=text"
+        - "--txt-owner-id=old-owner"
+        - "--policy=sync"
+        - "--provider=some-provider"
+        - "--registry=txt"
+        - "--interval=1m"
+        - "--source=ingress"
+```
+
+You can update your deployment to migrate like so :
+
+```yaml
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: external-dns
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: external-dns
+  strategy:
+    type: Recreate
+  template:
+    metadata:
+      labels:
+        app: external-dns
+    spec:
+      serviceAccountName: external-dns
+      containers:
+      - name: external-dns
+        imagePullPolicy: Always
+        image: registry.k8s.io/external-dns/external-dns:v0.19.0
+        args:
+        - "--txt-prefix=%{record_type}-"
+        - "--txt-cache-interval=2m"
+        - "--log-level=debug"
+        - "--log-format=text"
+        - "--txt-owner-id=new-owner"
+        - "--migrate-from-txt-owner=old-owner"
+        - "--policy=sync"
+        - "--provider=some-provider"
+        - "--registry=txt"
+        - "--interval=1m"
+        - "--source=ingress"
+```
+
+If you didn't set the owner ID, the value set by external-dns is `default`. You can set the
+`--migrate-from-txt-owner` flag to `default` to migrate the associated records.
diff --git a/docs/tutorials/cloudflare.md b/docs/tutorials/cloudflare.md
index d5b5764f..45ac5080 100644
--- a/docs/tutorials/cloudflare.md
+++ b/docs/tutorials/cloudflare.md
@@ -386,6 +386,20 @@ Requires [Cloudflare for SaaS](https://developers.cloudflare.com/cloudflare-for-
 
 **Note:** Due to using the legacy cloudflare-go v0 API for custom hostname management, the custom hostname page size is fixed at 50. This limitation will be addressed in a future migration to the v4 SDK.
 
+## Setting Cloudflare DNS Record Tags
+
+Cloudflare allows you to add descriptive tags to DNS records. This can be useful for organizing your records.
+For example one can apply tags by environment (`production`, `staging`) or by the team that owns them (`frontend-team`, `backend-team`). ExternalDNS can manage these tags for you.
+
+To assign tags to a DNS record, add the `external-dns.alpha.kubernetes.io/cloudflare-tags` annotation to your Kubernetes resource (like a Service or Ingress). The value should be a comma-separated list of your desired tags.
+
+```yaml
+metadata:
+  annotations:
+    # Assigns three tags to the DNS record created from this resource
+    external-dns.alpha.kubernetes.io/cloudflare-tags: "owner:frontend-team, env:dev, component:api"
+```
+
 ## Using CRD source to manage DNS records in Cloudflare
 
 Please refer to the [CRD source documentation](../sources/crd.md#example) for more information.
diff --git a/pkg/apis/externaldns/types.go b/pkg/apis/externaldns/types.go
index d9f8e6ef..c2fff68b 100644
--- a/pkg/apis/externaldns/types.go
+++ b/pkg/apis/externaldns/types.go
@@ -142,6 +142,7 @@ type Config struct {
 	Policy                                        string
 	Registry                                      string
 	TXTOwnerID                                    string
+	TXTOwnerOld                                   string
 	TXTPrefix                                     string
 	TXTSuffix                                     string
 	TXTEncryptEnabled                             bool
@@ -371,6 +372,7 @@ var defaultConfig = &Config{
 	TXTEncryptAESKey:             "",
 	TXTEncryptEnabled:            false,
 	TXTOwnerID:                   "default",
+	TXTOwnerOld:                  "",
 	TXTPrefix:                    "",
 	TXTSuffix:                    "",
 	TXTWildcardReplacement:       "",
@@ -782,6 +784,7 @@ func bindFlags(b FlagBinder, cfg *Config) {
 	b.StringVar("txt-wildcard-replacement", "When using the TXT registry, a custom string that's used instead of an asterisk for TXT records corresponding to wildcard DNS records (optional)", defaultConfig.TXTWildcardReplacement, &cfg.TXTWildcardReplacement)
 	b.BoolVar("txt-encrypt-enabled", "When using the TXT registry, set if TXT records should be encrypted before stored (default: disabled)", defaultConfig.TXTEncryptEnabled, &cfg.TXTEncryptEnabled)
 	b.StringVar("txt-encrypt-aes-key", "When using the TXT registry, set TXT record decryption and encryption 32 byte aes key (required when --txt-encrypt=true)", defaultConfig.TXTEncryptAESKey, &cfg.TXTEncryptAESKey)
+	b.StringVar("migrate-from-txt-owner", "Old txt-owner-id that needs to be overwritten (default: default)", defaultConfig.TXTOwnerOld, &cfg.TXTOwnerOld)
 	b.StringVar("dynamodb-region", "When using the DynamoDB registry, the AWS region of the DynamoDB table (optional)", cfg.AWSDynamoDBRegion, &cfg.AWSDynamoDBRegion)
 	b.StringVar("dynamodb-table", "When using the DynamoDB registry, the name of the DynamoDB table (default: \"external-dns\")", defaultConfig.AWSDynamoDBTable, &cfg.AWSDynamoDBTable)
 
diff --git a/plan/plan.go b/plan/plan.go
index 29962377..325a3d80 100644
--- a/plan/plan.go
+++ b/plan/plan.go
@@ -52,6 +52,8 @@ type Plan struct {
 	ExcludeRecords []string
 	// OwnerID of records to manage
 	OwnerID string
+	// Old owner ID we migrate from
+	OldOwnerId string
 }
 
 // Changes holds lists of actions to be executed by dns providers
@@ -224,7 +226,8 @@ func (p *Plan) Calculate() *Plan {
 				if records.current != nil && len(records.candidates) > 0 {
 					update := t.resolver.ResolveUpdate(records.current, records.candidates)
 
-					if shouldUpdateTTL(update, records.current) || targetChanged(update, records.current) || p.shouldUpdateProviderSpecific(update, records.current) {
+					if shouldUpdateTTL(update, records.current) || targetChanged(update, records.current) || p.shouldUpdateProviderSpecific(update, records.current) ||
+						p.isOldOwnerIdSetAndDifferent(records.current) {
 						inheritOwner(records.current, update)
 						changes.UpdateNew = append(changes.UpdateNew, update)
 						changes.UpdateOld = append(changes.UpdateOld, records.current)
@@ -276,6 +279,10 @@ func (p *Plan) Calculate() *Plan {
 	return plan
 }
 
+func (p *Plan) isOldOwnerIdSetAndDifferent(current *endpoint.Endpoint) bool {
+	return len(p.OldOwnerId) != 0 && current.Labels[endpoint.OwnerLabelKey] != p.OldOwnerId
+}
+
 func inheritOwner(from, to *endpoint.Endpoint) {
 	if to.Labels == nil {
 		to.Labels = map[string]string{}
diff --git a/provider/cloudflare/cloudflare.go b/provider/cloudflare/cloudflare.go
index e662a0fa..c852962d 100644
--- a/provider/cloudflare/cloudflare.go
+++ b/provider/cloudflare/cloudflare.go
@@ -279,6 +279,7 @@ func getUpdateDNSRecordParam(zoneID string, cfc cloudFlareChange) dns.RecordUpda
 			Content:  cloudflare.F(cfc.ResourceRecord.Content),
 			Priority: cloudflare.F(cfc.ResourceRecord.Priority),
 			Comment:  cloudflare.F(cfc.ResourceRecord.Comment),
+			Tags:     cloudflare.F(cfc.ResourceRecord.Tags),
 		},
 	}
 }
@@ -295,6 +296,7 @@ func getCreateDNSRecordParam(zoneID string, cfc *cloudFlareChange) dns.RecordNew
 			Content:  cloudflare.F(cfc.ResourceRecord.Content),
 			Priority: cloudflare.F(cfc.ResourceRecord.Priority),
 			Comment:  cloudflare.F(cfc.ResourceRecord.Comment),
+			Tags:     cloudflare.F(cfc.ResourceRecord.Tags),
 		},
 	}
 }
@@ -720,6 +722,21 @@ func (p *CloudFlareProvider) submitChanges(ctx context.Context, changes []*cloud
 	return nil
 }
 
+// parseTagsAnnotation is the single helper method to handle tags from the annotation string.
+// It splits the string, cleans up whitespace, and sorts the tags to create a canonical representation.
+func parseTagsAnnotation(tagString string) []string {
+	tags := strings.Split(tagString, ",")
+	cleanedTags := make([]string, 0, len(tags))
+	for _, tag := range tags {
+		trimmed := strings.TrimSpace(tag)
+		if trimmed != "" {
+			cleanedTags = append(cleanedTags, trimmed)
+		}
+	}
+	sort.Strings(cleanedTags)
+	return cleanedTags
+}
+
 // AdjustEndpoints modifies the endpoints as needed by the specific provider
 func (p *CloudFlareProvider) AdjustEndpoints(endpoints []*endpoint.Endpoint) ([]*endpoint.Endpoint, error) {
 	var adjustedEndpoints []*endpoint.Endpoint
@@ -741,6 +758,11 @@ func (p *CloudFlareProvider) AdjustEndpoints(endpoints []*endpoint.Endpoint) ([]
 			e.DeleteProviderSpecificProperty(annotations.CloudflareCustomHostnameKey)
 		}
 
+		if val, ok := e.GetProviderSpecificProperty(annotations.CloudflareTagsKey); ok {
+			sortedTags := parseTagsAnnotation(val)
+			e.SetProviderSpecificProperty(annotations.CloudflareTagsKey, strings.Join(sortedTags, ","))
+		}
+
 		p.adjustEndpointProviderSpecificRegionKeyProperty(e)
 
 		if p.DNSRecordsConfig.Comment != "" {
@@ -827,6 +849,11 @@ func (p *CloudFlareProvider) newCloudFlareChange(action changeAction, ep *endpoi
 		comment = val
 	}
 
+	var tags []string
+	if val, ok := ep.GetProviderSpecificProperty(annotations.CloudflareTagsKey); ok {
+		tags = parseTagsAnnotation(val)
+	}
+
 	if len(comment) > freeZoneMaxCommentLength {
 		comment = p.DNSRecordsConfig.trimAndValidateComment(ep.DNSName, comment, p.ZoneHasPaidPlan)
 	}
@@ -851,6 +878,7 @@ func (p *CloudFlareProvider) newCloudFlareChange(action changeAction, ep *endpoi
 			Type:     dns.RecordResponseType(ep.RecordType),
 			Content:  target,
 			Comment:  comment,
+			Tags:     tags,
 			Priority: priority,
 		},
 		RegionalHostname:    p.regionalHostname(ep),
@@ -1016,6 +1044,13 @@ func (p *CloudFlareProvider) groupByNameAndTypeWithCustomHostnames(records DNSRe
 			e = e.WithProviderSpecific(annotations.CloudflareRecordCommentKey, records[0].Comment)
 		}
 
+		if records[0].Tags != nil {
+			if tags, ok := records[0].Tags.([]string); ok && len(tags) > 0 {
+				sort.Strings(tags)
+				e = e.WithProviderSpecific(annotations.CloudflareTagsKey, strings.Join(tags, ","))
+			}
+		}
+
 		endpoints = append(endpoints, e)
 	}
 	return endpoints
diff --git a/registry/txt.go b/registry/txt.go
index b428cfe9..d8141921 100644
--- a/registry/txt.go
+++ b/registry/txt.go
@@ -60,6 +60,9 @@ type TXTRegistry struct {
 	txtEncryptEnabled bool
 	txtEncryptAESKey  []byte
 
+	//Handle Owner ID migration
+	oldOwnerID string
+
 	// existingTXTs is the TXT records that already exist in the zone so that
 	// ApplyChanges() can skip re-creating them. See the struct below for details.
 	existingTXTs *existingTXTs
@@ -114,7 +117,8 @@ func (im *existingTXTs) reset() {
 func NewTXTRegistry(provider provider.Provider, txtPrefix, txtSuffix, ownerID string,
 	cacheInterval time.Duration, txtWildcardReplacement string,
 	managedRecordTypes, excludeRecordTypes []string,
-	txtEncryptEnabled bool, txtEncryptAESKey []byte) (*TXTRegistry, error) {
+	txtEncryptEnabled bool, txtEncryptAESKey []byte,
+	oldOwnerID string) (*TXTRegistry, error) {
 	if ownerID == "" {
 		return nil, errors.New("owner id cannot be empty")
 	}
@@ -148,6 +152,7 @@ func NewTXTRegistry(provider provider.Provider, txtPrefix, txtSuffix, ownerID st
 		excludeRecordTypes:  excludeRecordTypes,
 		txtEncryptEnabled:   txtEncryptEnabled,
 		txtEncryptAESKey:    txtEncryptAESKey,
+		oldOwnerID:          oldOwnerID,
 		existingTXTs:        newExistingTXTs(),
 	}, nil
 }
@@ -252,6 +257,10 @@ func (im *TXTRegistry) Records(ctx context.Context) ([]*endpoint.Endpoint, error
 			}
 		}
 
+		if im.oldOwnerID != "" && ep.Labels[endpoint.OwnerLabelKey] == im.oldOwnerID {
+			ep.Labels[endpoint.OwnerLabelKey] = im.ownerID
+		}
+
 		// Handle the migration of TXT records created before the new format (introduced in v0.12.0).
 		// The migration is done for the TXT records owned by this instance only.
 		if len(txtRecordsMap) > 0 && ep.Labels[endpoint.OwnerLabelKey] == im.ownerID {
@@ -292,6 +301,11 @@ func (im *TXTRegistry) generateTXTRecordWithFilter(r *endpoint.Endpoint, filter
 	if isAlias, found := r.GetProviderSpecificProperty("alias"); found && isAlias == "true" && recordType == endpoint.RecordTypeA {
 		recordType = endpoint.RecordTypeCNAME
 	}
+
+	if im.oldOwnerID != "" && r.Labels[endpoint.OwnerLabelKey] == im.oldOwnerID {
+		r.Labels[endpoint.OwnerLabelKey] = im.ownerID
+	}
+
 	txtNew := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName, recordType), endpoint.RecordTypeTXT, r.Labels.Serialize(true, im.txtEncryptEnabled, im.txtEncryptAESKey))
 	if txtNew != nil {
 		txtNew.WithSetIdentifier(r.SetIdentifier)
@@ -315,6 +329,7 @@ func (im *TXTRegistry) ApplyChanges(ctx context.Context, changes *plan.Changes)
 		UpdateOld: endpoint.FilterEndpointsByOwnerID(im.ownerID, changes.UpdateOld),
 		Delete:    endpoint.FilterEndpointsByOwnerID(im.ownerID, changes.Delete),
 	}
+
 	for _, r := range filteredChanges.Create {
 		if r.Labels == nil {
 			r.Labels = make(map[string]string)
diff --git a/source/annotations/annotations.go b/source/annotations/annotations.go
index ba7695fd..87d60241 100644
--- a/source/annotations/annotations.go
+++ b/source/annotations/annotations.go
@@ -27,6 +27,7 @@ const (
 	CloudflareCustomHostnameKey = AnnotationKeyPrefix + "cloudflare-custom-hostname"
 	CloudflareRegionKey         = AnnotationKeyPrefix + "cloudflare-region-key"
 	CloudflareRecordCommentKey  = AnnotationKeyPrefix + "cloudflare-record-comment"
+	CloudflareTagsKey           = AnnotationKeyPrefix + "cloudflare-tags"
 
 	AWSPrefix        = AnnotationKeyPrefix + "aws-"
 	CoreDNSPrefix    = AnnotationKeyPrefix + "coredns-"
diff --git a/source/annotations/provider_specific.go b/source/annotations/provider_specific.go
index 0518effe..31f059a5 100644
--- a/source/annotations/provider_specific.go
+++ b/source/annotations/provider_specific.go
@@ -75,6 +75,11 @@ func ProviderSpecificAnnotations(annotations map[string]string) (endpoint.Provid
 					Name:  CloudflareRecordCommentKey,
 					Value: v,
 				})
+			} else if strings.Contains(k, CloudflareTagsKey) {
+				providerSpecificAnnotations = append(providerSpecificAnnotations, endpoint.ProviderSpecificProperty{
+					Name:  CloudflareTagsKey,
+					Value: v,
+				})
 			}
 		}
 	}



Test patch:
diff --git a/pkg/apis/externaldns/types_test.go b/pkg/apis/externaldns/types_test.go
index 9a35cd3e..5a6d7f96 100644
--- a/pkg/apis/externaldns/types_test.go
+++ b/pkg/apis/externaldns/types_test.go
@@ -104,6 +104,7 @@ var (
 		Policy:                                        "sync",
 		Registry:                                      "txt",
 		TXTOwnerID:                                    "default",
+		TXTOwnerOld:                                   "",
 		TXTPrefix:                                     "",
 		TXTCacheInterval:                              0,
 		Interval:                                      time.Minute,
@@ -217,6 +218,7 @@ var (
 		Registry:                                      "noop",
 		TXTOwnerID:                                    "owner-1",
 		TXTPrefix:                                     "associated-txt-record",
+		TXTOwnerOld:                                   "old-owner",
 		TXTCacheInterval:                              12 * time.Hour,
 		Interval:                                      10 * time.Minute,
 		MinEventSyncInterval:                          50 * time.Second,
@@ -360,6 +362,7 @@ func TestParseFlags(t *testing.T) {
 				"--policy=upsert-only",
 				"--registry=noop",
 				"--txt-owner-id=owner-1",
+				"--migrate-from-txt-owner=old-owner",
 				"--txt-prefix=associated-txt-record",
 				"--txt-cache-interval=12h",
 				"--dynamodb-table=custom-table",
@@ -482,6 +485,7 @@ func TestParseFlags(t *testing.T) {
 				"EXTERNAL_DNS_REGISTRY":                                          "noop",
 				"EXTERNAL_DNS_TXT_OWNER_ID":                                      "owner-1",
 				"EXTERNAL_DNS_TXT_PREFIX":                                        "associated-txt-record",
+				"EXTERNAL_DNS_MIGRATE_FROM_TXT_OWNER":                            "old-owner",
 				"EXTERNAL_DNS_TXT_CACHE_INTERVAL":                                "12h",
 				"EXTERNAL_DNS_TXT_NEW_FORMAT_ONLY":                               "1",
 				"EXTERNAL_DNS_INTERVAL":                                          "10m",
diff --git a/plan/plan_test.go b/plan/plan_test.go
index 10ce5a99..acce96b2 100644
--- a/plan/plan_test.go
+++ b/plan/plan_test.go
@@ -1017,6 +1017,32 @@ func (suite *PlanTestSuite) TestDualStackToSingleStack() {
 	validateEntries(suite.T(), changes.UpdateNew, expectNoChanges)
 }
 
+func (suite *PlanTestSuite) TestRecordOwnerIdMigration() {
+	suite.fooA5.Labels[endpoint.OwnerLabelKey] = "bar"
+	current := []*endpoint.Endpoint{suite.fooA5}
+	desired := []*endpoint.Endpoint{suite.fooA5}
+	expectedCreate := []*endpoint.Endpoint{}
+	expectedUpdateOld := []*endpoint.Endpoint{suite.fooA5}
+	expectedUpdateNew := []*endpoint.Endpoint{suite.fooA5}
+	expectedDelete := []*endpoint.Endpoint{}
+
+	p := &Plan{
+		Policies:       []Policy{&SyncPolicy{}},
+		Current:        current,
+		Desired:        desired,
+		ManagedRecords: []string{endpoint.RecordTypeA, endpoint.RecordTypeAAAA, endpoint.RecordTypeCNAME},
+		OwnerID:        suite.fooA5.Labels[endpoint.OwnerLabelKey],
+		OldOwnerId:     "foo",
+	}
+
+	changes := p.Calculate().Changes
+
+	validateEntries(suite.T(), changes.Create, expectedCreate)
+	validateEntries(suite.T(), changes.UpdateNew, expectedUpdateNew)
+	validateEntries(suite.T(), changes.UpdateOld, expectedUpdateOld)
+	validateEntries(suite.T(), changes.Delete, expectedDelete)
+}
+
 func TestPlan(t *testing.T) {
 	suite.Run(t, new(PlanTestSuite))
 }
diff --git a/provider/cloudflare/cloudflare_test.go b/provider/cloudflare/cloudflare_test.go
index 4acba50d..2de2f8e1 100644
--- a/provider/cloudflare/cloudflare_test.go
+++ b/provider/cloudflare/cloudflare_test.go
@@ -89,6 +89,116 @@ var ExampleDomain = []dns.RecordResponse{
 	},
 }
 
+func TestCloudflareProviderTags(t *testing.T) {
+	provider := &CloudFlareProvider{}
+	t.Run("parseTagsAnnotation", func(t *testing.T) {
+		testCases := []struct {
+			name     string
+			input    string
+			expected []string
+		}{
+			{
+				name:     "Correctly sorts and cleans tags",
+				input:    "owner:owner_name, env:dev, app:test ",
+				expected: []string{"app:test", "env:dev", "owner:owner_name"},
+			},
+			{
+				name:     "Handles a single tag",
+				input:    "owner:owner_name",
+				expected: []string{"owner:owner_name"},
+			},
+			{
+				name:     "Handles empty string",
+				input:    "",
+				expected: []string{},
+			},
+			{
+				name:     "Handles messy input with extra commas and spaces",
+				input:    " tag1 ,  tag2,,tag3 ",
+				expected: []string{"tag1", "tag2", "tag3"},
+			},
+		}
+
+		for _, tc := range testCases {
+			t.Run(tc.name, func(t *testing.T) {
+				assert.Equal(t, tc.expected, parseTagsAnnotation(tc.input))
+			})
+		}
+	})
+
+	// Test cases for the groupByNameAndTypeWithCustomHostnames function (handling API response)
+	t.Run("groupByNameAndTypeWithCustomHostnames", func(t *testing.T) {
+		records := DNSRecordsMap{
+			DNSRecordIndex{Name: "test.example.com", Type: "A", Content: "1.2.3.4"}: {
+				Name:    "test.example.com",
+				Type:    "A",
+				Content: "1.2.3.4",
+				Tags:    []string{"owner:owner_name", "env:dev", "app:test"},
+			},
+		}
+
+		endpoints := provider.groupByNameAndTypeWithCustomHostnames(records, nil)
+		require.Len(t, endpoints, 1)
+
+		val, ok := endpoints[0].GetProviderSpecificProperty(annotations.CloudflareTagsKey)
+		assert.True(t, ok)
+		assert.Equal(t, "app:test,env:dev,owner:owner_name", val, "Tags from API should be sorted")
+	})
+
+	// This sub-test verifies that AdjustEndpoints correctly sorts and cleans the tags string.
+	t.Run("AdjustEndpoints sorts tags", func(t *testing.T) {
+		// Arrange: Create an endpoint with unsorted tags, including extra whitespace.
+		endpointWithTags := []*endpoint.Endpoint{
+			{
+				DNSName:    "tags.example.com",
+				Targets:    endpoint.Targets{"1.2.3.4"},
+				RecordType: endpoint.RecordTypeA,
+				ProviderSpecific: endpoint.ProviderSpecific{
+					{
+						Name:  annotations.CloudflareTagsKey,
+						Value: "owner:team-b, env:prod, app:api ", // Unsorted and messy
+					},
+				},
+			},
+		}
+		expectedTagsString := "app:api,env:prod,owner:team-b"
+
+		// Act: Call the function under test.
+		adjustedEndpoints, err := provider.AdjustEndpoints(endpointWithTags)
+
+		// Assert: Check that the endpoint's tag property is now a sorted, clean string.
+		require.NoError(t, err)
+		require.Len(t, adjustedEndpoints, 1)
+		val, ok := adjustedEndpoints[0].GetProviderSpecificProperty(annotations.CloudflareTagsKey)
+		assert.True(t, ok)
+		assert.Equal(t, expectedTagsString, val)
+	})
+
+	// This sub-test verifies that newCloudFlareChange correctly creates a sorted slice of tags.
+	t.Run("newCloudFlareChange creates sorted tags slice", func(t *testing.T) {
+		// Arrange: Create an endpoint with unsorted tags.
+		endpointWithTags := &endpoint.Endpoint{
+			DNSName:    "tags.example.com",
+			Targets:    endpoint.Targets{"1.2.3.4"},
+			RecordType: endpoint.RecordTypeA,
+			ProviderSpecific: endpoint.ProviderSpecific{
+				{
+					Name:  annotations.CloudflareTagsKey,
+					Value: "owner:team-b, env:prod, app:api ",
+				},
+			},
+		}
+		expectedTagsSlice := []string{"app:api", "env:prod", "owner:team-b"}
+
+		// Act: Call the function under test.
+		cfc, err := provider.newCloudFlareChange(cloudFlareCreate, endpointWithTags, "1.2.3.4", nil)
+
+		// Assert: Check that the resulting change object contains a sorted slice of strings for its tags.
+		require.NoError(t, err)
+		assert.Equal(t, expectedTagsSlice, cfc.ResourceRecord.Tags)
+	})
+}
+
 func NewMockCloudFlareClient() *mockCloudFlareClient {
 	return &mockCloudFlareClient{
 		Zones: map[string]string{
diff --git a/registry/txt_encryption_test.go b/registry/txt_encryption_test.go
index 1b7f9f5c..65cee624 100644
--- a/registry/txt_encryption_test.go
+++ b/registry/txt_encryption_test.go
@@ -61,7 +61,7 @@ func TestNewTXTRegistryEncryptionConfig(t *testing.T) {
 		},
 	}
 	for _, test := range tests {
-		actual, err := NewTXTRegistry(p, "txt.", "", "owner", time.Hour, "", []string{}, []string{}, test.encEnabled, test.aesKeyRaw)
+		actual, err := NewTXTRegistry(p, "txt.", "", "owner", time.Hour, "", []string{}, []string{}, test.encEnabled, test.aesKeyRaw, "")
 		if test.errorExpected {
 			require.Error(t, err)
 		} else {
@@ -107,7 +107,7 @@ func TestGenerateTXTGenerateTextRecordEncryptionWihDecryption(t *testing.T) {
 		for _, k := range withEncryptionKeys {
 			t.Run(fmt.Sprintf("key '%s' with decrypted result '%s'", k, test.decrypted), func(t *testing.T) {
 				key := []byte(k)
-				r, err := NewTXTRegistry(p, "", "", "owner", time.Minute, "", []string{}, []string{}, true, key)
+				r, err := NewTXTRegistry(p, "", "", "owner", time.Minute, "", []string{}, []string{}, true, key, "")
 				assert.NoError(t, err, "Error creating TXT registry")
 				txtRecords := r.generateTXTRecord(test.record)
 				assert.Len(t, txtRecords, len(test.record.Targets))
@@ -144,7 +144,7 @@ func TestApplyRecordsWithEncryption(t *testing.T) {
 
 	key := []byte("ZPitL0NGVQBZbTD6DwXJzD8RiStSazzYXQsdUowLURY=")
 
-	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, true, key)
+	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, true, key, "")
 
 	_ = r.ApplyChanges(ctx, &plan.Changes{
 		Create: []*endpoint.Endpoint{
@@ -202,7 +202,7 @@ func TestApplyRecordsWithEncryptionKeyChanged(t *testing.T) {
 	}
 
 	for _, key := range withEncryptionKeys {
-		r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, true, []byte(key))
+		r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, true, []byte(key), "")
 		_ = r.ApplyChanges(ctx, &plan.Changes{
 			Create: []*endpoint.Endpoint{
 				newEndpointWithOwner("new-record-1.test-zone.example.org", "new-loadbalancer-1.lb.com", endpoint.RecordTypeCNAME, "owner"),
@@ -232,7 +232,7 @@ func TestApplyRecordsOnEncryptionKeyChangeWithKeyIdLabel(t *testing.T) {
 	}
 
 	for i, key := range withEncryptionKeys {
-		r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, true, []byte(key))
+		r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, true, []byte(key), "")
 		keyId := fmt.Sprintf("key-id-%d", i)
 		changes := []*endpoint.Endpoint{
 			newEndpointWithOwnerAndOwnedRecordWithKeyIDLabel("new-record-1.test-zone.example.org", "new-loadbalancer-1.lb.com", endpoint.RecordTypeCNAME, "owner", "", keyId),
diff --git a/registry/txt_test.go b/registry/txt_test.go
index 05cdd030..b5146aa4 100644
--- a/registry/txt_test.go
+++ b/registry/txt_test.go
@@ -49,20 +49,20 @@ func TestTXTRegistry(t *testing.T) {
 
 func testTXTRegistryNew(t *testing.T) {
 	p := inmemory.NewInMemoryProvider()
-	_, err := NewTXTRegistry(p, "txt", "", "", time.Hour, "", []string{}, []string{}, false, nil)
+	_, err := NewTXTRegistry(p, "txt", "", "", time.Hour, "", []string{}, []string{}, false, nil, "")
 	require.Error(t, err)
 
-	_, err = NewTXTRegistry(p, "", "txt", "", time.Hour, "", []string{}, []string{}, false, nil)
+	_, err = NewTXTRegistry(p, "", "txt", "", time.Hour, "", []string{}, []string{}, false, nil, "")
 	require.Error(t, err)
 
-	r, err := NewTXTRegistry(p, "txt", "", "owner", time.Hour, "", []string{}, []string{}, false, nil)
+	r, err := NewTXTRegistry(p, "txt", "", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
 	require.NoError(t, err)
 	assert.Equal(t, p, r.provider)
 
-	r, err = NewTXTRegistry(p, "", "txt", "owner", time.Hour, "", []string{}, []string{}, false, nil)
+	r, err = NewTXTRegistry(p, "", "txt", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
 	require.NoError(t, err)
 
-	_, err = NewTXTRegistry(p, "txt", "txt", "owner", time.Hour, "", []string{}, []string{}, false, nil)
+	_, err = NewTXTRegistry(p, "txt", "txt", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
 	require.Error(t, err)
 
 	_, ok := r.mapper.(affixNameMapper)
@@ -71,16 +71,16 @@ func testTXTRegistryNew(t *testing.T) {
 	assert.Equal(t, p, r.provider)
 
 	aesKey := []byte(";k&l)nUC/33:{?d{3)54+,AD?]SX%yh^")
-	_, err = NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil)
+	_, err = NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
 	require.NoError(t, err)
 
-	_, err = NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, aesKey)
+	_, err = NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, aesKey, "")
 	require.NoError(t, err)
 
-	_, err = NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, true, nil)
+	_, err = NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, true, nil, "")
 	require.Error(t, err)
 
-	r, err = NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, true, aesKey)
+	r, err = NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, true, aesKey, "")
 	require.NoError(t, err)
 
 	_, ok = r.mapper.(affixNameMapper)
@@ -228,13 +228,13 @@ func testTXTRegistryRecordsPrefixed(t *testing.T) {
 		},
 	}
 
-	r, _ := NewTXTRegistry(p, "txt.", "", "owner", time.Hour, "wc", []string{}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "txt.", "", "owner", time.Hour, "wc", []string{}, []string{}, false, nil, "")
 	records, _ := r.Records(ctx)
 
 	assert.True(t, testutils.SameEndpoints(records, expectedRecords))
 
 	// Ensure prefix is case-insensitive
-	r, _ = NewTXTRegistry(p, "TxT.", "", "owner", time.Hour, "wc", []string{}, []string{}, false, nil)
+	r, _ = NewTXTRegistry(p, "TxT.", "", "owner", time.Hour, "wc", []string{}, []string{}, false, nil, "")
 	records, _ = r.Records(ctx)
 
 	assert.True(t, testutils.SameEndpoints(records, expectedRecords))
@@ -363,13 +363,13 @@ func testTXTRegistryRecordsSuffixed(t *testing.T) {
 		},
 	}
 
-	r, _ := NewTXTRegistry(p, "", "-txt", "owner", time.Hour, "", []string{}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "", "-txt", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
 	records, _ := r.Records(ctx)
 
 	assert.True(t, testutils.SameEndpoints(records, expectedRecords))
 
 	// Ensure prefix is case-insensitive
-	r, _ = NewTXTRegistry(p, "", "-TxT", "owner", time.Hour, "", []string{}, []string{}, false, nil)
+	r, _ = NewTXTRegistry(p, "", "-TxT", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
 	records, _ = r.Records(ctx)
 
 	assert.True(t, testutils.SameEndpointLabels(records, expectedRecords))
@@ -490,7 +490,7 @@ func testTXTRegistryRecordsNoPrefix(t *testing.T) {
 		},
 	}
 
-	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
 	records, _ := r.Records(ctx)
 
 	assert.True(t, testutils.SameEndpoints(records, expectedRecords))
@@ -527,12 +527,12 @@ func testTXTRegistryRecordsPrefixedTemplated(t *testing.T) {
 		},
 	}
 
-	r, _ := NewTXTRegistry(p, "txt-%{record_type}.", "", "owner", time.Hour, "wc", []string{}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "txt-%{record_type}.", "", "owner", time.Hour, "wc", []string{}, []string{}, false, nil, "")
 	records, _ := r.Records(ctx)
 
 	assert.True(t, testutils.SameEndpoints(records, expectedRecords))
 
-	r, _ = NewTXTRegistry(p, "TxT-%{record_type}.", "", "owner", time.Hour, "wc", []string{}, []string{}, false, nil)
+	r, _ = NewTXTRegistry(p, "TxT-%{record_type}.", "", "owner", time.Hour, "wc", []string{}, []string{}, false, nil, "")
 	records, _ = r.Records(ctx)
 
 	assert.True(t, testutils.SameEndpoints(records, expectedRecords))
@@ -569,12 +569,12 @@ func testTXTRegistryRecordsSuffixedTemplated(t *testing.T) {
 		},
 	}
 
-	r, _ := NewTXTRegistry(p, "", "txt%{record_type}", "owner", time.Hour, "wc", []string{}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "", "txt%{record_type}", "owner", time.Hour, "wc", []string{}, []string{}, false, nil, "")
 	records, _ := r.Records(ctx)
 
 	assert.True(t, testutils.SameEndpoints(records, expectedRecords))
 
-	r, _ = NewTXTRegistry(p, "", "TxT%{record_type}", "owner", time.Hour, "wc", []string{}, []string{}, false, nil)
+	r, _ = NewTXTRegistry(p, "", "TxT%{record_type}", "owner", time.Hour, "wc", []string{}, []string{}, false, nil, "")
 	records, _ = r.Records(ctx)
 
 	assert.True(t, testutils.SameEndpoints(records, expectedRecords))
@@ -617,7 +617,7 @@ func testTXTRegistryApplyChangesWithPrefix(t *testing.T) {
 			newEndpointWithOwner("txt.cname-multiple.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner\"", endpoint.RecordTypeTXT, "").WithSetIdentifier("test-set-2"),
 		},
 	})
-	r, _ := NewTXTRegistry(p, "txt.", "", "owner", time.Hour, "", []string{}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "txt.", "", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
 
 	changes := &plan.Changes{
 		Create: []*endpoint.Endpoint{
@@ -698,7 +698,7 @@ func testTXTRegistryApplyChangesWithTemplatedPrefix(t *testing.T) {
 	p.ApplyChanges(ctx, &plan.Changes{
 		Create: []*endpoint.Endpoint{},
 	})
-	r, _ := NewTXTRegistry(p, "prefix%{record_type}.", "", "owner", time.Hour, "", []string{}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "prefix%{record_type}.", "", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
 	changes := &plan.Changes{
 		Create: []*endpoint.Endpoint{
 			newEndpointWithOwnerResource("new-record-1.test-zone.example.org", "new-loadbalancer-1.lb.com", endpoint.RecordTypeCNAME, "", "ingress/default/my-ingress"),
@@ -741,7 +741,7 @@ func testTXTRegistryApplyChangesWithTemplatedSuffix(t *testing.T) {
 	p.OnApplyChanges = func(ctx context.Context, got *plan.Changes) {
 		assert.Equal(t, ctxEndpoints, ctx.Value(provider.RecordsContextKey))
 	}
-	r, _ := NewTXTRegistry(p, "", "-%{record_type}suffix", "owner", time.Hour, "", []string{}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "", "-%{record_type}suffix", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
 	changes := &plan.Changes{
 		Create: []*endpoint.Endpoint{
 			newEndpointWithOwnerResource("new-record-1.test-zone.example.org", "new-loadbalancer-1.lb.com", endpoint.RecordTypeCNAME, "", "ingress/default/my-ingress"),
@@ -806,7 +806,7 @@ func testTXTRegistryApplyChangesWithSuffix(t *testing.T) {
 			newEndpointWithOwner("cname-multiple-txt.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner\"", endpoint.RecordTypeTXT, "").WithSetIdentifier("test-set-2"),
 		},
 	})
-	r, _ := NewTXTRegistry(p, "", "-txt", "owner", time.Hour, "wildcard", []string{}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "", "-txt", "owner", time.Hour, "wildcard", []string{}, []string{}, false, nil, "")
 
 	changes := &plan.Changes{
 		Create: []*endpoint.Endpoint{
@@ -900,7 +900,7 @@ func testTXTRegistryApplyChangesNoPrefix(t *testing.T) {
 			newEndpointWithOwner("cname-foobar.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner\"", endpoint.RecordTypeTXT, ""),
 		},
 	})
-	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
 
 	changes := &plan.Changes{
 		Create: []*endpoint.Endpoint{
@@ -1058,7 +1058,7 @@ func testTXTRegistryMissingRecordsNoPrefix(t *testing.T) {
 		},
 	}
 
-	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "wc", []string{endpoint.RecordTypeCNAME, endpoint.RecordTypeA, endpoint.RecordTypeNS}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "wc", []string{endpoint.RecordTypeCNAME, endpoint.RecordTypeA, endpoint.RecordTypeNS}, []string{}, false, nil, "")
 	records, _ := r.Records(ctx)
 
 	assert.True(t, testutils.SameEndpoints(records, expectedRecords))
@@ -1168,7 +1168,7 @@ func testTXTRegistryMissingRecordsWithPrefix(t *testing.T) {
 		},
 	}
 
-	r, _ := NewTXTRegistry(p, "txt.", "", "owner", time.Hour, "wc", []string{endpoint.RecordTypeCNAME, endpoint.RecordTypeA, endpoint.RecordTypeNS, endpoint.RecordTypeTXT}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "txt.", "", "owner", time.Hour, "wc", []string{endpoint.RecordTypeCNAME, endpoint.RecordTypeA, endpoint.RecordTypeNS, endpoint.RecordTypeTXT}, []string{}, false, nil, "")
 	records, _ := r.Records(ctx)
 
 	assert.True(t, testutils.SameEndpoints(records, expectedRecords))
@@ -1463,7 +1463,7 @@ func TestNewTXTScheme(t *testing.T) {
 			newEndpointWithOwner("cname-foobar.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner\"", endpoint.RecordTypeTXT, ""),
 		},
 	})
-	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
 
 	changes := &plan.Changes{
 		Create: []*endpoint.Endpoint{
@@ -1528,11 +1528,46 @@ func TestGenerateTXT(t *testing.T) {
 	}
 	p := inmemory.NewInMemoryProvider()
 	p.CreateZone(testZone)
-	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
 	gotTXT := r.generateTXTRecord(record)
 	assert.Equal(t, expectedTXT, gotTXT)
 }
 
+func TestGenerateTXTWithMigration(t *testing.T) {
+	record := newEndpointWithOwner("foo.test-zone.example.org", "1.2.3.4", endpoint.RecordTypeA, "owner")
+	expectedTXTBeforeMigration := []*endpoint.Endpoint{
+		{
+			DNSName:    "a-foo.test-zone.example.org",
+			Targets:    endpoint.Targets{"\"heritage=external-dns,external-dns/owner=owner\""},
+			RecordType: endpoint.RecordTypeTXT,
+			Labels: map[string]string{
+				endpoint.OwnedRecordLabelKey: "foo.test-zone.example.org",
+			},
+		},
+	}
+	p := inmemory.NewInMemoryProvider()
+	p.CreateZone(testZone)
+	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
+	gotTXTBeforeMigration := r.generateTXTRecord(record)
+	assert.Equal(t, expectedTXTBeforeMigration, gotTXTBeforeMigration)
+
+	expectedTXTAfterMigration := []*endpoint.Endpoint{
+		{
+			DNSName:    "a-foo.test-zone.example.org",
+			Targets:    endpoint.Targets{"\"heritage=external-dns,external-dns/owner=foobar\""},
+			RecordType: endpoint.RecordTypeTXT,
+			Labels: map[string]string{
+				endpoint.OwnedRecordLabelKey: "foo.test-zone.example.org",
+			},
+		},
+	}
+
+	rMigrated, _ := NewTXTRegistry(p, "", "", "foobar", time.Hour, "", []string{}, []string{}, false, nil, "owner")
+	gotTXTAfterMigration := rMigrated.generateTXTRecord(record)
+	assert.Equal(t, expectedTXTAfterMigration, gotTXTAfterMigration)
+
+}
+
 func TestGenerateTXTForAAAA(t *testing.T) {
 	record := newEndpointWithOwner("foo.test-zone.example.org", "2001:DB8::1", endpoint.RecordTypeAAAA, "owner")
 	expectedTXT := []*endpoint.Endpoint{
@@ -1547,7 +1582,7 @@ func TestGenerateTXTForAAAA(t *testing.T) {
 	}
 	p := inmemory.NewInMemoryProvider()
 	p.CreateZone(testZone)
-	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
 	gotTXT := r.generateTXTRecord(record)
 	assert.Equal(t, expectedTXT, gotTXT)
 }
@@ -1564,7 +1599,7 @@ func TestFailGenerateTXT(t *testing.T) {
 	expectedTXT := []*endpoint.Endpoint{}
 	p := inmemory.NewInMemoryProvider()
 	p.CreateZone(testZone)
-	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
 	gotTXT := r.generateTXTRecord(cnameRecord)
 	assert.Equal(t, expectedTXT, gotTXT)
 }
@@ -1582,7 +1617,7 @@ func TestTXTRegistryApplyChangesEncrypt(t *testing.T) {
 		},
 	})
 
-	r, _ := NewTXTRegistry(p, "txt.", "", "owner", time.Hour, "", []string{}, []string{}, true, []byte("12345678901234567890123456789012"))
+	r, _ := NewTXTRegistry(p, "txt.", "", "owner", time.Hour, "", []string{}, []string{}, true, []byte("12345678901234567890123456789012"), "")
 	records, _ := r.Records(ctx)
 	changes := &plan.Changes{
 		Delete: records,
@@ -1628,7 +1663,7 @@ func TestMultiClusterDifferentRecordTypeOwnership(t *testing.T) {
 		},
 	})
 
-	r, _ := NewTXTRegistry(p, "_owner.", "", "bar", time.Hour, "", []string{}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "_owner.", "", "bar", time.Hour, "", []string{}, []string{}, false, nil, "")
 	records, _ := r.Records(ctx)
 
 	// new cluster has same ingress host as other cluster and uses CNAME ingress address
@@ -1713,7 +1748,7 @@ func TestGenerateTXTRecordWithNewFormatOnly(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil)
+			r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
 			records := r.generateTXTRecord(tc.endpoint)
 
 			assert.Len(t, records, tc.expectedRecords, tc.description)
@@ -1742,7 +1777,7 @@ func TestApplyChangesWithNewFormatOnly(t *testing.T) {
 	p.CreateZone(testZone)
 	ctx := context.Background()
 
-	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
 
 	changes := &plan.Changes{
 		Create: []*endpoint.Endpoint{
@@ -1790,7 +1825,7 @@ func TestTXTRegistryRecordsWithEmptyTargets(t *testing.T) {
 		},
 	})
 
-	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil)
+	r, _ := NewTXTRegistry(p, "", "", "owner", time.Hour, "", []string{}, []string{}, false, nil, "")
 	hook := testutils.LogsUnderTestWithLogLevel(log.ErrorLevel, t)
 	records, err := r.Records(ctx)
 	require.NoError(t, err)
@@ -1994,7 +2029,7 @@ func TestTXTRegistryRecreatesMissingRecords(t *testing.T) {
 
 					// When: Apply changes to recreate missing A records
 					managedRecords := []string{endpoint.RecordTypeA, endpoint.RecordTypeCNAME, endpoint.RecordTypeAAAA, endpoint.RecordTypeTXT}
-					registry, err := NewTXTRegistry(p, "", "", ownerId, time.Hour, "", managedRecords, nil, false, nil)
+					registry, err := NewTXTRegistry(p, "", "", ownerId, time.Hour, "", managedRecords, nil, false, nil, "")
 					assert.NoError(t, err)
 
 					expectedRecords := append(existing, expectedCreate...)
@@ -2028,3 +2063,49 @@ func TestTXTRegistryRecreatesMissingRecords(t *testing.T) {
 		}
 	}
 }
+
+func TestTXTRecordMigration(t *testing.T) {
+	ctx := context.Background()
+	p := inmemory.NewInMemoryProvider()
+	p.CreateZone(testZone)
+
+	r, _ := NewTXTRegistry(p, "%{record_type}-", "", "foo", time.Hour, "", []string{}, []string{}, false, nil, "")
+
+	r.ApplyChanges(ctx, &plan.Changes{
+		Create: []*endpoint.Endpoint{
+			// records on cluster using A record for ingress address
+			newEndpointWithOwnerAndLabels("bar.test-zone.example.org", "1.2.3.4", endpoint.RecordTypeA, "foo", endpoint.Labels{endpoint.OwnerLabelKey: "owner"}),
+		},
+	})
+
+	createdRecords, _ := r.Records(ctx)
+
+	newTXTRecord := r.generateTXTRecord(createdRecords[0])
+
+	expectedTXTRecords := []*endpoint.Endpoint{
+		{
+			DNSName:    "a-bar.test-zone.example.org",
+			Targets:    endpoint.Targets{"\"heritage=external-dns,external-dns/owner=foo\""},
+			RecordType: endpoint.RecordTypeTXT,
+		},
+	}
+
+	assert.Equal(t, expectedTXTRecords[0].Targets, newTXTRecord[0].Targets)
+
+	r, _ = NewTXTRegistry(p, "%{record_type}-", "", "foobar", time.Hour, "", []string{}, []string{}, false, nil, "foo")
+
+	updatedRecords, _ := r.Records(ctx)
+
+	updatedTXTRecord := r.generateTXTRecord(updatedRecords[0])
+
+	expectedFinalTXT := []*endpoint.Endpoint{
+		{
+			DNSName:    "a-bar.test-zone.example.org",
+			Targets:    endpoint.Targets{"\"heritage=external-dns,external-dns/owner=foobar\""},
+			RecordType: endpoint.RecordTypeTXT,
+		},
+	}
+
+	assert.Equal(t, updatedTXTRecord[0].Targets, expectedFinalTXT[0].Targets)
+
+}
diff --git a/source/annotations/provider_specific_test.go b/source/annotations/provider_specific_test.go
index eb56e1d1..684ea431 100644
--- a/source/annotations/provider_specific_test.go
+++ b/source/annotations/provider_specific_test.go
@@ -94,6 +94,41 @@ func TestProviderSpecificAnnotations(t *testing.T) {
 }
 
 func TestGetProviderSpecificCloudflareAnnotations(t *testing.T) {
+
+	for _, tc := range []struct {
+		title         string
+		annotations   map[string]string
+		expectedKey   string
+		expectedValue string
+	}{
+		{
+			title:         "Cloudflare tags annotation is set correctly",
+			annotations:   map[string]string{CloudflareTagsKey: "env:test,owner:team-a"},
+			expectedKey:   CloudflareTagsKey,
+			expectedValue: "env:test,owner:team-a",
+		},
+		{
+			title: "Cloudflare tags annotation among another annotations is set correctly",
+			annotations: map[string]string{
+				"random annotation 1": "random value 1",
+				CloudflareTagsKey:     "env:test,owner:team-b",
+				"random annotation 2": "random value 2"},
+			expectedKey:   CloudflareTagsKey,
+			expectedValue: "env:test,owner:team-b",
+		},
+	} {
+		t.Run(tc.title, func(t *testing.T) {
+			providerSpecificAnnotations, _ := ProviderSpecificAnnotations(tc.annotations)
+			for _, providerSpecificAnnotation := range providerSpecificAnnotations {
+				if providerSpecificAnnotation.Name == tc.expectedKey {
+					assert.Equal(t, tc.expectedValue, providerSpecificAnnotation.Value)
+					return
+				}
+			}
+			t.Errorf("Cloudflare provider specific annotation %s is not set correctly to %s", tc.expectedKey, tc.expectedValue)
+		})
+	}
+
 	for _, tc := range []struct {
 		title         string
 		annotations   map[string]string