diff --git a/controller/execute.go b/controller/execute.go
index 45da78dbc..50d93e2e6 100644
--- a/controller/execute.go
+++ b/controller/execute.go
@@ -242,8 +242,10 @@ func buildProvider(
 				CertificateAuthority: cfg.CloudflareCustomHostnamesCertificateAuthority,
 			},
 			cloudflare.DNSRecordsConfig{
-				PerPage: cfg.CloudflareDNSRecordsPerPage,
-				Comment: cfg.CloudflareDNSRecordsComment,
+				PerPage:             cfg.CloudflareDNSRecordsPerPage,
+				Comment:             cfg.CloudflareDNSRecordsComment,
+				BatchChangeSize:     cfg.BatchChangeSize,
+				BatchChangeInterval: cfg.BatchChangeInterval,
 			})
 	case "google":
 		p, err = google.NewGoogleProvider(ctx, cfg.GoogleProject, domainFilter, zoneIDFilter, cfg.GoogleBatchChangeSize, cfg.GoogleBatchChangeInterval, cfg.GoogleZoneVisibility, cfg.DryRun)
diff --git a/docs/flags.md b/docs/flags.md
index 0e88b3146..cb22807a9 100644
--- a/docs/flags.md
+++ b/docs/flags.md
@@ -85,6 +85,8 @@
 | `--azure-user-assigned-identity-client-id=""` | When using the Azure provider, override the client id of user assigned identity in config file (optional) |
 | `--azure-zones-cache-duration=0s` | When using the Azure provider, set the zones list cache TTL (0s to disable). |
 | `--azure-maxretries-count=3` | When using the Azure provider, set the number of retries for API calls (When less than 0, it disables retries). (optional) |
+| `--batch-change-size=200` | Set the maximum number of DNS record changes that will be submitted to the provider in each batch (optional) |
+| `--batch-change-interval=1s` | Set the interval between batch changes (optional, default: 1s) |
 | `--[no-]cloudflare-proxied` | When using the Cloudflare provider, specify if the proxy mode must be enabled (default: disabled) |
 | `--[no-]cloudflare-custom-hostnames` | When using the Cloudflare provider, specify if the Custom Hostnames feature will be used. Requires "Cloudflare for SaaS" enabled. (default: disabled) |
 | `--cloudflare-custom-hostnames-min-tls-version=1.0` | When using the Cloudflare provider with the Custom Hostnames, specify which Minimum TLS Version will be used by default. (default: 1.0, options: 1.0, 1.1, 1.2, 1.3) |
diff --git a/docs/tutorials/cloudflare.md b/docs/tutorials/cloudflare.md
index 031347137..b8c9f05d4 100644
--- a/docs/tutorials/cloudflare.md
+++ b/docs/tutorials/cloudflare.md
@@ -33,6 +33,21 @@ If you would like to further restrict the API permissions to a specific zone (or
 Cloudflare API has a [global rate limit of 1,200 requests per five minutes](https://developers.cloudflare.com/fundamentals/api/reference/limits/). Running several fast polling ExternalDNS instances in a given account can easily hit that limit.
 The AWS Provider [docs](./aws.md#throttling) has some recommendations that can be followed here too, but in particular, consider passing `--cloudflare-dns-records-per-page` with a high value (maximum is 5,000).
 
+## Batch API
+
+The Cloudflare provider submits DNS record changes using Cloudflare's [Batch DNS Records API](https://developers.cloudflare.com/api/resources/dns/subresources/records/methods/batch/).
+All creates, updates, and deletes for a zone are grouped into transactional chunks and sent in a single API call per chunk,
+significantly reducing the total number of requests made.
+
+The batch API is transactional — if a chunk fails, the entire chunk is rolled back by Cloudflare.
+In that case, ExternalDNS automatically retries each record change in the chunk individually.
+Record types that are not supported by the batch PUT operation (e.g. SRV, CAA) are always submitted individually rather than through the batch API.
+
+| Flag | Default | Description |
+| :--- | :------ | :---------- |
+| `--batch-change-size` | `200` | Maximum number of DNS operations (creates + updates + deletes) per batch chunk. |
+| `--batch-change-interval` | `1s` | Pause between consecutive batch chunks. |
+
 ## Deploy ExternalDNS
 
 Connect your `kubectl` client to the cluster you want to test ExternalDNS with.
diff --git a/pkg/apis/externaldns/types.go b/pkg/apis/externaldns/types.go
index e4f8fc165..f299e7cb4 100644
--- a/pkg/apis/externaldns/types.go
+++ b/pkg/apis/externaldns/types.go
@@ -112,6 +112,8 @@ type Config struct {
 	AzureActiveDirectoryAuthorityHost             string
 	AzureZonesCacheDuration                       time.Duration
 	AzureMaxRetriesCount                          int
+	BatchChangeSize                               int
+	BatchChangeInterval                           time.Duration
 	CloudflareProxied                             bool
 	CloudflareCustomHostnames                     bool
 	CloudflareDNSRecordsPerPage                   int
@@ -256,6 +258,8 @@ var defaultConfig = &Config{
 	AzureSubscriptionID:         "",
 	AzureZonesCacheDuration:     0 * time.Second,
 	AzureMaxRetriesCount:        3,
+	BatchChangeSize:             200,
+	BatchChangeInterval:         time.Second,
 	CloudflareCustomHostnamesCertificateAuthority: "none",
 	CloudflareCustomHostnames:                     false,
 	CloudflareCustomHostnamesMinTLSVersion:        "1.0",
@@ -587,6 +591,8 @@ func bindFlags(b flags.FlagBinder, cfg *Config) {
 	b.DurationVar("azure-zones-cache-duration", "When using the Azure provider, set the zones list cache TTL (0s to disable).", defaultConfig.AzureZonesCacheDuration, &cfg.AzureZonesCacheDuration)
 	b.IntVar("azure-maxretries-count", "When using the Azure provider, set the number of retries for API calls (When less than 0, it disables retries). (optional)", defaultConfig.AzureMaxRetriesCount, &cfg.AzureMaxRetriesCount)
 
+	b.IntVar("batch-change-size", "Set the maximum number of DNS record changes that will be submitted to the provider in each batch (optional)", defaultConfig.BatchChangeSize, &cfg.BatchChangeSize)
+	b.DurationVar("batch-change-interval", "Set the interval between batch changes (optional, default: 1s)", defaultConfig.BatchChangeInterval, &cfg.BatchChangeInterval)
 	b.BoolVar("cloudflare-proxied", "When using the Cloudflare provider, specify if the proxy mode must be enabled (default: disabled)", false, &cfg.CloudflareProxied)
 	b.BoolVar("cloudflare-custom-hostnames", "When using the Cloudflare provider, specify if the Custom Hostnames feature will be used. Requires \"Cloudflare for SaaS\" enabled. (default: disabled)", false, &cfg.CloudflareCustomHostnames)
 	b.EnumVar("cloudflare-custom-hostnames-min-tls-version", "When using the Cloudflare provider with the Custom Hostnames, specify which Minimum TLS Version will be used by default. (default: 1.0, options: 1.0, 1.1, 1.2, 1.3)", "1.0", &cfg.CloudflareCustomHostnamesMinTLSVersion, "1.0", "1.1", "1.2", "1.3")
diff --git a/pkg/apis/externaldns/types_test.go b/pkg/apis/externaldns/types_test.go
index fa3af6ec0..e8c4c9ad8 100644
--- a/pkg/apis/externaldns/types_test.go
+++ b/pkg/apis/externaldns/types_test.go
@@ -76,6 +76,8 @@ var (
 		AzureResourceGroup:                     "",
 		AzureSubscriptionID:                    "",
 		AzureMaxRetriesCount:                   3,
+		BatchChangeSize:                        200,
+		BatchChangeInterval:                    time.Second,
 		CloudflareProxied:                      false,
 		CloudflareCustomHostnames:              false,
 		CloudflareCustomHostnamesMinTLSVersion: "1.0",
@@ -185,6 +187,8 @@ var (
 		AzureResourceGroup:                     "arg",
 		AzureSubscriptionID:                    "arg",
 		AzureMaxRetriesCount:                   4,
+		BatchChangeSize:                        200,
+		BatchChangeInterval:                    time.Second,
 		CloudflareProxied:                      true,
 		CloudflareCustomHostnames:              true,
 		CloudflareCustomHostnamesMinTLSVersion: "1.3",
@@ -426,6 +430,7 @@ func TestParseFlags(t *testing.T) {
 				"--rfc2136-load-balancing-strategy=round-robin",
 				"--rfc2136-host=rfc2136-host1",
 				"--rfc2136-host=rfc2136-host2",
+				"--batch-change-size=200",
 			},
 			envVars: map[string]string{},
 			expected: func(cfg *Config) {
@@ -547,6 +552,7 @@ func TestParseFlags(t *testing.T) {
 				"EXTERNAL_DNS_RFC2136_BATCH_CHANGE_SIZE":                         "100",
 				"EXTERNAL_DNS_RFC2136_LOAD_BALANCING_STRATEGY":                   "round-robin",
 				"EXTERNAL_DNS_RFC2136_HOST":                                      "rfc2136-host1\nrfc2136-host2",
+				"EXTERNAL_DNS_BATCH_CHANGE_SIZE":                                 "200",
 			},
 			expected: func(cfg *Config) {
 				assert.Equal(t, overriddenConfig, cfg)
diff --git a/provider/cloudflare/cloudflare.go b/provider/cloudflare/cloudflare.go
index 0a91896e2..5cae9b000 100644
--- a/provider/cloudflare/cloudflare.go
+++ b/provider/cloudflare/cloudflare.go
@@ -20,11 +20,13 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"os"
 	"sort"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/cloudflare/cloudflare-go/v5"
 	"github.com/cloudflare/cloudflare-go/v5/addressing"
@@ -96,6 +98,7 @@ type cloudFlareDNS interface {
 	ListZones(ctx context.Context, params zones.ZoneListParams) autoPager[zones.Zone]
 	GetZone(ctx context.Context, zoneID string) (*zones.Zone, error)
 	ListDNSRecords(ctx context.Context, params dns.RecordListParams) autoPager[dns.RecordResponse]
+	BatchDNSRecords(ctx context.Context, params dns.RecordBatchParams) (*dns.RecordBatchResponse, error)
 	CreateDNSRecord(ctx context.Context, params dns.RecordNewParams) (*dns.RecordResponse, error)
 	DeleteDNSRecord(ctx context.Context, recordID string, params dns.RecordDeleteParams) error
 	UpdateDNSRecord(ctx context.Context, recordID string, params dns.RecordUpdateParams) (*dns.RecordResponse, error)
@@ -163,8 +166,10 @@ func listZonesV4Params() zones.ZoneListParams {
 }
 
 type DNSRecordsConfig struct {
-	PerPage int
-	Comment string
+	PerPage             int
+	Comment             string
+	BatchChangeSize     int
+	BatchChangeInterval time.Duration
 }
 
 func (c *DNSRecordsConfig) trimAndValidateComment(dnsName, comment string, paidZone func(string) bool) string {
@@ -229,40 +234,6 @@ type cloudFlareChange struct {
 	CustomHostnamesPrev []string
 }
 
-// updateDNSRecordParam is a function that returns the appropriate Record Param based on the cloudFlareChange passed in
-func getUpdateDNSRecordParam(zoneID string, cfc cloudFlareChange) dns.RecordUpdateParams {
-	return dns.RecordUpdateParams{
-		ZoneID: cloudflare.F(zoneID),
-		Body: dns.RecordUpdateParamsBody{
-			Name:     cloudflare.F(cfc.ResourceRecord.Name),
-			TTL:      cloudflare.F(cfc.ResourceRecord.TTL),
-			Proxied:  cloudflare.F(cfc.ResourceRecord.Proxied),
-			Type:     cloudflare.F(dns.RecordUpdateParamsBodyType(cfc.ResourceRecord.Type)),
-			Content:  cloudflare.F(cfc.ResourceRecord.Content),
-			Priority: cloudflare.F(cfc.ResourceRecord.Priority),
-			Comment:  cloudflare.F(cfc.ResourceRecord.Comment),
-			Tags:     cloudflare.F(cfc.ResourceRecord.Tags),
-		},
-	}
-}
-
-// getCreateDNSRecordParam is a function that returns the appropriate Record Param based on the cloudFlareChange passed in
-func getCreateDNSRecordParam(zoneID string, cfc *cloudFlareChange) dns.RecordNewParams {
-	return dns.RecordNewParams{
-		ZoneID: cloudflare.F(zoneID),
-		Body: dns.RecordNewParamsBody{
-			Name:     cloudflare.F(cfc.ResourceRecord.Name),
-			TTL:      cloudflare.F(cfc.ResourceRecord.TTL),
-			Proxied:  cloudflare.F(cfc.ResourceRecord.Proxied),
-			Type:     cloudflare.F(dns.RecordNewParamsBodyType(cfc.ResourceRecord.Type)),
-			Content:  cloudflare.F(cfc.ResourceRecord.Content),
-			Priority: cloudflare.F(cfc.ResourceRecord.Priority),
-			Comment:  cloudflare.F(cfc.ResourceRecord.Comment),
-			Tags:     cloudflare.F(cfc.ResourceRecord.Tags),
-		},
-	}
-}
-
 func convertCloudflareError(err error) error {
 	// Handle CloudFlare v5 SDK errors according to the documentation:
 	// https://github.com/cloudflare/cloudflare-go?tab=readme-ov-file#errors
@@ -278,7 +249,14 @@ func convertCloudflareError(err error) error {
 		return err
 	}
 
-	// Also check for rate limit indicators in error message strings as a fallback.
+	// Transport-level errors that the SDK does not wrap as *cloudflare.Error.
+	// Both are transient and worth retrying at the external-dns level.
+	//   ErrUnexpectedEOF – connection closed mid-response (during body read)
+	//   EOF              – connection closed before any response bytes arrived
+	if errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, io.EOF) {
+		return provider.NewSoftError(err)
+	}
+
 	// The v5 SDK's retry logic and error wrapping can hide the structured error type,
 	// so we need string matching to catch rate limits in wrapped errors like:
 	// "exceeded available rate limit retries" from the SDK's auto-retry mechanism.
@@ -534,64 +512,55 @@ func (p *CloudFlareProvider) submitChanges(ctx context.Context, changes []*cloud
 				"action": change.Action.String(),
 				"zone":   zoneID,
 			}
-
 			log.WithFields(logFields).Info("Changing record.")
-
-			if p.DryRun {
-				continue
-			}
-
-			records, err := p.getDNSRecordsMap(ctx, zoneID)
-			if err != nil {
-				return fmt.Errorf("could not fetch records from zone, %w", err)
-			}
-			chs, chErr := p.listCustomHostnamesWithPagination(ctx, zoneID)
-			if chErr != nil {
-				return fmt.Errorf("could not fetch custom hostnames from zone, %w", chErr)
-			}
-			switch change.Action {
-			case cloudFlareUpdate:
-				if !p.submitCustomHostnameChanges(ctx, zoneID, change, chs, logFields) {
-					failedChange = true
-				}
-				recordID := p.getRecordID(records, change.ResourceRecord)
-				if recordID == "" {
-					log.WithFields(logFields).Errorf("failed to find previous record: %v", change.ResourceRecord)
-					continue
-				}
-				recordParam := getUpdateDNSRecordParam(zoneID, *change)
-				_, err := p.Client.UpdateDNSRecord(ctx, recordID, recordParam)
-				if err != nil {
-					failedChange = true
-					log.WithFields(logFields).Errorf("failed to update record: %v", err)
-				}
-			case cloudFlareDelete:
-				recordID := p.getRecordID(records, change.ResourceRecord)
-				if recordID == "" {
-					log.WithFields(logFields).Errorf("failed to find previous record: %v", change.ResourceRecord)
-					continue
-				}
-				err := p.Client.DeleteDNSRecord(ctx, recordID, dns.RecordDeleteParams{ZoneID: cloudflare.F(zoneID)})
-				if err != nil {
-					failedChange = true
-					log.WithFields(logFields).Errorf("failed to delete record: %v", err)
-				}
-				if !p.submitCustomHostnameChanges(ctx, zoneID, change, chs, logFields) {
-					failedChange = true
-				}
-			case cloudFlareCreate:
-				recordParam := getCreateDNSRecordParam(zoneID, change)
-				_, err := p.Client.CreateDNSRecord(ctx, recordParam)
-				if err != nil {
-					failedChange = true
-					log.WithFields(logFields).Errorf("failed to create record: %v", err)
-				}
-				if !p.submitCustomHostnameChanges(ctx, zoneID, change, chs, logFields) {
-					failedChange = true
-				}
-			}
 		}
 
+		if p.DryRun {
+			// In dry-run mode, skip all DNS record mutations but still process
+			// regional hostname changes (which have their own dry-run logging).
+			if p.RegionalServicesConfig.Enabled {
+				desiredRegionalHostnames, err := desiredRegionalHostnames(zoneChanges)
+				if err != nil {
+					return fmt.Errorf("failed to build desired regional hostnames: %w", err)
+				}
+				if len(desiredRegionalHostnames) > 0 {
+					regionalHostnames, err := p.listDataLocalisationRegionalHostnames(ctx, zoneID)
+					if err != nil {
+						return fmt.Errorf("could not fetch regional hostnames from zone, %w", err)
+					}
+					regionalHostnamesChanges := regionalHostnamesChanges(desiredRegionalHostnames, regionalHostnames)
+					if !p.submitRegionalHostnameChanges(ctx, zoneID, regionalHostnamesChanges) {
+						failedChange = true
+					}
+				}
+			}
+			if failedChange {
+				failedZones = append(failedZones, zoneID)
+			}
+			continue
+		}
+
+		// Fetch the zone's current DNS records and custom hostnames once, rather
+		// than once per change, to avoid O(n) API calls for n changes.
+		records, err := p.getDNSRecordsMap(ctx, zoneID)
+		if err != nil {
+			return fmt.Errorf("could not fetch records from zone, %w", err)
+		}
+		chs, chErr := p.listCustomHostnamesWithPagination(ctx, zoneID)
+		if chErr != nil {
+			return fmt.Errorf("could not fetch custom hostnames from zone, %w", chErr)
+		}
+
+		// Apply custom hostname side-effects (separate Cloudflare API), then
+		// classify DNS record changes into batch collections.
+		if p.processCustomHostnameChanges(ctx, zoneID, zoneChanges, chs) {
+			failedChange = true
+		}
+		bc := p.buildBatchCollections(zoneID, zoneChanges, records)
+
+		if p.submitDNSRecordChanges(ctx, zoneID, bc, records) {
+			failedChange = true
+		}
 		if p.RegionalServicesConfig.Enabled {
 			desiredRegionalHostnames, err := desiredRegionalHostnames(zoneChanges)
 			if err != nil {
diff --git a/provider/cloudflare/cloudflare_batch.go b/provider/cloudflare/cloudflare_batch.go
new file mode 100644
index 000000000..5d9d4f0dd
--- /dev/null
+++ b/provider/cloudflare/cloudflare_batch.go
@@ -0,0 +1,447 @@
+/*
+Copyright 2026 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cloudflare
+
+import (
+	"context"
+	"time"
+
+	"github.com/cloudflare/cloudflare-go/v5"
+	"github.com/cloudflare/cloudflare-go/v5/dns"
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	// defaultBatchChangeSize is the default maximum number of DNS record
+	// operations included in each Cloudflare batch request.
+	defaultBatchChangeSize = 200
+)
+
+// batchCollections groups the parallel slices that are assembled while
+// classifying per-zone changes. It is passed as a unit to
+// submitDNSRecordChanges and chunkBatchChanges, replacing the previous
+// eight-parameter signatures and making it clear which slices travel
+// together.
+type batchCollections struct {
+	// Batch API parameters in server-execution order: deletes → puts → posts.
+	batchDeletes []dns.RecordBatchParamsDelete
+	batchPosts   []dns.RecordBatchParamsPostUnion
+	batchPuts    []dns.BatchPutUnionParam
+
+	// Parallel change slices — one entry per batch param, in the same order,
+	// so that a failed batch chunk can be replayed with per-record fallback.
+	deleteChanges []*cloudFlareChange
+	createChanges []*cloudFlareChange
+	updateChanges []*cloudFlareChange
+
+	// fallbackUpdates holds changes for record types whose batch-put param
+	// requires structured Data fields (e.g. SRV, CAA). These are submitted
+	// via individual UpdateDNSRecord calls instead of the batch API.
+	fallbackUpdates []*cloudFlareChange
+}
+
+// batchChunk holds a DNS record batch request alongside the source changes
+// that produced it, enabling per-record fallback when a batch fails.
+type batchChunk struct {
+	params        dns.RecordBatchParams
+	deleteChanges []*cloudFlareChange
+	createChanges []*cloudFlareChange
+	updateChanges []*cloudFlareChange
+}
+
+// BatchDNSRecords submits a batch of DNS record changes to the Cloudflare API.
+func (z zoneService) BatchDNSRecords(ctx context.Context, params dns.RecordBatchParams) (*dns.RecordBatchResponse, error) {
+	return z.service.DNS.Records.Batch(ctx, params)
+}
+
+// getUpdateDNSRecordParam returns the RecordUpdateParams for an individual update.
+func getUpdateDNSRecordParam(zoneID string, cfc cloudFlareChange) dns.RecordUpdateParams {
+	return dns.RecordUpdateParams{
+		ZoneID: cloudflare.F(zoneID),
+		Body: dns.RecordUpdateParamsBody{
+			Name:     cloudflare.F(cfc.ResourceRecord.Name),
+			TTL:      cloudflare.F(cfc.ResourceRecord.TTL),
+			Proxied:  cloudflare.F(cfc.ResourceRecord.Proxied),
+			Type:     cloudflare.F(dns.RecordUpdateParamsBodyType(cfc.ResourceRecord.Type)),
+			Content:  cloudflare.F(cfc.ResourceRecord.Content),
+			Priority: cloudflare.F(cfc.ResourceRecord.Priority),
+			Comment:  cloudflare.F(cfc.ResourceRecord.Comment),
+			Tags:     cloudflare.F(cfc.ResourceRecord.Tags),
+		},
+	}
+}
+
+// getCreateDNSRecordParam returns the RecordNewParams for an individual create.
+func getCreateDNSRecordParam(zoneID string, cfc *cloudFlareChange) dns.RecordNewParams {
+	return dns.RecordNewParams{
+		ZoneID: cloudflare.F(zoneID),
+		Body: dns.RecordNewParamsBody{
+			Name:     cloudflare.F(cfc.ResourceRecord.Name),
+			TTL:      cloudflare.F(cfc.ResourceRecord.TTL),
+			Proxied:  cloudflare.F(cfc.ResourceRecord.Proxied),
+			Type:     cloudflare.F(dns.RecordNewParamsBodyType(cfc.ResourceRecord.Type)),
+			Content:  cloudflare.F(cfc.ResourceRecord.Content),
+			Priority: cloudflare.F(cfc.ResourceRecord.Priority),
+			Comment:  cloudflare.F(cfc.ResourceRecord.Comment),
+			Tags:     cloudflare.F(cfc.ResourceRecord.Tags),
+		},
+	}
+}
+
+// chunkBatchChanges splits DNS record batch operations into batchChunks,
+// each containing at most <limit> total operations. Operations are distributed
+// in server-execution order: deletes first, then puts, then posts.
+// The parallel change slices track which cloudFlareChange produced each batch
+// param so that individual fallback is possible when a chunk fails.
+func chunkBatchChanges(zoneID string, bc batchCollections, limit int) []batchChunk {
+	deletes, deleteChanges := bc.batchDeletes, bc.deleteChanges
+	posts, createChanges := bc.batchPosts, bc.createChanges
+	puts, updateChanges := bc.batchPuts, bc.updateChanges
+
+	var chunks []batchChunk
+	di, pi, ui := 0, 0, 0
+	for di < len(deletes) || pi < len(posts) || ui < len(puts) {
+		remaining := limit
+		chunk := batchChunk{
+			params: dns.RecordBatchParams{ZoneID: cloudflare.F(zoneID)},
+		}
+
+		if di < len(deletes) && remaining > 0 {
+			end := min(di+remaining, len(deletes))
+			chunk.params.Deletes = cloudflare.F(deletes[di:end])
+			chunk.deleteChanges = deleteChanges[di:end]
+			remaining -= end - di
+			di = end
+		}
+
+		if ui < len(puts) && remaining > 0 {
+			end := min(ui+remaining, len(puts))
+			chunk.params.Puts = cloudflare.F(puts[ui:end])
+			chunk.updateChanges = updateChanges[ui:end]
+			remaining -= end - ui
+			ui = end
+		}
+
+		if pi < len(posts) && remaining > 0 {
+			end := min(pi+remaining, len(posts))
+			chunk.params.Posts = cloudflare.F(posts[pi:end])
+			chunk.createChanges = createChanges[pi:end]
+			pi = end
+		}
+
+		chunks = append(chunks, chunk)
+	}
+	return chunks
+}
+
+// tagsFromResponse converts a RecordResponse Tags field (any) to the typed tag slice.
+func tagsFromResponse(tags any) []dns.RecordTagsParam {
+	if ts, ok := tags.([]string); ok {
+		return ts
+	}
+	return nil
+}
+
+// buildBatchPostParam constructs a RecordBatchParamsPost for creating a DNS record in a batch.
+func buildBatchPostParam(r dns.RecordResponse) dns.RecordBatchParamsPost {
+	return dns.RecordBatchParamsPost{
+		Name:     cloudflare.F(r.Name),
+		TTL:      cloudflare.F(r.TTL),
+		Type:     cloudflare.F(dns.RecordBatchParamsPostsType(r.Type)),
+		Content:  cloudflare.F(r.Content),
+		Proxied:  cloudflare.F(r.Proxied),
+		Priority: cloudflare.F(r.Priority),
+		Comment:  cloudflare.F(r.Comment),
+		Tags:     cloudflare.F[any](tagsFromResponse(r.Tags)),
+	}
+}
+
+// buildBatchPutParam constructs a BatchPutUnionParam for updating a DNS record in a batch.
+// Returns (nil, false) for record types that use structured Data fields (e.g. SRV, CAA),
+// which fall back to individual UpdateDNSRecord calls.
+func buildBatchPutParam(id string, r dns.RecordResponse) (dns.BatchPutUnionParam, bool) {
+	tags := tagsFromResponse(r.Tags)
+	comment := r.Comment
+	switch r.Type {
+	case dns.RecordResponseTypeA:
+		return dns.BatchPutARecordParam{
+			ID: cloudflare.F(id),
+			ARecordParam: dns.ARecordParam{
+				Name:    cloudflare.F(r.Name),
+				TTL:     cloudflare.F(r.TTL),
+				Type:    cloudflare.F(dns.ARecordTypeA),
+				Content: cloudflare.F(r.Content),
+				Proxied: cloudflare.F(r.Proxied),
+				Comment: cloudflare.F(comment),
+				Tags:    cloudflare.F(tags),
+			},
+		}, true
+	case dns.RecordResponseTypeAAAA:
+		return dns.BatchPutAAAARecordParam{
+			ID: cloudflare.F(id),
+			AAAARecordParam: dns.AAAARecordParam{
+				Name:    cloudflare.F(r.Name),
+				TTL:     cloudflare.F(r.TTL),
+				Type:    cloudflare.F(dns.AAAARecordTypeAAAA),
+				Content: cloudflare.F(r.Content),
+				Proxied: cloudflare.F(r.Proxied),
+				Comment: cloudflare.F(comment),
+				Tags:    cloudflare.F(tags),
+			},
+		}, true
+	case dns.RecordResponseTypeCNAME:
+		return dns.BatchPutCNAMERecordParam{
+			ID: cloudflare.F(id),
+			CNAMERecordParam: dns.CNAMERecordParam{
+				Name:    cloudflare.F(r.Name),
+				TTL:     cloudflare.F(r.TTL),
+				Type:    cloudflare.F(dns.CNAMERecordTypeCNAME),
+				Content: cloudflare.F(r.Content),
+				Proxied: cloudflare.F(r.Proxied),
+				Comment: cloudflare.F(comment),
+				Tags:    cloudflare.F(tags),
+			},
+		}, true
+	case dns.RecordResponseTypeTXT:
+		return dns.BatchPutTXTRecordParam{
+			ID: cloudflare.F(id),
+			TXTRecordParam: dns.TXTRecordParam{
+				Name:    cloudflare.F(r.Name),
+				TTL:     cloudflare.F(r.TTL),
+				Type:    cloudflare.F(dns.TXTRecordTypeTXT),
+				Content: cloudflare.F(r.Content),
+				Proxied: cloudflare.F(r.Proxied),
+				Comment: cloudflare.F(comment),
+				Tags:    cloudflare.F(tags),
+			},
+		}, true
+	case dns.RecordResponseTypeMX:
+		return dns.BatchPutMXRecordParam{
+			ID: cloudflare.F(id),
+			MXRecordParam: dns.MXRecordParam{
+				Name:     cloudflare.F(r.Name),
+				TTL:      cloudflare.F(r.TTL),
+				Type:     cloudflare.F(dns.MXRecordTypeMX),
+				Content:  cloudflare.F(r.Content),
+				Proxied:  cloudflare.F(r.Proxied),
+				Comment:  cloudflare.F(comment),
+				Tags:     cloudflare.F(tags),
+				Priority: cloudflare.F(r.Priority),
+			},
+		}, true
+	case dns.RecordResponseTypeNS:
+		return dns.BatchPutNSRecordParam{
+			ID: cloudflare.F(id),
+			NSRecordParam: dns.NSRecordParam{
+				Name:    cloudflare.F(r.Name),
+				TTL:     cloudflare.F(r.TTL),
+				Type:    cloudflare.F(dns.NSRecordTypeNS),
+				Content: cloudflare.F(r.Content),
+				Proxied: cloudflare.F(r.Proxied),
+				Comment: cloudflare.F(comment),
+				Tags:    cloudflare.F(tags),
+			},
+		}, true
+	default:
+		// Record types that use structured Data fields (SRV, CAA, etc.) are not
+		// supported in the generic batch put and fall back to individual updates.
+		return nil, false
+	}
+}
+
+// buildBatchCollections classifies per-zone changes into batch collections.
+// Custom hostname side-effects are handled separately by
+// processCustomHostnameChanges before this is called.
+func (p *CloudFlareProvider) buildBatchCollections(
+	zoneID string,
+	changes []*cloudFlareChange,
+	records DNSRecordsMap,
+) batchCollections {
+	var bc batchCollections
+
+	for _, change := range changes {
+		logFields := log.Fields{
+			"record": change.ResourceRecord.Name,
+			"type":   change.ResourceRecord.Type,
+			"ttl":    change.ResourceRecord.TTL,
+			"action": change.Action.String(),
+			"zone":   zoneID,
+		}
+
+		switch change.Action {
+		case cloudFlareCreate:
+			bc.batchPosts = append(bc.batchPosts, buildBatchPostParam(change.ResourceRecord))
+			bc.createChanges = append(bc.createChanges, change)
+
+		case cloudFlareDelete:
+			recordID := p.getRecordID(records, change.ResourceRecord)
+			if recordID == "" {
+				log.WithFields(logFields).Errorf("failed to find previous record: %v", change.ResourceRecord)
+				continue
+			}
+			bc.batchDeletes = append(bc.batchDeletes, dns.RecordBatchParamsDelete{ID: cloudflare.F(recordID)})
+			bc.deleteChanges = append(bc.deleteChanges, change)
+
+		case cloudFlareUpdate:
+			recordID := p.getRecordID(records, change.ResourceRecord)
+			if recordID == "" {
+				log.WithFields(logFields).Errorf("failed to find previous record: %v", change.ResourceRecord)
+				continue
+			}
+			if putParam, ok := buildBatchPutParam(recordID, change.ResourceRecord); ok {
+				bc.batchPuts = append(bc.batchPuts, putParam)
+				bc.updateChanges = append(bc.updateChanges, change)
+			} else {
+				log.WithFields(logFields).Debugf("batch PUT not supported for type %s, using individual update", change.ResourceRecord.Type)
+				bc.fallbackUpdates = append(bc.fallbackUpdates, change)
+			}
+		}
+	}
+
+	return bc
+}
+
+// submitDNSRecordChanges submits the pre-built batch collections and any
+// fallback individual updates for a single zone. When a batch chunk fails,
+// the provider falls back to individual API calls for that chunk's changes
+// (since the batch is transactional — failure means full rollback).
+// Returns true if any operation fails.
+func (p *CloudFlareProvider) submitDNSRecordChanges(
+	ctx context.Context,
+	zoneID string,
+	bc batchCollections,
+	records DNSRecordsMap,
+) bool {
+	failed := false
+	if len(bc.batchDeletes) > 0 || len(bc.batchPosts) > 0 || len(bc.batchPuts) > 0 {
+		limit := max(p.DNSRecordsConfig.BatchChangeSize, defaultBatchChangeSize)
+		chunks := chunkBatchChanges(zoneID, bc, limit)
+		for i, chunk := range chunks {
+			log.Debugf("Submitting batch DNS records for zone %s (chunk %d/%d): %d deletes, %d creates, %d updates",
+				zoneID, i+1, len(chunks),
+				len(chunk.params.Deletes.Value),
+				len(chunk.params.Posts.Value),
+				len(chunk.params.Puts.Value),
+			)
+			if _, err := p.Client.BatchDNSRecords(ctx, chunk.params); err != nil {
+				log.Warnf("Batch DNS operation failed for zone %s (chunk %d/%d): %v — falling back to individual operations",
+					zoneID, i+1, len(chunks), convertCloudflareError(err))
+				if p.fallbackIndividualChanges(ctx, zoneID, chunk, records) {
+					failed = true
+				}
+			} else {
+				log.Debugf("Successfully submitted batch DNS records for zone %s (chunk %d/%d)", zoneID, i+1, len(chunks))
+			}
+			if i < len(chunks)-1 && p.DNSRecordsConfig.BatchChangeInterval > 0 {
+				time.Sleep(p.DNSRecordsConfig.BatchChangeInterval)
+			}
+		}
+	}
+	for _, change := range bc.fallbackUpdates {
+		logFields := log.Fields{
+			"record": change.ResourceRecord.Name,
+			"type":   change.ResourceRecord.Type,
+			"ttl":    change.ResourceRecord.TTL,
+			"action": change.Action.String(),
+			"zone":   zoneID,
+		}
+		recordID := p.getRecordID(records, change.ResourceRecord)
+		recordParam := getUpdateDNSRecordParam(zoneID, *change)
+		if _, err := p.Client.UpdateDNSRecord(ctx, recordID, recordParam); err != nil {
+			failed = true
+			log.WithFields(logFields).Errorf("failed to update record: %v", err)
+		} else {
+			log.WithFields(logFields).Debugf("individual update succeeded")
+		}
+	}
+	return failed
+}
+
+// fallbackIndividualChanges replays a failed (rolled-back) batch chunk as
+// individual API calls. Because the batch API is transactional, a failure means
+// zero state was changed in that chunk, so these individual calls are the first
+// real mutations. Individual calls return Cloudflare's own per-record error
+// details.
+//
+// Execution order matches the batch contract: deletes → updates → creates.
+// Returns true if any operation failed.
+func (p *CloudFlareProvider) fallbackIndividualChanges(
+	ctx context.Context,
+	zoneID string,
+	chunk batchChunk,
+	records DNSRecordsMap,
+) bool {
+	failed := false
+
+	// Process in batch execution order: deletes → updates → creates.
+	groups := []struct {
+		changes []*cloudFlareChange
+	}{
+		{chunk.deleteChanges},
+		{chunk.updateChanges},
+		{chunk.createChanges},
+	}
+
+	for _, group := range groups {
+		for _, change := range group.changes {
+			logFields := log.Fields{
+				"record":  change.ResourceRecord.Name,
+				"type":    change.ResourceRecord.Type,
+				"content": change.ResourceRecord.Content,
+				"action":  change.Action.String(),
+				"zone":    zoneID,
+			}
+
+			var err error
+			switch change.Action {
+			case cloudFlareCreate:
+				params := getCreateDNSRecordParam(zoneID, change)
+				_, err = p.Client.CreateDNSRecord(ctx, params)
+
+			case cloudFlareDelete:
+				recordID := p.getRecordID(records, change.ResourceRecord)
+				if recordID == "" {
+					// Record is already absent — the desired state is achieved.
+					log.WithFields(logFields).Info("fallback: record already gone, treating delete as success")
+					continue
+				}
+				err = p.Client.DeleteDNSRecord(ctx, recordID, dns.RecordDeleteParams{
+					ZoneID: cloudflare.F(zoneID),
+				})
+
+			case cloudFlareUpdate:
+				recordID := p.getRecordID(records, change.ResourceRecord)
+				if recordID == "" {
+					// Record is gone; let the next sync cycle issue a fresh CREATE.
+					log.WithFields(logFields).Info("fallback: record unexpectedly not found for update, will re-evaluate on next sync")
+					continue
+				}
+				params := getUpdateDNSRecordParam(zoneID, *change)
+				_, err = p.Client.UpdateDNSRecord(ctx, recordID, params)
+			}
+
+			if err != nil {
+				failed = true
+				log.WithFields(logFields).Errorf("fallback: individual %s failed: %v", change.Action, convertCloudflareError(err))
+			} else {
+				log.WithFields(logFields).Debugf("fallback: individual %s succeeded", change.Action)
+			}
+		}
+	}
+
+	return failed
+}
diff --git a/provider/cloudflare/cloudflare_batch_test.go b/provider/cloudflare/cloudflare_batch_test.go
new file mode 100644
index 000000000..b8113fba5
--- /dev/null
+++ b/provider/cloudflare/cloudflare_batch_test.go
@@ -0,0 +1,709 @@
+/*
+Copyright 2026 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cloudflare
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"maps"
+	"strings"
+	"testing"
+
+	"github.com/cloudflare/cloudflare-go/v5"
+	"github.com/cloudflare/cloudflare-go/v5/dns"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"sigs.k8s.io/external-dns/endpoint"
+	"sigs.k8s.io/external-dns/plan"
+)
+
+func (m *mockCloudFlareClient) BatchDNSRecords(_ context.Context, params dns.RecordBatchParams) (*dns.RecordBatchResponse, error) {
+	m.BatchDNSRecordsCalls++
+	zoneID := params.ZoneID.Value
+
+	// Snapshot zone state for transactional rollback on error.
+	// The real Cloudflare batch API is fully transactional — if any
+	// operation fails, the entire batch is rolled back.
+	var snapshot map[string]dns.RecordResponse
+	if zone, ok := m.Records[zoneID]; ok {
+		snapshot = make(map[string]dns.RecordResponse, len(zone))
+		maps.Copy(snapshot, zone)
+	}
+	actionsStart := len(m.Actions)
+
+	var firstErr error
+
+	// Process Deletes first to mirror the real API's ordering.
+	for _, del := range params.Deletes.Value {
+		recordID := del.ID.Value
+		m.Actions = append(m.Actions, MockAction{
+			Name:     "Delete",
+			ZoneId:   zoneID,
+			RecordId: recordID,
+		})
+		if zone, ok := m.Records[zoneID]; ok {
+			if rec, exists := zone[recordID]; exists {
+				name := rec.Name
+				delete(zone, recordID)
+				if strings.HasPrefix(name, "newerror-delete-") && firstErr == nil {
+					firstErr = errors.New("failed to delete erroring DNS record")
+				}
+			}
+		}
+	}
+
+	// Process Puts (updates) before Posts (creates) to mirror the real API's
+	// server-side execution order: Deletes → Patches → Puts → Posts.
+	for _, putUnion := range params.Puts.Value {
+		id, record := extractBatchPutData(putUnion)
+		m.Actions = append(m.Actions, MockAction{
+			Name:       "Update",
+			ZoneId:     zoneID,
+			RecordId:   id,
+			RecordData: record,
+		})
+		if zone, ok := m.Records[zoneID]; ok {
+			if _, exists := zone[id]; exists {
+				if strings.HasPrefix(record.Name, "newerror-update-") {
+					if firstErr == nil {
+						firstErr = errors.New("failed to update erroring DNS record")
+					}
+				} else {
+					zone[id] = record
+				}
+			}
+		}
+	}
+
+	// Process Posts (creates).
+	for _, postUnion := range params.Posts.Value {
+		post, ok := postUnion.(dns.RecordBatchParamsPost)
+		if !ok {
+			continue
+		}
+		typeStr := string(post.Type.Value)
+		record := dns.RecordResponse{
+			ID:       generateDNSRecordID(typeStr, post.Name.Value, post.Content.Value),
+			Name:     post.Name.Value,
+			TTL:      dns.TTL(post.TTL.Value),
+			Proxied:  post.Proxied.Value,
+			Type:     dns.RecordResponseType(typeStr),
+			Content:  post.Content.Value,
+			Priority: post.Priority.Value,
+		}
+		m.Actions = append(m.Actions, MockAction{
+			Name:       "Create",
+			ZoneId:     zoneID,
+			RecordId:   record.ID,
+			RecordData: record,
+		})
+		if zone, ok := m.Records[zoneID]; ok {
+			zone[record.ID] = record
+		}
+		if record.Name == "newerror.bar.com" && firstErr == nil {
+			firstErr = fmt.Errorf("failed to create record")
+		}
+	}
+
+	// Transactional: on error, rollback all state and action changes.
+	if firstErr != nil {
+		if snapshot != nil {
+			m.Records[zoneID] = snapshot
+		}
+		m.Actions = m.Actions[:actionsStart]
+		return nil, firstErr
+	}
+
+	return &dns.RecordBatchResponse{}, nil
+}
+
+// extractBatchPutData unpacks a BatchPutUnionParam into a record ID and a RecordResponse
+// suitable for recording in the mock's Actions list.
+func extractBatchPutData(put dns.BatchPutUnionParam) (string, dns.RecordResponse) {
+	switch p := put.(type) {
+	case dns.BatchPutARecordParam:
+		return p.ID.Value, dns.RecordResponse{
+			ID:      p.ID.Value,
+			Name:    p.Name.Value,
+			TTL:     p.TTL.Value,
+			Proxied: p.Proxied.Value,
+			Type:    dns.RecordResponseTypeA,
+			Content: p.Content.Value,
+		}
+	case dns.BatchPutAAAARecordParam:
+		return p.ID.Value, dns.RecordResponse{
+			ID:      p.ID.Value,
+			Name:    p.Name.Value,
+			TTL:     p.TTL.Value,
+			Proxied: p.Proxied.Value,
+			Type:    dns.RecordResponseTypeAAAA,
+			Content: p.Content.Value,
+		}
+	case dns.BatchPutCNAMERecordParam:
+		return p.ID.Value, dns.RecordResponse{
+			ID:      p.ID.Value,
+			Name:    p.Name.Value,
+			TTL:     p.TTL.Value,
+			Proxied: p.Proxied.Value,
+			Type:    dns.RecordResponseTypeCNAME,
+			Content: p.Content.Value,
+		}
+	case dns.BatchPutTXTRecordParam:
+		return p.ID.Value, dns.RecordResponse{
+			ID:      p.ID.Value,
+			Name:    p.Name.Value,
+			TTL:     p.TTL.Value,
+			Proxied: p.Proxied.Value,
+			Type:    dns.RecordResponseTypeTXT,
+			Content: p.Content.Value,
+		}
+	case dns.BatchPutMXRecordParam:
+		return p.ID.Value, dns.RecordResponse{
+			ID:       p.ID.Value,
+			Name:     p.Name.Value,
+			TTL:      p.TTL.Value,
+			Proxied:  p.Proxied.Value,
+			Type:     dns.RecordResponseTypeMX,
+			Content:  p.Content.Value,
+			Priority: p.Priority.Value,
+		}
+	case dns.BatchPutNSRecordParam:
+		return p.ID.Value, dns.RecordResponse{
+			ID:      p.ID.Value,
+			Name:    p.Name.Value,
+			TTL:     p.TTL.Value,
+			Proxied: p.Proxied.Value,
+			Type:    dns.RecordResponseTypeNS,
+			Content: p.Content.Value,
+		}
+	default:
+		panic(fmt.Sprintf("extractBatchPutData: unexpected BatchPutUnionParam type %T", put))
+	}
+}
+
+// generateDNSRecordID builds the deterministic record ID used by the mock client.
+func generateDNSRecordID(rrtype string, name string, content string) string {
+	return fmt.Sprintf("%s-%s-%s", name, rrtype, content)
+}
+
+func TestBatchFallbackIndividual(t *testing.T) {
+	t.Run("batch failure falls back to individual operations", func(t *testing.T) {
+		// Create a provider with pre-existing records.
+		client := NewMockCloudFlareClientWithRecords(map[string][]dns.RecordResponse{
+			"001": {
+				{ID: "existing-1", Name: "ok.bar.com", Type: "A", Content: "1.2.3.4", TTL: 120},
+			},
+		})
+		p := &CloudFlareProvider{
+			Client: client,
+		}
+
+		// Apply changes that include a good create and a bad create.
+		// "newerror.bar.com" triggers a batch failure in the mock BatchDNSRecords,
+		// then an individual fallback failure in CreateDNSRecord.
+		changes := &plan.Changes{
+			Create: []*endpoint.Endpoint{
+				{DNSName: "good.bar.com", Targets: endpoint.Targets{"5.6.7.8"}, RecordType: "A"},
+				{DNSName: "newerror.bar.com", Targets: endpoint.Targets{"9.10.11.12"}, RecordType: "A"},
+			},
+		}
+
+		err := p.ApplyChanges(t.Context(), changes)
+		require.Error(t, err, "should return error when individual fallback has failures")
+		assert.Equal(t, 1, client.BatchDNSRecordsCalls, "batch path should be attempted before fallback")
+
+		// The batch should have failed (because of newerror.bar.com), then
+		// fallback should have applied "good.bar.com" individually (success)
+		// and "newerror.bar.com" individually (failure).
+
+		// Verify the good record was created via individual fallback.
+		zone001 := client.Records["001"]
+		goodID := generateDNSRecordID("A", "good.bar.com", "5.6.7.8")
+		assert.Contains(t, zone001, goodID, "good record should exist after individual fallback")
+	})
+
+	t.Run("failed individual delete is reported", func(t *testing.T) {
+		// When a batch containing two deletes fails, the fallback replays them
+		// individually. The one that ultimately fails should be reported;
+		// the one that succeeds should not block the overall zone from converging.
+		client := NewMockCloudFlareClientWithRecords(map[string][]dns.RecordResponse{
+			"001": {
+				{ID: "del-ok", Name: "deleteme.bar.com", Type: "A", Content: "1.2.3.4", TTL: 120},
+				{ID: "del-err", Name: "newerror-delete-1.bar.com", Type: "A", Content: "5.6.7.8", TTL: 120},
+			},
+		})
+		p := &CloudFlareProvider{
+			Client: client,
+		}
+
+		changes := &plan.Changes{
+			Delete: []*endpoint.Endpoint{
+				{DNSName: "deleteme.bar.com", Targets: endpoint.Targets{"1.2.3.4"}, RecordType: "A"},
+				{DNSName: "newerror-delete-1.bar.com", Targets: endpoint.Targets{"5.6.7.8"}, RecordType: "A"},
+			},
+		}
+		err := p.ApplyChanges(t.Context(), changes)
+		require.Error(t, err, "should return error for the failing delete")
+
+		// The good delete should have succeeded via individual fallback.
+		assert.NotContains(t, client.Records["001"], "del-ok", "successfully deleted record should be gone")
+	})
+
+	t.Run("fallback update failure is reported", func(t *testing.T) {
+		client := NewMockCloudFlareClientWithRecords(map[string][]dns.RecordResponse{
+			"001": {
+				{ID: "upd-err", Name: "newerror-update-1.bar.com", Type: "A", Content: "1.2.3.4", TTL: 120},
+			},
+		})
+		p := &CloudFlareProvider{
+			Client: client,
+		}
+
+		changes := &plan.Changes{
+			UpdateNew: []*endpoint.Endpoint{
+				{DNSName: "newerror-update-1.bar.com", Targets: endpoint.Targets{"1.2.3.4"}, RecordType: "A", RecordTTL: 300},
+			},
+			UpdateOld: []*endpoint.Endpoint{
+				{DNSName: "newerror-update-1.bar.com", Targets: endpoint.Targets{"1.2.3.4"}, RecordType: "A", RecordTTL: 120},
+			},
+		}
+		err := p.ApplyChanges(t.Context(), changes)
+		require.Error(t, err, "should return error for the failing update")
+	})
+}
+
+func TestChunkBatchChanges(t *testing.T) {
+	// Build sample changes and batch params.
+	mkDelete := func(id string) dns.RecordBatchParamsDelete {
+		return dns.RecordBatchParamsDelete{ID: cloudflare.F(id)}
+	}
+	mkPost := func(name, content string) dns.RecordBatchParamsPostUnion {
+		return dns.RecordBatchParamsPost{
+			Name:    cloudflare.F(name),
+			Type:    cloudflare.F(dns.RecordBatchParamsPostsTypeA),
+			Content: cloudflare.F(content),
+		}
+	}
+	mkPut := func(id, name, content string) dns.BatchPutUnionParam {
+		return dns.BatchPutARecordParam{
+			ID: cloudflare.F(id),
+			ARecordParam: dns.ARecordParam{
+				Name:    cloudflare.F(name),
+				Type:    cloudflare.F(dns.ARecordTypeA),
+				Content: cloudflare.F(content),
+			},
+		}
+	}
+	mkChange := func(action changeAction, name, content string) *cloudFlareChange {
+		return &cloudFlareChange{
+			Action:         action,
+			ResourceRecord: dns.RecordResponse{Name: name, Type: "A", Content: content},
+		}
+	}
+
+	deletes := []dns.RecordBatchParamsDelete{mkDelete("d1"), mkDelete("d2")}
+	deleteChanges := []*cloudFlareChange{
+		mkChange(cloudFlareDelete, "del1.bar.com", "1.1.1.1"),
+		mkChange(cloudFlareDelete, "del2.bar.com", "2.2.2.2"),
+	}
+	posts := []dns.RecordBatchParamsPostUnion{mkPost("create1.bar.com", "3.3.3.3")}
+	createChanges := []*cloudFlareChange{
+		mkChange(cloudFlareCreate, "create1.bar.com", "3.3.3.3"),
+	}
+	puts := []dns.BatchPutUnionParam{mkPut("u1", "update1.bar.com", "4.4.4.4")}
+	updateChanges := []*cloudFlareChange{
+		mkChange(cloudFlareUpdate, "update1.bar.com", "4.4.4.4"),
+	}
+
+	t.Run("single chunk when under limit", func(t *testing.T) {
+		bc := batchCollections{
+			batchDeletes:  deletes,
+			deleteChanges: deleteChanges,
+			batchPosts:    posts,
+			createChanges: createChanges,
+			batchPuts:     puts,
+			updateChanges: updateChanges,
+		}
+		chunks := chunkBatchChanges("zone1", bc, 10)
+		require.Len(t, chunks, 1)
+		assert.Len(t, chunks[0].deleteChanges, 2)
+		assert.Len(t, chunks[0].createChanges, 1)
+		assert.Len(t, chunks[0].updateChanges, 1)
+	})
+
+	t.Run("splits into multiple chunks at limit", func(t *testing.T) {
+		bc := batchCollections{
+			batchDeletes:  deletes,
+			deleteChanges: deleteChanges,
+			batchPosts:    posts,
+			createChanges: createChanges,
+			batchPuts:     puts,
+			updateChanges: updateChanges,
+		}
+		chunks := chunkBatchChanges("zone1", bc, 2)
+		require.Len(t, chunks, 2)
+		// First chunk: 2 deletes (fills limit)
+		assert.Len(t, chunks[0].deleteChanges, 2)
+		assert.Empty(t, chunks[0].updateChanges)
+		assert.Empty(t, chunks[0].createChanges)
+		// Second chunk: 1 put then 1 post
+		assert.Empty(t, chunks[1].deleteChanges)
+		assert.Len(t, chunks[1].updateChanges, 1)
+		assert.Len(t, chunks[1].createChanges, 1)
+	})
+
+	t.Run("preserves operation order across chunk boundaries", func(t *testing.T) {
+		bc := batchCollections{
+			batchDeletes: []dns.RecordBatchParamsDelete{mkDelete("d1")},
+			deleteChanges: []*cloudFlareChange{
+				mkChange(cloudFlareDelete, "del1.bar.com", "1.1.1.1"),
+			},
+			batchPuts: []dns.BatchPutUnionParam{
+				mkPut("u1", "update1.bar.com", "2.2.2.2"),
+				mkPut("u2", "update2.bar.com", "3.3.3.3"),
+			},
+			updateChanges: []*cloudFlareChange{
+				mkChange(cloudFlareUpdate, "update1.bar.com", "2.2.2.2"),
+				mkChange(cloudFlareUpdate, "update2.bar.com", "3.3.3.3"),
+			},
+			batchPosts: []dns.RecordBatchParamsPostUnion{
+				mkPost("create1.bar.com", "4.4.4.4"),
+				mkPost("create2.bar.com", "5.5.5.5"),
+			},
+			createChanges: []*cloudFlareChange{
+				mkChange(cloudFlareCreate, "create1.bar.com", "4.4.4.4"),
+				mkChange(cloudFlareCreate, "create2.bar.com", "5.5.5.5"),
+			},
+		}
+
+		chunks := chunkBatchChanges("zone1", bc, 2)
+		require.Len(t, chunks, 3)
+
+		assert.Len(t, chunks[0].deleteChanges, 1)
+		assert.Len(t, chunks[0].updateChanges, 1)
+		assert.Empty(t, chunks[0].createChanges)
+
+		assert.Empty(t, chunks[1].deleteChanges)
+		assert.Len(t, chunks[1].updateChanges, 1)
+		assert.Len(t, chunks[1].createChanges, 1)
+
+		assert.Empty(t, chunks[2].deleteChanges)
+		assert.Empty(t, chunks[2].updateChanges)
+		assert.Len(t, chunks[2].createChanges, 1)
+	})
+}
+
+func TestTagsFromResponse(t *testing.T) {
+	t.Run("nil input returns nil", func(t *testing.T) {
+		assert.Nil(t, tagsFromResponse(nil))
+	})
+	t.Run("non-string-slice returns nil", func(t *testing.T) {
+		assert.Nil(t, tagsFromResponse(42))
+	})
+	t.Run("string slice is returned unchanged", func(t *testing.T) {
+		tags := []string{"tag1", "tag2"}
+		assert.Equal(t, tags, tagsFromResponse(tags))
+	})
+}
+
+func TestBuildBatchPutParam(t *testing.T) {
+	base := dns.RecordResponse{
+		Name:    "example.bar.com",
+		TTL:     120,
+		Proxied: false,
+		Comment: "test-comment",
+	}
+
+	t.Run("AAAA record", func(t *testing.T) {
+		r := base
+		r.Type = dns.RecordResponseTypeAAAA
+		r.Content = "2001:db8::1"
+		param, ok := buildBatchPutParam("id-aaaa", r)
+		require.True(t, ok)
+		p, cast := param.(dns.BatchPutAAAARecordParam)
+		require.True(t, cast)
+		assert.Equal(t, "id-aaaa", p.ID.Value)
+		assert.Equal(t, "2001:db8::1", p.Content.Value)
+		assert.Equal(t, dns.AAAARecordTypeAAAA, p.Type.Value)
+	})
+
+	t.Run("CNAME record", func(t *testing.T) {
+		r := base
+		r.Type = dns.RecordResponseTypeCNAME
+		r.Content = "target.bar.com"
+		param, ok := buildBatchPutParam("id-cname", r)
+		require.True(t, ok)
+		p, cast := param.(dns.BatchPutCNAMERecordParam)
+		require.True(t, cast)
+		assert.Equal(t, "id-cname", p.ID.Value)
+		assert.Equal(t, "target.bar.com", p.Content.Value)
+		assert.Equal(t, dns.CNAMERecordTypeCNAME, p.Type.Value)
+	})
+
+	t.Run("TXT record", func(t *testing.T) {
+		r := base
+		r.Type = dns.RecordResponseTypeTXT
+		r.Content = "v=spf1 include:example.com ~all"
+		param, ok := buildBatchPutParam("id-txt", r)
+		require.True(t, ok)
+		p, cast := param.(dns.BatchPutTXTRecordParam)
+		require.True(t, cast)
+		assert.Equal(t, "id-txt", p.ID.Value)
+		assert.Equal(t, dns.TXTRecordTypeTXT, p.Type.Value)
+	})
+
+	t.Run("MX record with priority", func(t *testing.T) {
+		r := base
+		r.Type = dns.RecordResponseTypeMX
+		r.Content = "mail.example.com"
+		r.Priority = 10
+		param, ok := buildBatchPutParam("id-mx", r)
+		require.True(t, ok)
+		p, cast := param.(dns.BatchPutMXRecordParam)
+		require.True(t, cast)
+		assert.Equal(t, "id-mx", p.ID.Value)
+		assert.InDelta(t, float64(10), float64(p.Priority.Value), 0)
+		assert.Equal(t, dns.MXRecordTypeMX, p.Type.Value)
+	})
+
+	t.Run("NS record", func(t *testing.T) {
+		r := base
+		r.Type = dns.RecordResponseTypeNS
+		r.Content = "ns1.example.com"
+		param, ok := buildBatchPutParam("id-ns", r)
+		require.True(t, ok)
+		p, cast := param.(dns.BatchPutNSRecordParam)
+		require.True(t, cast)
+		assert.Equal(t, "id-ns", p.ID.Value)
+		assert.Equal(t, dns.NSRecordTypeNS, p.Type.Value)
+	})
+
+	t.Run("SRV record falls back (returns nil, false)", func(t *testing.T) {
+		r := base
+		r.Type = dns.RecordResponseTypeSRV
+		r.Content = "10 20 443 target.bar.com"
+		param, ok := buildBatchPutParam("id-srv", r)
+		assert.False(t, ok)
+		assert.Nil(t, param)
+	})
+
+	t.Run("CAA record falls back (returns nil, false)", func(t *testing.T) {
+		r := base
+		r.Type = dns.RecordResponseTypeCAA
+		r.Content = "0 issue letsencrypt.org"
+		param, ok := buildBatchPutParam("id-caa", r)
+		assert.False(t, ok)
+		assert.Nil(t, param)
+	})
+}
+
+func TestBuildBatchCollections_EdgeCases(t *testing.T) {
+	p := &CloudFlareProvider{}
+
+	t.Run("update with missing record ID is skipped", func(t *testing.T) {
+		changes := []*cloudFlareChange{
+			{
+				Action: cloudFlareUpdate,
+				ResourceRecord: dns.RecordResponse{
+					Name:    "missing.bar.com",
+					Type:    dns.RecordResponseTypeA,
+					Content: "1.2.3.4",
+				},
+			},
+		}
+		// Empty records map — getRecordID will return ""
+		bc := p.buildBatchCollections("zone1", changes, make(DNSRecordsMap))
+		assert.Empty(t, bc.batchPuts, "missing record should not be added to batch puts")
+		assert.Empty(t, bc.updateChanges)
+		assert.Empty(t, bc.fallbackUpdates)
+	})
+
+	t.Run("SRV update goes to fallbackUpdates", func(t *testing.T) {
+		srvRecord := dns.RecordResponse{
+			ID:      "srv-1",
+			Name:    "srv.bar.com",
+			Type:    dns.RecordResponseTypeSRV,
+			Content: "10 20 443 target.bar.com",
+		}
+		records := DNSRecordsMap{
+			newDNSRecordIndex(srvRecord): srvRecord,
+		}
+		changes := []*cloudFlareChange{
+			{
+				Action:         cloudFlareUpdate,
+				ResourceRecord: srvRecord,
+			},
+		}
+		bc := p.buildBatchCollections("zone1", changes, records)
+		assert.Empty(t, bc.batchPuts, "SRV should not be in batch puts")
+		assert.Empty(t, bc.updateChanges)
+		require.Len(t, bc.fallbackUpdates, 1)
+		assert.Equal(t, "srv.bar.com", bc.fallbackUpdates[0].ResourceRecord.Name)
+	})
+
+	t.Run("delete with missing record ID is skipped", func(t *testing.T) {
+		changes := []*cloudFlareChange{
+			{
+				Action: cloudFlareDelete,
+				ResourceRecord: dns.RecordResponse{
+					Name:    "gone.bar.com",
+					Type:    dns.RecordResponseTypeA,
+					Content: "1.2.3.4",
+				},
+			},
+		}
+		bc := p.buildBatchCollections("zone1", changes, make(DNSRecordsMap))
+		assert.Empty(t, bc.batchDeletes, "missing record should not be added to batch deletes")
+		assert.Empty(t, bc.deleteChanges)
+	})
+}
+
+func TestSubmitDNSRecordChanges_BatchInterval(t *testing.T) {
+	// Build 201 creates so they span 2 chunks (defaultBatchChangeSize=200),
+	// triggering the time.Sleep(BatchChangeInterval) code path between chunks.
+	client := NewMockCloudFlareClientWithRecords(map[string][]dns.RecordResponse{
+		"001": {},
+	})
+	p := &CloudFlareProvider{
+		Client: client,
+		DNSRecordsConfig: DNSRecordsConfig{
+			BatchChangeInterval: 1, // 1 nanosecond — non-zero triggers sleep
+		},
+	}
+
+	const nRecords = defaultBatchChangeSize + 1
+	var posts []dns.RecordBatchParamsPostUnion
+	var createChanges []*cloudFlareChange
+	for i := range nRecords {
+		name := fmt.Sprintf("record%d.bar.com", i)
+		posts = append(posts, dns.RecordBatchParamsPost{
+			Name:    cloudflare.F(name),
+			Type:    cloudflare.F(dns.RecordBatchParamsPostsTypeA),
+			Content: cloudflare.F("1.2.3.4"),
+		})
+		createChanges = append(createChanges, &cloudFlareChange{
+			Action:         cloudFlareCreate,
+			ResourceRecord: dns.RecordResponse{Name: name, Type: "A", Content: "1.2.3.4"},
+		})
+	}
+
+	bc := batchCollections{
+		batchPosts:    posts,
+		createChanges: createChanges,
+	}
+
+	failed := p.submitDNSRecordChanges(t.Context(), "001", bc, make(DNSRecordsMap))
+	assert.False(t, failed, "should not fail")
+	assert.Equal(t, 2, client.BatchDNSRecordsCalls, "two chunks should require two batch API calls")
+}
+
+func TestSubmitDNSRecordChanges_FallbackUpdates(t *testing.T) {
+	t.Run("successful SRV fallback update", func(t *testing.T) {
+		srvRecord := dns.RecordResponse{
+			ID:      "srv-1",
+			Name:    "srv.bar.com",
+			Type:    dns.RecordResponseTypeSRV,
+			Content: "10 20 443 target.bar.com",
+		}
+		client := NewMockCloudFlareClientWithRecords(map[string][]dns.RecordResponse{
+			"001": {srvRecord},
+		})
+		p := &CloudFlareProvider{Client: client}
+
+		records := DNSRecordsMap{
+			newDNSRecordIndex(srvRecord): srvRecord,
+		}
+		bc := batchCollections{
+			fallbackUpdates: []*cloudFlareChange{
+				{Action: cloudFlareUpdate, ResourceRecord: srvRecord},
+			},
+		}
+
+		failed := p.submitDNSRecordChanges(t.Context(), "001", bc, records)
+		assert.False(t, failed, "successful SRV fallback update should not report failure")
+		assert.Equal(t, 0, client.BatchDNSRecordsCalls, "batch API not called for fallback-only changes")
+	})
+
+	t.Run("failed SRV fallback update is reported", func(t *testing.T) {
+		srvRecord := dns.RecordResponse{
+			ID:      "newerror-upd-srv",
+			Name:    "newerror-update-srv.bar.com",
+			Type:    dns.RecordResponseTypeSRV,
+			Content: "10 20 443 target.bar.com",
+		}
+		client := NewMockCloudFlareClientWithRecords(map[string][]dns.RecordResponse{
+			"001": {srvRecord},
+		})
+		p := &CloudFlareProvider{Client: client}
+
+		records := DNSRecordsMap{
+			newDNSRecordIndex(srvRecord): srvRecord,
+		}
+		bc := batchCollections{
+			fallbackUpdates: []*cloudFlareChange{
+				{Action: cloudFlareUpdate, ResourceRecord: srvRecord},
+			},
+		}
+
+		failed := p.submitDNSRecordChanges(t.Context(), "001", bc, records)
+		assert.True(t, failed, "failed SRV fallback update should be reported")
+	})
+}
+
+func TestFallbackIndividualChanges_MissingRecord(t *testing.T) {
+	client := NewMockCloudFlareClientWithRecords(map[string][]dns.RecordResponse{
+		"001": {},
+	})
+	p := &CloudFlareProvider{Client: client}
+	emptyRecords := make(DNSRecordsMap)
+
+	t.Run("delete where record is already gone succeeds silently", func(t *testing.T) {
+		chunk := batchChunk{
+			deleteChanges: []*cloudFlareChange{
+				{
+					Action: cloudFlareDelete,
+					ResourceRecord: dns.RecordResponse{
+						Name:    "gone.bar.com",
+						Type:    dns.RecordResponseTypeA,
+						Content: "1.2.3.4",
+					},
+				},
+			},
+		}
+		failed := p.fallbackIndividualChanges(t.Context(), "001", chunk, emptyRecords)
+		assert.False(t, failed, "delete of already-absent record should not report failure")
+	})
+
+	t.Run("update where record is not found skips gracefully", func(t *testing.T) {
+		chunk := batchChunk{
+			updateChanges: []*cloudFlareChange{
+				{
+					Action: cloudFlareUpdate,
+					ResourceRecord: dns.RecordResponse{
+						Name:    "missing.bar.com",
+						Type:    dns.RecordResponseTypeA,
+						Content: "1.2.3.4",
+					},
+				},
+			},
+		}
+		failed := p.fallbackIndividualChanges(t.Context(), "001", chunk, emptyRecords)
+		assert.False(t, failed, "update of missing record should not report failure")
+	})
+}
diff --git a/provider/cloudflare/cloudflare_custom_hostnames.go b/provider/cloudflare/cloudflare_custom_hostnames.go
index 940e89ec3..08320be76 100644
--- a/provider/cloudflare/cloudflare_custom_hostnames.go
+++ b/provider/cloudflare/cloudflare_custom_hostnames.go
@@ -277,6 +277,30 @@ func (p *CloudFlareProvider) listCustomHostnamesWithPagination(ctx context.Conte
 	return chs, nil
 }
 
+// processCustomHostnameChanges applies custom hostname side-effects for each
+// change in the set and returns true if any operation failed.
+func (p *CloudFlareProvider) processCustomHostnameChanges(
+	ctx context.Context,
+	zoneID string,
+	changes []*cloudFlareChange,
+	chs customHostnamesMap,
+) bool {
+	failed := false
+	for _, change := range changes {
+		logFields := log.Fields{
+			"record": change.ResourceRecord.Name,
+			"type":   change.ResourceRecord.Type,
+			"ttl":    change.ResourceRecord.TTL,
+			"action": change.Action.String(),
+			"zone":   zoneID,
+		}
+		if !p.submitCustomHostnameChanges(ctx, zoneID, change, chs, logFields) {
+			failed = true
+		}
+	}
+	return failed
+}
+
 // listAllCustomHostnames extracts all custom hostnames from the iterator
 func listAllCustomHostnames(iter autoPager[custom_hostnames.CustomHostnameListResponse]) ([]customHostname, error) {
 	var customHostnames []customHostname
diff --git a/provider/cloudflare/cloudflare_regional_test.go b/provider/cloudflare/cloudflare_regional_test.go
index e6ddacc9e..7a69672f3 100644
--- a/provider/cloudflare/cloudflare_regional_test.go
+++ b/provider/cloudflare/cloudflare_regional_test.go
@@ -28,6 +28,7 @@ import (
 	log "github.com/sirupsen/logrus"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"sigs.k8s.io/external-dns/endpoint"
 	"sigs.k8s.io/external-dns/internal/testutils"
@@ -1294,3 +1295,89 @@ func TestCloudflareAdjustEndpointsRegionalServices(t *testing.T) {
 		})
 	}
 }
+
+// TestSubmitChanges_DryRun_RegionalErrors covers error paths in the dry-run branch of
+// submitChanges. Two statements in that branch are not covered by any test:
+//
+//   - the `failedChange = true` body inside `if !p.submitRegionalHostnameChanges(...)`
+//   - the subsequent `failedZones = append(...)` when failedChange is true
+//
+// Both are unreachable because submitRegionalHostnameChange unconditionally returns true
+// when DryRun=true (it logs and returns before making any API call), so the failure
+// branch can never be entered without changing the production code.
+func TestSubmitChanges_DryRun_RegionalErrors(t *testing.T) {
+	t.Run("desiredRegionalHostnames conflict returns error", func(t *testing.T) {
+		// Two changes for the same hostname with different region keys →
+		// desiredRegionalHostnames returns a conflict error.
+		client := NewMockCloudFlareClient()
+		p := &CloudFlareProvider{
+			Client: client,
+			DryRun: true,
+			RegionalServicesConfig: RegionalServicesConfig{
+				Enabled: true,
+			},
+		}
+
+		// Build conflicting cloudFlareChanges directly and call submitChanges,
+		// which is in the same package.
+		changes := []*cloudFlareChange{
+			{
+				Action:         cloudFlareCreate,
+				ResourceRecord: dns.RecordResponse{Name: "foo.bar.com", Type: "A", Content: "1.2.3.4"},
+				RegionalHostname: regionalHostname{
+					hostname:  "foo.bar.com",
+					regionKey: "us",
+				},
+			},
+			{
+				Action:         cloudFlareUpdate,
+				ResourceRecord: dns.RecordResponse{Name: "foo.bar.com", Type: "A", Content: "1.2.3.4"},
+				RegionalHostname: regionalHostname{
+					hostname:  "foo.bar.com",
+					regionKey: "eu", // different from "us" → conflict
+				},
+			},
+		}
+		err := p.submitChanges(t.Context(), changes)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "failed to build desired regional hostnames")
+	})
+
+	t.Run("listDataLocalisationRegionalHostnames error in dry-run returns error", func(t *testing.T) {
+		// Zone ID containing "rherror" causes the mock to return an error from
+		// ListDataLocalizationRegionalHostnames.
+		client := &mockCloudFlareClient{
+			Zones: map[string]string{
+				"rherror-zone1": "rherror.bar.com",
+			},
+			Records: map[string]map[string]dns.RecordResponse{
+				"rherror-zone1": {},
+			},
+			customHostnames:   map[string][]customHostname{},
+			regionalHostnames: map[string][]regionalHostname{},
+		}
+		p := &CloudFlareProvider{
+			Client: client,
+			DryRun: true,
+			RegionalServicesConfig: RegionalServicesConfig{
+				Enabled:   true,
+				RegionKey: "us",
+			},
+			domainFilter: endpoint.NewDomainFilter([]string{"rherror.bar.com"}),
+		}
+
+		changes := []*cloudFlareChange{
+			{
+				Action:         cloudFlareCreate,
+				ResourceRecord: dns.RecordResponse{Name: "foo.rherror.bar.com", Type: "A", Content: "1.2.3.4"},
+				RegionalHostname: regionalHostname{
+					hostname:  "foo.rherror.bar.com",
+					regionKey: "us",
+				},
+			},
+		}
+		err := p.submitChanges(t.Context(), changes)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "could not fetch regional hostnames from zone")
+	})
+}
diff --git a/provider/cloudflare/cloudflare_test.go b/provider/cloudflare/cloudflare_test.go
index 05d75fdb8..35ba44eb9 100644
--- a/provider/cloudflare/cloudflare_test.go
+++ b/provider/cloudflare/cloudflare_test.go
@@ -18,8 +18,10 @@ package cloudflare
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -29,6 +31,7 @@ import (
 
 	"github.com/cloudflare/cloudflare-go/v5"
 	"github.com/cloudflare/cloudflare-go/v5/dns"
+	"github.com/cloudflare/cloudflare-go/v5/option"
 	"github.com/cloudflare/cloudflare-go/v5/zones"
 	"github.com/maxatome/go-testdeep/td"
 	log "github.com/sirupsen/logrus"
@@ -100,6 +103,7 @@ type mockCloudFlareClient struct {
 	Zones                map[string]string
 	Records              map[string]map[string]dns.RecordResponse
 	Actions              []MockAction
+	BatchDNSRecordsCalls int
 	listZonesError       error // For v4 ListZones
 	getZoneError         error // For v4 GetZone
 	dnsRecordsError      error
@@ -1680,19 +1684,6 @@ func TestCloudflareComplexUpdate(t *testing.T) {
 			ZoneId:   "001",
 			RecordId: "2345678901",
 		},
-		{
-			Name:     "Create",
-			ZoneId:   "001",
-			RecordId: generateDNSRecordID("A", "foobar.bar.com", "2.3.4.5"),
-			RecordData: dns.RecordResponse{
-				ID:      generateDNSRecordID("A", "foobar.bar.com", "2.3.4.5"),
-				Name:    "foobar.bar.com",
-				Type:    "A",
-				Content: "2.3.4.5",
-				TTL:     1,
-				Proxied: true,
-			},
-		},
 		{
 			Name:     "Update",
 			ZoneId:   "001",
@@ -1706,6 +1697,19 @@ func TestCloudflareComplexUpdate(t *testing.T) {
 				Proxied: true,
 			},
 		},
+		{
+			Name:     "Create",
+			ZoneId:   "001",
+			RecordId: generateDNSRecordID("A", "foobar.bar.com", "2.3.4.5"),
+			RecordData: dns.RecordResponse{
+				ID:      generateDNSRecordID("A", "foobar.bar.com", "2.3.4.5"),
+				Name:    "foobar.bar.com",
+				Type:    "A",
+				Content: "2.3.4.5",
+				TTL:     1,
+				Proxied: true,
+			},
+		},
 	})
 }
 
@@ -2627,6 +2631,24 @@ func TestConvertCloudflareError(t *testing.T) {
 			expectSoftError: true,
 			description:     "Server error (503) should be converted to soft error",
 		},
+		{
+			name:            "io.ErrUnexpectedEOF is soft",
+			inputError:      io.ErrUnexpectedEOF,
+			expectSoftError: true,
+			description:     "Unexpected EOF (connection closed mid-response) should be converted to soft error",
+		},
+		{
+			name:            "io.EOF is soft",
+			inputError:      io.EOF,
+			expectSoftError: true,
+			description:     "EOF (connection closed before response) should be converted to soft error",
+		},
+		{
+			name:            "wrapped io.ErrUnexpectedEOF is soft",
+			inputError:      fmt.Errorf("transport error: %w", io.ErrUnexpectedEOF),
+			expectSoftError: true,
+			description:     "Wrapped unexpected EOF should be converted to soft error",
+		},
 		{
 			name:            "Rate limit string error",
 			inputError:      errors.New("exceeded available rate limit retries"),
@@ -2980,8 +3002,226 @@ func TestZoneService(t *testing.T) {
 		err := client.CreateCustomHostname(ctx, zoneID, customHostname{})
 		assert.ErrorIs(t, err, context.Canceled)
 	})
+
+	t.Run("BatchDNSRecords", func(t *testing.T) {
+		t.Parallel()
+		_, err := client.BatchDNSRecords(ctx, dns.RecordBatchParams{ZoneID: cloudflare.F(zoneID)})
+		assert.ErrorIs(t, err, context.Canceled)
+	})
 }
 
-func generateDNSRecordID(rrtype string, name string, content string) string {
-	return fmt.Sprintf("%s-%s-%s", name, rrtype, content)
+func TestSubmitChanges_ErrorPaths(t *testing.T) {
+	t.Run("getDNSRecordsMap error returns error from submitChanges", func(t *testing.T) {
+		client := NewMockCloudFlareClient()
+		client.dnsRecordsError = errors.New("dns list failed")
+		p := &CloudFlareProvider{Client: client}
+
+		changes := &plan.Changes{
+			Create: []*endpoint.Endpoint{
+				{DNSName: "test.bar.com", Targets: endpoint.Targets{"1.2.3.4"}, RecordType: "A"},
+			},
+		}
+		err := p.ApplyChanges(t.Context(), changes)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "could not fetch records from zone")
+	})
+
+	t.Run("listCustomHostnamesWithPagination error returns error from submitChanges", func(t *testing.T) {
+		// The mock returns an error for CustomHostnames() when zoneID starts with "newerror-".
+		// CustomHostnamesConfig.Enabled must be true to reach that code path.
+		client := &mockCloudFlareClient{
+			Zones: map[string]string{
+				"newerror-zone1": "errorcf.com",
+			},
+			Records: map[string]map[string]dns.RecordResponse{
+				"newerror-zone1": {},
+			},
+			customHostnames:   map[string][]customHostname{},
+			regionalHostnames: map[string][]regionalHostname{},
+		}
+		p := &CloudFlareProvider{
+			Client:                client,
+			domainFilter:          endpoint.NewDomainFilter([]string{"errorcf.com"}),
+			CustomHostnamesConfig: CustomHostnamesConfig{Enabled: true},
+		}
+
+		changes := &plan.Changes{
+			Create: []*endpoint.Endpoint{
+				{DNSName: "sub.errorcf.com", Targets: endpoint.Targets{"1.2.3.4"}, RecordType: "A"},
+			},
+		}
+		err := p.ApplyChanges(t.Context(), changes)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "could not fetch custom hostnames from zone")
+	})
+
+	t.Run("processCustomHostnameChanges failure sets failedChange", func(t *testing.T) {
+		// The mock's CreateCustomHostname fails for "newerror-create.foo.fancybar.com".
+		// With CustomHostnames enabled, the failing create causes processCustomHostnameChanges
+		// to return true, which sets failedChange=true for the zone.
+		client := NewMockCloudFlareClient()
+		p := &CloudFlareProvider{
+			Client:                client,
+			CustomHostnamesConfig: CustomHostnamesConfig{Enabled: true},
+		}
+
+		changes := &plan.Changes{
+			Create: []*endpoint.Endpoint{
+				{
+					DNSName:    "a.bar.com",
+					Targets:    endpoint.Targets{"1.2.3.4"},
+					RecordType: "A",
+					ProviderSpecific: endpoint.ProviderSpecific{
+						{
+							Name:  "external-dns.alpha.kubernetes.io/cloudflare-custom-hostname",
+							Value: "newerror-create.foo.fancybar.com",
+						},
+					},
+				},
+			},
+		}
+		err := p.ApplyChanges(t.Context(), changes)
+		require.Error(t, err, "failing custom hostname create should cause an error")
+	})
+
+	t.Run("Zones error propagates from submitChanges", func(t *testing.T) {
+		// Setting listZonesError causes p.Zones() to fail inside submitChanges,
+		// exercising the `if err != nil { return err }` block at the top of the loop.
+		client := NewMockCloudFlareClient()
+		client.listZonesError = errors.New("zones fetch failed")
+		p := &CloudFlareProvider{Client: client}
+
+		changes := &plan.Changes{
+			Create: []*endpoint.Endpoint{
+				{DNSName: "test.bar.com", Targets: endpoint.Targets{"1.2.3.4"}, RecordType: "A"},
+			},
+		}
+		err := p.ApplyChanges(t.Context(), changes)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "zones fetch failed")
+	})
+}
+
+func TestParseTagsAnnotation(t *testing.T) {
+	t.Run("parses comma-separated tags", func(t *testing.T) {
+		tags := parseTagsAnnotation("tag1,tag2,tag3")
+		assert.Equal(t, []string{"tag1", "tag2", "tag3"}, tags)
+	})
+	t.Run("trims whitespace from each tag", func(t *testing.T) {
+		tags := parseTagsAnnotation("  z-tag ,  a-tag  ")
+		assert.Equal(t, []string{"a-tag", "z-tag"}, tags)
+	})
+	t.Run("sorts tags canonically", func(t *testing.T) {
+		tags := parseTagsAnnotation("c,a,b")
+		assert.Equal(t, []string{"a", "b", "c"}, tags)
+	})
+	t.Run("skips empty tokens", func(t *testing.T) {
+		tags := parseTagsAnnotation("tag1,,,, tag2")
+		assert.Equal(t, []string{"tag1", "tag2"}, tags)
+	})
+}
+
+func TestAdjustEndpoints_TagsAnnotation(t *testing.T) {
+	// parseTagsAnnotation is only invoked when the CloudflareTagsKey annotation
+	// is present on the endpoint. This test exercises that branch via AdjustEndpoints.
+	p := &CloudFlareProvider{}
+	ep := &endpoint.Endpoint{
+		RecordType: "A",
+		DNSName:    "test.bar.com",
+		Targets:    endpoint.Targets{"1.2.3.4"},
+		ProviderSpecific: endpoint.ProviderSpecific{
+			{
+				Name:  annotations.CloudflareTagsKey,
+				Value: "beta, alpha, gamma",
+			},
+		},
+	}
+	adjusted, err := p.AdjustEndpoints([]*endpoint.Endpoint{ep})
+	require.NoError(t, err)
+	require.Len(t, adjusted, 1)
+
+	val, ok := adjusted[0].GetProviderSpecificProperty(annotations.CloudflareTagsKey)
+	require.True(t, ok, "tags annotation should still be present after AdjustEndpoints")
+	// Tags should be sorted and whitespace-trimmed
+	assert.Equal(t, "alpha,beta,gamma", val)
+}
+
+func TestZoneServiceZoneIDByName(t *testing.T) {
+	// Build a minimal cloudflare API response page for /zones.
+	writeZonesPage := func(w http.ResponseWriter, zones []map[string]any) {
+		w.Header().Set("Content-Type", "application/json")
+		if err := json.NewEncoder(w).Encode(map[string]any{
+			"result": zones,
+			"result_info": map[string]any{
+				"count":       len(zones),
+				"total_count": len(zones),
+				"page":        1,
+				"per_page":    20,
+			},
+			"success":  true,
+			"errors":   []any{},
+			"messages": []any{},
+		}); err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+		}
+	}
+
+	t.Run("zone found returns its ID", func(t *testing.T) {
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+			writeZonesPage(w, []map[string]any{
+				{"id": "zone-abc", "name": "example.com", "plan": map[string]any{"is_subscribed": false}},
+			})
+		}))
+		defer ts.Close()
+
+		svc := &zoneService{service: cloudflare.NewClient(
+			option.WithBaseURL(ts.URL+"/"),
+			option.WithAPIToken("test-token"),
+			option.WithMaxRetries(0),
+		)}
+		id, err := svc.ZoneIDByName("example.com")
+		require.NoError(t, err)
+		assert.Equal(t, "zone-abc", id)
+	})
+
+	t.Run("zone not found returns descriptive error", func(t *testing.T) {
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+			writeZonesPage(w, []map[string]any{})
+		}))
+		defer ts.Close()
+
+		svc := &zoneService{service: cloudflare.NewClient(
+			option.WithBaseURL(ts.URL+"/"),
+			option.WithAPIToken("test-token"),
+			option.WithMaxRetries(0),
+		)}
+		id, err := svc.ZoneIDByName("missing.com")
+		assert.Empty(t, id)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "not found in CloudFlare account")
+	})
+
+	t.Run("server error causes wrapped iterator error", func(t *testing.T) {
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+			w.Header().Set("Content-Type", "application/json")
+			w.WriteHeader(http.StatusInternalServerError)
+			_ = json.NewEncoder(w).Encode(map[string]any{
+				"result":   nil,
+				"success":  false,
+				"errors":   []map[string]any{{"code": 500, "message": "internal server error"}},
+				"messages": []any{},
+			})
+		}))
+		defer ts.Close()
+
+		svc := &zoneService{service: cloudflare.NewClient(
+			option.WithBaseURL(ts.URL+"/"),
+			option.WithAPIToken("test-token"),
+			option.WithMaxRetries(0),
+		)}
+		id, err := svc.ZoneIDByName("any.com")
+		assert.Empty(t, id)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "failed to list zones from CloudFlare API")
+	})
 }