diff --git a/controller/controller.go b/controller/controller.go
index 41976c5f1..72058037c 100644
--- a/controller/controller.go
+++ b/controller/controller.go
@@ -223,6 +223,7 @@ func (c *Controller) RunOnce(ctx context.Context) error {
 		Desired:        endpoints,
 		DomainFilter:   endpoint.MatchAllDomainFilters{&c.DomainFilter, &registryFilter},
 		ManagedRecords: c.ManagedRecordTypes,
+		OwnerID:        c.Registry.OwnerID(),
 	}
 
 	plan = plan.Calculate()
diff --git a/endpoint/endpoint.go b/endpoint/endpoint.go
index 7df3f882f..d2106ff35 100644
--- a/endpoint/endpoint.go
+++ b/endpoint/endpoint.go
@@ -277,10 +277,31 @@ func (e *Endpoint) Key() EndpointKey {
 	}
 }
 
+// IsOwnedBy returns true if the endpoint owner label matches the given ownerID, false otherwise
+func (e *Endpoint) IsOwnedBy(ownerID string) bool {
+	endpointOwner, ok := e.Labels[OwnerLabelKey]
+	return ok && endpointOwner == ownerID
+}
+
 func (e *Endpoint) String() string {
 	return fmt.Sprintf("%s %d IN %s %s %s %s", e.DNSName, e.RecordTTL, e.RecordType, e.SetIdentifier, e.Targets, e.ProviderSpecific)
 }
 
+// Apply filter to slice of endpoints and return new filtered slice that includes
+// only endpoints that match.
+func FilterEndpointsByOwnerID(ownerID string, eps []*Endpoint) []*Endpoint {
+	filtered := []*Endpoint{}
+	for _, ep := range eps {
+		if endpointOwner, ok := ep.Labels[OwnerLabelKey]; !ok || endpointOwner != ownerID {
+			log.Debugf(`Skipping endpoint %v because owner id does not match, found: "%s", required: "%s"`, ep, endpointOwner, ownerID)
+		} else {
+			filtered = append(filtered, ep)
+		}
+	}
+
+	return filtered
+}
+
 // DNSEndpointSpec defines the desired state of DNSEndpoint
 type DNSEndpointSpec struct {
 	Endpoints []*Endpoint `json:"endpoints,omitempty"`
diff --git a/endpoint/endpoint_test.go b/endpoint/endpoint_test.go
index 81e7f4c33..47c23e2f4 100644
--- a/endpoint/endpoint_test.go
+++ b/endpoint/endpoint_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package endpoint
 
 import (
+	"reflect"
 	"testing"
 )
 
@@ -115,3 +116,103 @@ func TestIsLess(t *testing.T) {
 		}
 	}
 }
+
+func TestFilterEndpointsByOwnerID(t *testing.T) {
+	foo1 := &Endpoint{
+		DNSName:    "foo.com",
+		RecordType: RecordTypeA,
+		Labels: Labels{
+			OwnerLabelKey: "foo",
+		},
+	}
+	foo2 := &Endpoint{
+		DNSName:    "foo.com",
+		RecordType: RecordTypeCNAME,
+		Labels: Labels{
+			OwnerLabelKey: "foo",
+		},
+	}
+	bar := &Endpoint{
+		DNSName:    "foo.com",
+		RecordType: RecordTypeA,
+		Labels: Labels{
+			OwnerLabelKey: "bar",
+		},
+	}
+	type args struct {
+		ownerID string
+		eps     []*Endpoint
+	}
+	tests := []struct {
+		name string
+		args args
+		want []*Endpoint
+	}{
+		{
+			name: "filter values",
+			args: args{
+				ownerID: "foo",
+				eps: []*Endpoint{
+					foo1,
+					foo2,
+					bar,
+				},
+			},
+			want: []*Endpoint{
+				foo1,
+				foo2,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := FilterEndpointsByOwnerID(tt.args.ownerID, tt.args.eps); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("ApplyEndpointFilter() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestIsOwnedBy(t *testing.T) {
+	type fields struct {
+		Labels Labels
+	}
+	type args struct {
+		ownerID string
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   bool
+	}{
+		{
+			name:   "empty labels",
+			fields: fields{Labels: Labels{}},
+			args:   args{ownerID: "foo"},
+			want:   false,
+		},
+		{
+			name:   "owner label not match",
+			fields: fields{Labels: Labels{OwnerLabelKey: "bar"}},
+			args:   args{ownerID: "foo"},
+			want:   false,
+		},
+		{
+			name:   "owner label match",
+			fields: fields{Labels: Labels{OwnerLabelKey: "foo"}},
+			args:   args{ownerID: "foo"},
+			want:   true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			e := &Endpoint{
+				Labels: tt.fields.Labels,
+			}
+			if got := e.IsOwnedBy(tt.args.ownerID); got != tt.want {
+				t.Errorf("Endpoint.IsOwnedBy() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
diff --git a/plan/conflict.go b/plan/conflict.go
index 9b1f29cfe..3044059b7 100644
--- a/plan/conflict.go
+++ b/plan/conflict.go
@@ -19,6 +19,8 @@ package plan
 import (
 	"sort"
 
+	log "github.com/sirupsen/logrus"
+
 	"sigs.k8s.io/external-dns/endpoint"
 )
 
@@ -27,6 +29,7 @@ import (
 type ConflictResolver interface {
 	ResolveCreate(candidates []*endpoint.Endpoint) *endpoint.Endpoint
 	ResolveUpdate(current *endpoint.Endpoint, candidates []*endpoint.Endpoint) *endpoint.Endpoint
+	ResolveRecordTypes(key planKey, row *planTableRow) map[string]*domainEndpoints
 }
 
 // PerResource allows only one resource to own a given dns name
@@ -62,6 +65,58 @@ func (s PerResource) ResolveUpdate(current *endpoint.Endpoint, candidates []*end
 	return s.ResolveCreate(candidates)
 }
 
+// ResolveRecordTypes attempts to detect and resolve record type conflicts in desired
+// endpoints for a domain. For eample if the there is more than 1 candidate and at lease one
+// of them is a CNAME. Per [RFC 1034 3.6.2] domains that contain a CNAME can not contain any
+// other record types. The default policy will prefer A and AAAA record types when a conflict is
+// detected (consistent with [endpoint.Targets.Less]).
+//
+// [RFC 1034 3.6.2]: https://datatracker.ietf.org/doc/html/rfc1034#autoid-15
+func (s PerResource) ResolveRecordTypes(key planKey, row *planTableRow) map[string]*domainEndpoints {
+	// no conflicts if only a single desired record type for the domain
+	if len(row.candidates) <= 1 {
+		return row.records
+	}
+
+	cname := false
+	other := false
+	for _, c := range row.candidates {
+		if c.RecordType == endpoint.RecordTypeCNAME {
+			cname = true
+		} else {
+			other = true
+		}
+
+		if cname && other {
+			break
+		}
+	}
+
+	// conflict was found, remove candiates of non-preferred record types
+	if cname && other {
+		log.Infof("Domain %s contains conflicting record type candidates; discarding CNAME record", key.dnsName)
+		records := map[string]*domainEndpoints{}
+		for recordType, recs := range row.records {
+			// policy is to prefer the non-CNAME record types when a conflict is found
+			if recordType == endpoint.RecordTypeCNAME {
+				// discard candidates of conflicting records
+				// keep currect so they can be deleted
+				records[recordType] = &domainEndpoints{
+					current:    recs.current,
+					candidates: []*endpoint.Endpoint{},
+				}
+			} else {
+				records[recordType] = recs
+			}
+		}
+
+		return records
+	}
+
+	// no conflict, return all records types
+	return row.records
+}
+
 // less returns true if endpoint x is less than y
 func (s PerResource) less(x, y *endpoint.Endpoint) bool {
 	return x.Targets.IsLess(y.Targets)
diff --git a/plan/conflict_test.go b/plan/conflict_test.go
index 491b42bff..d09d02837 100644
--- a/plan/conflict_test.go
+++ b/plan/conflict_test.go
@@ -17,10 +17,10 @@ limitations under the License.
 package plan
 
 import (
+	"reflect"
 	"testing"
 
 	"github.com/stretchr/testify/suite"
-
 	"sigs.k8s.io/external-dns/endpoint"
 )
 
@@ -34,6 +34,7 @@ type ResolverSuite struct {
 	fooV2Cname          *endpoint.Endpoint
 	fooV2CnameDuplicate *endpoint.Endpoint
 	fooA5               *endpoint.Endpoint
+	fooAAAA5            *endpoint.Endpoint
 	bar127A             *endpoint.Endpoint
 	bar192A             *endpoint.Endpoint
 	bar127AAnother      *endpoint.Endpoint
@@ -76,6 +77,14 @@ func (suite *ResolverSuite) SetupTest() {
 			endpoint.ResourceLabelKey: "ingress/default/foo-5",
 		},
 	}
+	suite.fooAAAA5 = &endpoint.Endpoint{
+		DNSName:    "foo",
+		Targets:    endpoint.Targets{"2001:DB8::1"},
+		RecordType: "AAAA",
+		Labels: map[string]string{
+			endpoint.ResourceLabelKey: "ingress/default/foo-5",
+		},
+	}
 	suite.bar127A = &endpoint.Endpoint{
 		DNSName:    "bar",
 		Targets:    endpoint.Targets{"127.0.0.1"},
@@ -133,6 +142,156 @@ func (suite *ResolverSuite) TestStrictResolver() {
 	suite.Equal(suite.bar127A, suite.perResource.ResolveUpdate(suite.legacyBar192A, []*endpoint.Endpoint{suite.bar127A, suite.bar192A}), " legacy record's resource value will not match, should pick minimum")
 }
 
+func (suite *ResolverSuite) TestPerResource_ResolveRecordTypes() {
+	type args struct {
+		key planKey
+		row *planTableRow
+	}
+	tests := []struct {
+		name string
+		args args
+		want map[string]*domainEndpoints
+	}{
+		{
+			name: "no conflict: cname record",
+			args: args{
+				key: planKey{dnsName: "foo"},
+				row: &planTableRow{
+					candidates: []*endpoint.Endpoint{suite.fooV1Cname},
+					records: map[string]*domainEndpoints{
+						endpoint.RecordTypeCNAME: {
+							candidates: []*endpoint.Endpoint{suite.fooV1Cname},
+						},
+					},
+				},
+			},
+			want: map[string]*domainEndpoints{
+				endpoint.RecordTypeCNAME: {
+					candidates: []*endpoint.Endpoint{suite.fooV1Cname},
+				},
+			},
+		},
+		{
+			name: "no conflict: a record",
+			args: args{
+				key: planKey{dnsName: "foo"},
+				row: &planTableRow{
+					current:    []*endpoint.Endpoint{suite.fooA5},
+					candidates: []*endpoint.Endpoint{suite.fooA5},
+					records: map[string]*domainEndpoints{
+						endpoint.RecordTypeA: {
+							current:    suite.fooA5,
+							candidates: []*endpoint.Endpoint{suite.fooA5},
+						},
+					},
+				},
+			},
+			want: map[string]*domainEndpoints{
+				endpoint.RecordTypeA: {
+					current:    suite.fooA5,
+					candidates: []*endpoint.Endpoint{suite.fooA5},
+				},
+			},
+		},
+		{
+			name: "no conflict: a and aaaa records",
+			args: args{
+				key: planKey{dnsName: "foo"},
+				row: &planTableRow{
+					candidates: []*endpoint.Endpoint{suite.fooA5, suite.fooAAAA5},
+					records: map[string]*domainEndpoints{
+						endpoint.RecordTypeA: {
+							candidates: []*endpoint.Endpoint{suite.fooA5},
+						},
+						endpoint.RecordTypeAAAA: {
+							candidates: []*endpoint.Endpoint{suite.fooAAAA5},
+						},
+					},
+				},
+			},
+			want: map[string]*domainEndpoints{
+				endpoint.RecordTypeA: {
+					candidates: []*endpoint.Endpoint{suite.fooA5},
+				},
+				endpoint.RecordTypeAAAA: {
+					candidates: []*endpoint.Endpoint{suite.fooAAAA5},
+				},
+			},
+		},
+		{
+			name: "conflict: cname and a records",
+			args: args{
+				key: planKey{dnsName: "foo"},
+				row: &planTableRow{
+					current:    []*endpoint.Endpoint{suite.fooV1Cname},
+					candidates: []*endpoint.Endpoint{suite.fooV1Cname, suite.fooA5},
+					records: map[string]*domainEndpoints{
+						endpoint.RecordTypeCNAME: {
+							current:    suite.fooV1Cname,
+							candidates: []*endpoint.Endpoint{suite.fooV1Cname},
+						},
+						endpoint.RecordTypeA: {
+							candidates: []*endpoint.Endpoint{suite.fooA5},
+						},
+					},
+				},
+			},
+			want: map[string]*domainEndpoints{
+				endpoint.RecordTypeCNAME: {
+					current:    suite.fooV1Cname,
+					candidates: []*endpoint.Endpoint{},
+				},
+				endpoint.RecordTypeA: {
+					candidates: []*endpoint.Endpoint{suite.fooA5},
+				},
+			},
+		},
+		{
+			name: "conflict: cname, a, and aaaa records",
+			args: args{
+				key: planKey{dnsName: "foo"},
+				row: &planTableRow{
+					current:    []*endpoint.Endpoint{suite.fooA5, suite.fooAAAA5},
+					candidates: []*endpoint.Endpoint{suite.fooV1Cname, suite.fooA5, suite.fooAAAA5},
+					records: map[string]*domainEndpoints{
+						endpoint.RecordTypeCNAME: {
+							candidates: []*endpoint.Endpoint{suite.fooV1Cname},
+						},
+						endpoint.RecordTypeA: {
+							current:    suite.fooA5,
+							candidates: []*endpoint.Endpoint{suite.fooA5},
+						},
+						endpoint.RecordTypeAAAA: {
+							current:    suite.fooAAAA5,
+							candidates: []*endpoint.Endpoint{suite.fooAAAA5},
+						},
+					},
+				},
+			},
+			want: map[string]*domainEndpoints{
+				endpoint.RecordTypeCNAME: {
+					candidates: []*endpoint.Endpoint{},
+				},
+				endpoint.RecordTypeA: {
+					current:    suite.fooA5,
+					candidates: []*endpoint.Endpoint{suite.fooA5},
+				},
+				endpoint.RecordTypeAAAA: {
+					current:    suite.fooAAAA5,
+					candidates: []*endpoint.Endpoint{suite.fooAAAA5},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		suite.Run(tt.name, func() {
+			if got := suite.perResource.ResolveRecordTypes(tt.args.key, tt.args.row); !reflect.DeepEqual(got, tt.want) {
+				suite.T().Errorf("PerResource.ResolveRecordTypes() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
 func TestConflictResolver(t *testing.T) {
 	suite.Run(t, new(ResolverSuite))
 }
diff --git a/plan/plan.go b/plan/plan.go
index d6624cced..a9299cf9c 100644
--- a/plan/plan.go
+++ b/plan/plan.go
@@ -46,6 +46,8 @@ type Plan struct {
 	DomainFilter endpoint.MatchAllDomainFilters
 	// DNS record types that will be considered for management
 	ManagedRecords []string
+	// OwnerID of records to manage
+	OwnerID string
 }
 
 // Changes holds lists of actions to be executed by dns providers
@@ -64,22 +66,26 @@ type Changes struct {
 type planKey struct {
 	dnsName       string
 	setIdentifier string
-	recordType    string
 }
 
 // planTable is a supplementary struct for Plan
-// each row correspond to a planKey -> (current record + all desired records)
-/*
-planTable: (-> = target)
---------------------------------------------------------
-DNSName | Current record | Desired Records             |
---------------------------------------------------------
-foo.com | -> 1.1.1.1     | [->1.1.1.1, ->elb.com]      |  = no action
---------------------------------------------------------
-bar.com |                | [->191.1.1.1, ->190.1.1.1]  |  = create (bar.com -> 190.1.1.1)
---------------------------------------------------------
-"=", i.e. result of calculation relies on supplied ConflictResolver
-*/
+// each row correspond to a planKey -> (current records + all desired records)
+//
+//	planTable (-> = target)
+//	--------------------------------------------------------------
+//	DNSName | Current record       | Desired Records             |
+//	--------------------------------------------------------------
+//	foo.com | [->1.1.1.1 ]         | [->1.1.1.1]                 |  = no action
+//	--------------------------------------------------------------
+//	bar.com |                      | [->191.1.1.1, ->190.1.1.1]  |  = create (bar.com [-> 190.1.1.1])
+//	--------------------------------------------------------------
+//	dog.com | [->1.1.1.2]          |                             |  = delete (dog.com [-> 1.1.1.2])
+//	--------------------------------------------------------------
+//	cat.com | [->::1, ->1.1.1.3]   | [->1.1.1.3]                 |  = update old (cat.com [-> ::1, -> 1.1.1.3]) new (cat.com [-> 1.1.1.3])
+//	--------------------------------------------------------------
+//	big.com | [->1.1.1.4]          | [->ing.elb.com]             |  = update old (big.com [-> 1.1.1.4]) new (big.com [-> ing.elb.com])
+//	--------------------------------------------------------------
+//	"=", i.e. result of calculation relies on supplied ConflictResolver
 type planTable struct {
 	rows     map[planKey]*planTableRow
 	resolver ConflictResolver
@@ -89,11 +95,26 @@ func newPlanTable() planTable { // TODO: make resolver configurable
 	return planTable{map[planKey]*planTableRow{}, PerResource{}}
 }
 
-// planTableRow
-// current corresponds to the record currently occupying dns name on the dns provider
-// candidates corresponds to the list of records which would like to have this dnsName
+// planTableRow represents a set of current and desired domain resource records.
 type planTableRow struct {
-	current    *endpoint.Endpoint
+	// current corresponds to the records currently occupying dns name on the dns provider. More than one record may
+	// be represented here: for example A and AAAA. If the current domain record is a CNAME, no other record types
+	// are allowed per [RFC 1034 3.6.2]
+	//
+	// [RFC 1034 3.6.2]: https://datatracker.ietf.org/doc/html/rfc1034#autoid-15
+	current []*endpoint.Endpoint
+	// candidates corresponds to the list of records which would like to have this dnsName.
+	candidates []*endpoint.Endpoint
+	// records is a grouping of current and candidates by record type, for example A, AAAA, CNAME.
+	records map[string]*domainEndpoints
+}
+
+// domainEndpoints is a grouping of current, which are existing records from the registry, and candidates,
+// which are desired records from the source. All records in this grouping have the same record type.
+type domainEndpoints struct {
+	// current corresponds to existing record from the registry. Maybe nil if no current record of the type exists.
+	current *endpoint.Endpoint
+	// candidates corresponds to the list of records which would like to have this dnsName.
 	candidates []*endpoint.Endpoint
 }
 
@@ -103,23 +124,32 @@ func (t planTableRow) String() string {
 
 func (t planTable) addCurrent(e *endpoint.Endpoint) {
 	key := t.newPlanKey(e)
-	t.rows[key].current = e
+	t.rows[key].current = append(t.rows[key].current, e)
+	t.rows[key].records[e.RecordType].current = e
 }
 
 func (t planTable) addCandidate(e *endpoint.Endpoint) {
 	key := t.newPlanKey(e)
 	t.rows[key].candidates = append(t.rows[key].candidates, e)
+	t.rows[key].records[e.RecordType].candidates = append(t.rows[key].records[e.RecordType].candidates, e)
 }
 
 func (t *planTable) newPlanKey(e *endpoint.Endpoint) planKey {
 	key := planKey{
 		dnsName:       normalizeDNSName(e.DNSName),
 		setIdentifier: e.SetIdentifier,
-		recordType:    e.RecordType,
 	}
+
 	if _, ok := t.rows[key]; !ok {
-		t.rows[key] = &planTableRow{}
+		t.rows[key] = &planTableRow{
+			records: make(map[string]*domainEndpoints),
+		}
 	}
+
+	if _, ok := t.rows[key].records[e.RecordType]; !ok {
+		t.rows[key].records[e.RecordType] = &domainEndpoints{}
+	}
+
 	return key
 }
 
@@ -149,30 +179,82 @@ func (p *Plan) Calculate() *Plan {
 
 	changes := &Changes{}
 
-	for _, row := range t.rows {
-		if row.current == nil { // dns name not taken
-			changes.Create = append(changes.Create, t.resolver.ResolveCreate(row.candidates))
-		}
-		if row.current != nil && len(row.candidates) == 0 {
-			changes.Delete = append(changes.Delete, row.current)
+	for key, row := range t.rows {
+		// dns name not taken
+		if len(row.current) == 0 {
+			recordsByType := t.resolver.ResolveRecordTypes(key, row)
+			for _, records := range recordsByType {
+				if len(records.candidates) > 0 {
+					changes.Create = append(changes.Create, t.resolver.ResolveCreate(records.candidates))
+				}
+			}
 		}
 
-		// TODO: allows record type change, which might not be supported by all dns providers
-		if row.current != nil && len(row.candidates) > 0 { // dns name is taken
-			update := t.resolver.ResolveUpdate(row.current, row.candidates)
-			// compare "update" to "current" to figure out if actual update is required
-			if shouldUpdateTTL(update, row.current) || targetChanged(update, row.current) || p.shouldUpdateProviderSpecific(update, row.current) {
-				inheritOwner(row.current, update)
-				changes.UpdateNew = append(changes.UpdateNew, update)
-				changes.UpdateOld = append(changes.UpdateOld, row.current)
+		// dns name released or possibly owned by a different external dns
+		if len(row.current) > 0 && len(row.candidates) == 0 {
+			changes.Delete = append(changes.Delete, row.current...)
+		}
+
+		// dns name is taken
+		if len(row.current) > 0 && len(row.candidates) > 0 {
+			creates := []*endpoint.Endpoint{}
+
+			// apply changes for each record type
+			recordsByType := t.resolver.ResolveRecordTypes(key, row)
+			for _, records := range recordsByType {
+				// record type not desired
+				if records.current != nil && len(records.candidates) == 0 {
+					changes.Delete = append(changes.Delete, records.current)
+				}
+
+				// new record type desired
+				if records.current == nil && len(records.candidates) > 0 {
+					update := t.resolver.ResolveCreate(records.candidates)
+					// creates are evaluated after all domain records have been processed to
+					// validate that this external dns has ownership claim on the domain before
+					// adding the records to planned changes.
+					creates = append(creates, update)
+				}
+
+				// update existing record
+				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) {
+						inheritOwner(records.current, update)
+						changes.UpdateNew = append(changes.UpdateNew, update)
+						changes.UpdateOld = append(changes.UpdateOld, records.current)
+					}
+				}
+			}
+
+			if len(creates) > 0 {
+				// only add creates if the external dns has ownership claim on the domain
+				ownersMatch := true
+				for _, current := range row.current {
+					if p.OwnerID != "" && !current.IsOwnedBy(p.OwnerID) {
+						ownersMatch = false
+					}
+				}
+
+				if ownersMatch {
+					changes.Create = append(changes.Create, creates...)
+				}
 			}
-			continue
 		}
 	}
+
 	for _, pol := range p.Policies {
 		changes = pol.Apply(changes)
 	}
 
+	// filter out updates this external dns does not have ownership claim over
+	if p.OwnerID != "" {
+		changes.Delete = endpoint.FilterEndpointsByOwnerID(p.OwnerID, changes.Delete)
+		changes.UpdateOld = endpoint.FilterEndpointsByOwnerID(p.OwnerID, changes.UpdateOld)
+		changes.UpdateNew = endpoint.FilterEndpointsByOwnerID(p.OwnerID, changes.UpdateNew)
+	}
+
 	plan := &Plan{
 		Current:        p.Current,
 		Desired:        p.Desired,
diff --git a/plan/plan_test.go b/plan/plan_test.go
index d5f583f5f..a98d115a6 100644
--- a/plan/plan_test.go
+++ b/plan/plan_test.go
@@ -124,7 +124,7 @@ func (suite *PlanTestSuite) SetupTest() {
 	}
 	suite.dsAAAA = &endpoint.Endpoint{
 		DNSName:    "ds",
-		Targets:    endpoint.Targets{"1.1.1.1"},
+		Targets:    endpoint.Targets{"2001:DB8::1"},
 		RecordType: "AAAA",
 		Labels: map[string]string{
 			endpoint.ResourceLabelKey: "ingress/default/ds-AAAAA",
@@ -452,19 +452,204 @@ func (suite *PlanTestSuite) TestIdempotency() {
 	validateEntries(suite.T(), changes.Delete, expectedDelete)
 }
 
-func (suite *PlanTestSuite) TestDifferentTypes() {
+func (suite *PlanTestSuite) TestRecordTypeChange() {
 	current := []*endpoint.Endpoint{suite.fooV1Cname}
-	desired := []*endpoint.Endpoint{suite.fooV2Cname, suite.fooA5}
+	desired := []*endpoint.Endpoint{suite.fooA5}
 	expectedCreate := []*endpoint.Endpoint{suite.fooA5}
-	expectedUpdateOld := []*endpoint.Endpoint{suite.fooV1Cname}
-	expectedUpdateNew := []*endpoint.Endpoint{suite.fooV2Cname}
+	expectedUpdateOld := []*endpoint.Endpoint{}
+	expectedUpdateNew := []*endpoint.Endpoint{}
+	expectedDelete := []*endpoint.Endpoint{suite.fooV1Cname}
+
+	p := &Plan{
+		Policies:       []Policy{&SyncPolicy{}},
+		Current:        current,
+		Desired:        desired,
+		ManagedRecords: []string{endpoint.RecordTypeA, endpoint.RecordTypeAAAA, endpoint.RecordTypeCNAME},
+		OwnerID:        suite.fooV1Cname.Labels[endpoint.OwnerLabelKey],
+	}
+
+	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 (suite *PlanTestSuite) TestExistingCNameWithDualStackDesired() {
+	current := []*endpoint.Endpoint{suite.fooV1Cname}
+	desired := []*endpoint.Endpoint{suite.fooA5, suite.fooAAAA}
+	expectedCreate := []*endpoint.Endpoint{suite.fooA5, suite.fooAAAA}
+	expectedUpdateOld := []*endpoint.Endpoint{}
+	expectedUpdateNew := []*endpoint.Endpoint{}
+	expectedDelete := []*endpoint.Endpoint{suite.fooV1Cname}
+
+	p := &Plan{
+		Policies:       []Policy{&SyncPolicy{}},
+		Current:        current,
+		Desired:        desired,
+		ManagedRecords: []string{endpoint.RecordTypeA, endpoint.RecordTypeAAAA, endpoint.RecordTypeCNAME},
+		OwnerID:        suite.fooV1Cname.Labels[endpoint.OwnerLabelKey],
+	}
+
+	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 (suite *PlanTestSuite) TestExistingDualStackWithCNameDesired() {
+	suite.fooA5.Labels[endpoint.OwnerLabelKey] = "nerf"
+	suite.fooAAAA.Labels[endpoint.OwnerLabelKey] = "nerf"
+	current := []*endpoint.Endpoint{suite.fooA5, suite.fooAAAA}
+	desired := []*endpoint.Endpoint{suite.fooV2Cname}
+	expectedCreate := []*endpoint.Endpoint{suite.fooV2Cname}
+	expectedUpdateOld := []*endpoint.Endpoint{}
+	expectedUpdateNew := []*endpoint.Endpoint{}
+	expectedDelete := []*endpoint.Endpoint{suite.fooA5, suite.fooAAAA}
+
+	p := &Plan{
+		Policies:       []Policy{&SyncPolicy{}},
+		Current:        current,
+		Desired:        desired,
+		ManagedRecords: []string{endpoint.RecordTypeA, endpoint.RecordTypeAAAA, endpoint.RecordTypeCNAME},
+		OwnerID:        suite.fooA5.Labels[endpoint.OwnerLabelKey],
+	}
+
+	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)
+}
+
+// TestExistingOwnerNotMatchingDualStackDesired validates that if there is an existing
+// record for a domain but there is no ownership claim over it and there are desired
+// records no changes are planed. Only domains that have explicit ownership claims should
+// be updated.
+func (suite *PlanTestSuite) TestExistingOwnerNotMatchingDualStackDesired() {
+	suite.fooA5.Labels = nil
+	current := []*endpoint.Endpoint{suite.fooA5}
+	desired := []*endpoint.Endpoint{suite.fooV2Cname}
+	expectedCreate := []*endpoint.Endpoint{}
+	expectedUpdateOld := []*endpoint.Endpoint{}
+	expectedUpdateNew := []*endpoint.Endpoint{}
 	expectedDelete := []*endpoint.Endpoint{}
 
 	p := &Plan{
 		Policies:       []Policy{&SyncPolicy{}},
 		Current:        current,
 		Desired:        desired,
-		ManagedRecords: []string{endpoint.RecordTypeA, endpoint.RecordTypeCNAME},
+		ManagedRecords: []string{endpoint.RecordTypeA, endpoint.RecordTypeAAAA, endpoint.RecordTypeCNAME},
+		OwnerID:        "pwner",
+	}
+
+	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)
+}
+
+// TestConflictingCurrentNonConflictingDesired is a bit of a corner case as it would indicate
+// that the provider is not following valid DNS rules or there may be some
+// caching issues. In this case since the desired records are not conflicting
+// the updates will end up with the conflict resolved.
+func (suite *PlanTestSuite) TestConflictingCurrentNonConflictingDesired() {
+	suite.fooA5.Labels[endpoint.OwnerLabelKey] = suite.fooV1Cname.Labels[endpoint.OwnerLabelKey]
+	current := []*endpoint.Endpoint{suite.fooV1Cname, suite.fooA5}
+	desired := []*endpoint.Endpoint{suite.fooA5}
+	expectedCreate := []*endpoint.Endpoint{}
+	expectedUpdateOld := []*endpoint.Endpoint{}
+	expectedUpdateNew := []*endpoint.Endpoint{}
+	expectedDelete := []*endpoint.Endpoint{suite.fooV1Cname}
+
+	p := &Plan{
+		Policies:       []Policy{&SyncPolicy{}},
+		Current:        current,
+		Desired:        desired,
+		ManagedRecords: []string{endpoint.RecordTypeA, endpoint.RecordTypeAAAA, endpoint.RecordTypeCNAME},
+		OwnerID:        suite.fooV1Cname.Labels[endpoint.OwnerLabelKey],
+	}
+
+	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)
+}
+
+// TestConflictingCurrentNoDesired is a bit of a corner case as it would indicate
+// that the provider is not following valid DNS rules or there may be some
+// caching issues. In this case there are no desired enpoint candidates so plan
+// on deleting the records.
+func (suite *PlanTestSuite) TestConflictingCurrentNoDesired() {
+	suite.fooA5.Labels[endpoint.OwnerLabelKey] = suite.fooV1Cname.Labels[endpoint.OwnerLabelKey]
+	current := []*endpoint.Endpoint{suite.fooV1Cname, suite.fooA5}
+	desired := []*endpoint.Endpoint{}
+	expectedCreate := []*endpoint.Endpoint{}
+	expectedUpdateOld := []*endpoint.Endpoint{}
+	expectedUpdateNew := []*endpoint.Endpoint{}
+	expectedDelete := []*endpoint.Endpoint{suite.fooV1Cname, suite.fooA5}
+
+	p := &Plan{
+		Policies:       []Policy{&SyncPolicy{}},
+		Current:        current,
+		Desired:        desired,
+		ManagedRecords: []string{endpoint.RecordTypeA, endpoint.RecordTypeAAAA, endpoint.RecordTypeCNAME},
+		OwnerID:        suite.fooV1Cname.Labels[endpoint.OwnerLabelKey],
+	}
+
+	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)
+}
+
+// TestCurrentWithConflictingDesired simulates where the desired records result in conflicting records types.
+// This could be the result of multiple sources generating conflicting records types. In this case the conflict
+// resolver should prefer the A and AAAA record candidate and delete the other records.
+func (suite *PlanTestSuite) TestCurrentWithConflictingDesired() {
+	suite.fooV1Cname.Labels[endpoint.OwnerLabelKey] = "nerf"
+	current := []*endpoint.Endpoint{suite.fooV1Cname}
+	desired := []*endpoint.Endpoint{suite.fooV1Cname, suite.fooA5, suite.fooAAAA}
+	expectedCreate := []*endpoint.Endpoint{suite.fooA5, suite.fooAAAA}
+	expectedUpdateOld := []*endpoint.Endpoint{}
+	expectedUpdateNew := []*endpoint.Endpoint{}
+	expectedDelete := []*endpoint.Endpoint{suite.fooV1Cname}
+
+	p := &Plan{
+		Policies:       []Policy{&SyncPolicy{}},
+		Current:        current,
+		Desired:        desired,
+		ManagedRecords: []string{endpoint.RecordTypeA, endpoint.RecordTypeAAAA, endpoint.RecordTypeCNAME},
+		OwnerID:        suite.fooV1Cname.Labels[endpoint.OwnerLabelKey],
+	}
+
+	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)
+}
+
+// TestNoCurrentWithConflictingDesired simulates where the desired records result in conflicting records types.
+// This could be the result of multiple sources generating conflicting records types. In this case there the
+// conflict resolver should prefer the A and AAAA record and drop the other candidate record types.
+func (suite *PlanTestSuite) TestNoCurrentWithConflictingDesired() {
+	current := []*endpoint.Endpoint{}
+	desired := []*endpoint.Endpoint{suite.fooV1Cname, suite.fooA5, suite.fooAAAA}
+	expectedCreate := []*endpoint.Endpoint{suite.fooA5, suite.fooAAAA}
+	expectedUpdateOld := []*endpoint.Endpoint{}
+	expectedUpdateNew := []*endpoint.Endpoint{}
+	expectedDelete := []*endpoint.Endpoint{}
+
+	p := &Plan{
+		Policies:       []Policy{&SyncPolicy{}},
+		Current:        current,
+		Desired:        desired,
+		ManagedRecords: []string{endpoint.RecordTypeA, endpoint.RecordTypeAAAA, endpoint.RecordTypeCNAME},
 	}
 
 	changes := p.Calculate().Changes
@@ -654,10 +839,10 @@ func (suite *PlanTestSuite) TestDomainFiltersUpdate() {
 }
 
 func (suite *PlanTestSuite) TestAAAARecords() {
-
 	current := []*endpoint.Endpoint{}
 	desired := []*endpoint.Endpoint{suite.fooAAAA}
 	expectedCreate := []*endpoint.Endpoint{suite.fooAAAA}
+	expectNoChanges := []*endpoint.Endpoint{}
 
 	p := &Plan{
 		Policies:       []Policy{&SyncPolicy{}},
@@ -668,12 +853,16 @@ func (suite *PlanTestSuite) TestAAAARecords() {
 
 	changes := p.Calculate().Changes
 	validateEntries(suite.T(), changes.Create, expectedCreate)
+	validateEntries(suite.T(), changes.Delete, expectNoChanges)
+	validateEntries(suite.T(), changes.UpdateOld, expectNoChanges)
+	validateEntries(suite.T(), changes.UpdateNew, expectNoChanges)
 }
 
 func (suite *PlanTestSuite) TestDualStackRecords() {
 	current := []*endpoint.Endpoint{}
 	desired := []*endpoint.Endpoint{suite.dsA, suite.dsAAAA}
 	expectedCreate := []*endpoint.Endpoint{suite.dsA, suite.dsAAAA}
+	expectNoChanges := []*endpoint.Endpoint{}
 
 	p := &Plan{
 		Policies:       []Policy{&SyncPolicy{}},
@@ -684,6 +873,49 @@ func (suite *PlanTestSuite) TestDualStackRecords() {
 
 	changes := p.Calculate().Changes
 	validateEntries(suite.T(), changes.Create, expectedCreate)
+	validateEntries(suite.T(), changes.Delete, expectNoChanges)
+	validateEntries(suite.T(), changes.UpdateOld, expectNoChanges)
+	validateEntries(suite.T(), changes.UpdateNew, expectNoChanges)
+}
+
+func (suite *PlanTestSuite) TestDualStackRecordsDelete() {
+	current := []*endpoint.Endpoint{suite.dsA, suite.dsAAAA}
+	desired := []*endpoint.Endpoint{}
+	expectedDelete := []*endpoint.Endpoint{suite.dsA, suite.dsAAAA}
+	expectNoChanges := []*endpoint.Endpoint{}
+
+	p := &Plan{
+		Policies:       []Policy{&SyncPolicy{}},
+		Current:        current,
+		Desired:        desired,
+		ManagedRecords: []string{endpoint.RecordTypeA, endpoint.RecordTypeAAAA, endpoint.RecordTypeCNAME},
+	}
+
+	changes := p.Calculate().Changes
+	validateEntries(suite.T(), changes.Delete, expectedDelete)
+	validateEntries(suite.T(), changes.Create, expectNoChanges)
+	validateEntries(suite.T(), changes.UpdateOld, expectNoChanges)
+	validateEntries(suite.T(), changes.UpdateNew, expectNoChanges)
+}
+
+func (suite *PlanTestSuite) TestDualStackToSingleStack() {
+	current := []*endpoint.Endpoint{suite.dsA, suite.dsAAAA}
+	desired := []*endpoint.Endpoint{suite.dsA}
+	expectedDelete := []*endpoint.Endpoint{suite.dsAAAA}
+	expectNoChanges := []*endpoint.Endpoint{}
+
+	p := &Plan{
+		Policies:       []Policy{&SyncPolicy{}},
+		Current:        current,
+		Desired:        desired,
+		ManagedRecords: []string{endpoint.RecordTypeA, endpoint.RecordTypeAAAA, endpoint.RecordTypeCNAME},
+	}
+
+	changes := p.Calculate().Changes
+	validateEntries(suite.T(), changes.Delete, expectedDelete)
+	validateEntries(suite.T(), changes.Create, expectNoChanges)
+	validateEntries(suite.T(), changes.UpdateOld, expectNoChanges)
+	validateEntries(suite.T(), changes.UpdateNew, expectNoChanges)
 }
 
 func TestPlan(t *testing.T) {
diff --git a/registry/aws_sd_registry.go b/registry/aws_sd_registry.go
index 1918f6dbc..bfb82b75a 100644
--- a/registry/aws_sd_registry.go
+++ b/registry/aws_sd_registry.go
@@ -46,6 +46,10 @@ func (sdr *AWSSDRegistry) GetDomainFilter() endpoint.DomainFilter {
 	return sdr.provider.GetDomainFilter()
 }
 
+func (im *AWSSDRegistry) OwnerID() string {
+	return im.ownerID
+}
+
 // Records calls AWS SD API and expects AWS SD provider to provider Owner/Resource information as a serialized
 // value in the AWSSDDescriptionLabel value in the Labels map
 func (sdr *AWSSDRegistry) Records(ctx context.Context) ([]*endpoint.Endpoint, error) {
@@ -72,9 +76,9 @@ func (sdr *AWSSDRegistry) Records(ctx context.Context) ([]*endpoint.Endpoint, er
 func (sdr *AWSSDRegistry) ApplyChanges(ctx context.Context, changes *plan.Changes) error {
 	filteredChanges := &plan.Changes{
 		Create:    changes.Create,
-		UpdateNew: filterOwnedRecords(sdr.ownerID, changes.UpdateNew),
-		UpdateOld: filterOwnedRecords(sdr.ownerID, changes.UpdateOld),
-		Delete:    filterOwnedRecords(sdr.ownerID, changes.Delete),
+		UpdateNew: endpoint.FilterEndpointsByOwnerID(sdr.ownerID, changes.UpdateNew),
+		UpdateOld: endpoint.FilterEndpointsByOwnerID(sdr.ownerID, changes.UpdateOld),
+		Delete:    endpoint.FilterEndpointsByOwnerID(sdr.ownerID, changes.Delete),
 	}
 
 	sdr.updateLabels(filteredChanges.Create)
diff --git a/registry/dynamodb.go b/registry/dynamodb.go
index 7ccb5f4f1..056587e10 100644
--- a/registry/dynamodb.go
+++ b/registry/dynamodb.go
@@ -104,6 +104,10 @@ func (im *DynamoDBRegistry) GetDomainFilter() endpoint.DomainFilter {
 	return im.provider.GetDomainFilter()
 }
 
+func (im *DynamoDBRegistry) OwnerID() string {
+	return im.ownerID
+}
+
 // Records returns the current records from the registry.
 func (im *DynamoDBRegistry) Records(ctx context.Context) ([]*endpoint.Endpoint, error) {
 	// If we have the zones cached AND we have refreshed the cache since the
@@ -211,9 +215,9 @@ func (im *DynamoDBRegistry) Records(ctx context.Context) ([]*endpoint.Endpoint,
 func (im *DynamoDBRegistry) ApplyChanges(ctx context.Context, changes *plan.Changes) error {
 	filteredChanges := &plan.Changes{
 		Create:    changes.Create,
-		UpdateNew: filterOwnedRecords(im.ownerID, changes.UpdateNew),
-		UpdateOld: filterOwnedRecords(im.ownerID, changes.UpdateOld),
-		Delete:    filterOwnedRecords(im.ownerID, changes.Delete),
+		UpdateNew: endpoint.FilterEndpointsByOwnerID(im.ownerID, changes.UpdateNew),
+		UpdateOld: endpoint.FilterEndpointsByOwnerID(im.ownerID, changes.UpdateOld),
+		Delete:    endpoint.FilterEndpointsByOwnerID(im.ownerID, changes.Delete),
 	}
 
 	statements := make([]*dynamodb.BatchStatementRequest, 0, len(filteredChanges.Create)+len(filteredChanges.UpdateNew))
diff --git a/registry/noop.go b/registry/noop.go
index eb10bb319..b58187405 100644
--- a/registry/noop.go
+++ b/registry/noop.go
@@ -40,6 +40,10 @@ func (im *NoopRegistry) GetDomainFilter() endpoint.DomainFilter {
 	return im.provider.GetDomainFilter()
 }
 
+func (im *NoopRegistry) OwnerID() string {
+	return ""
+}
+
 // Records returns the current records from the dns provider
 func (im *NoopRegistry) Records(ctx context.Context) ([]*endpoint.Endpoint, error) {
 	return im.provider.Records(ctx)
diff --git a/registry/registry.go b/registry/registry.go
index dc244e0d3..5b59a27b2 100644
--- a/registry/registry.go
+++ b/registry/registry.go
@@ -19,8 +19,6 @@ package registry
 import (
 	"context"
 
-	log "github.com/sirupsen/logrus"
-
 	"sigs.k8s.io/external-dns/endpoint"
 	"sigs.k8s.io/external-dns/plan"
 )
@@ -34,17 +32,5 @@ type Registry interface {
 	ApplyChanges(ctx context.Context, changes *plan.Changes) error
 	AdjustEndpoints(endpoints []*endpoint.Endpoint) []*endpoint.Endpoint
 	GetDomainFilter() endpoint.DomainFilter
-}
-
-// TODO(ideahitme): consider moving this to Plan
-func filterOwnedRecords(ownerID string, eps []*endpoint.Endpoint) []*endpoint.Endpoint {
-	filtered := []*endpoint.Endpoint{}
-	for _, ep := range eps {
-		if endpointOwner, ok := ep.Labels[endpoint.OwnerLabelKey]; !ok || endpointOwner != ownerID {
-			log.Debugf(`Skipping endpoint %v because owner id does not match, found: "%s", required: "%s"`, ep, endpointOwner, ownerID)
-			continue
-		}
-		filtered = append(filtered, ep)
-	}
-	return filtered
+	OwnerID() string
 }
diff --git a/registry/txt.go b/registry/txt.go
index 74a8faa59..3f37ff01c 100644
--- a/registry/txt.go
+++ b/registry/txt.go
@@ -97,6 +97,10 @@ func (im *TXTRegistry) GetDomainFilter() endpoint.DomainFilter {
 	return im.provider.GetDomainFilter()
 }
 
+func (im *TXTRegistry) OwnerID() string {
+	return im.ownerID
+}
+
 // Records returns the current records from the registry excluding TXT Records
 // If TXT records was created previously to indicate ownership its corresponding value
 // will be added to the endpoints Labels map
@@ -236,9 +240,9 @@ func (im *TXTRegistry) generateTXTRecord(r *endpoint.Endpoint) []*endpoint.Endpo
 func (im *TXTRegistry) ApplyChanges(ctx context.Context, changes *plan.Changes) error {
 	filteredChanges := &plan.Changes{
 		Create:    changes.Create,
-		UpdateNew: filterOwnedRecords(im.ownerID, changes.UpdateNew),
-		UpdateOld: filterOwnedRecords(im.ownerID, changes.UpdateOld),
-		Delete:    filterOwnedRecords(im.ownerID, changes.Delete),
+		UpdateNew: endpoint.FilterEndpointsByOwnerID(im.ownerID, changes.UpdateNew),
+		UpdateOld: endpoint.FilterEndpointsByOwnerID(im.ownerID, changes.UpdateOld),
+		Delete:    endpoint.FilterEndpointsByOwnerID(im.ownerID, changes.Delete),
 	}
 	for _, r := range filteredChanges.Create {
 		if r.Labels == nil {
diff --git a/registry/txt_test.go b/registry/txt_test.go
index 1c98b3843..dbcd6d2b7 100644
--- a/registry/txt_test.go
+++ b/registry/txt_test.go
@@ -1440,6 +1440,68 @@ func TestFailGenerateTXT(t *testing.T) {
 	assert.Equal(t, expectedTXT, gotTXT)
 }
 
+// TestMultiClusterDifferentRecordTypeOwnership validates the registry handles environments where the same zone is managed by
+// external-dns in different clusters and the ingress record type is different. For example one uses A records and the other
+// uses CNAME. In this environment the first cluster that establishes the owner record should maintain ownership even
+// if the same ingress host is deployed to the other. With the introduction of Dual Record support each record type
+// was treated independently and would cause each cluster to fight over ownership. This tests ensure that the default
+// Dual Stack record support only treats AAAA records independently and while keeping A and CNAME record ownership intact.
+func TestMultiClusterDifferentRecordTypeOwnership(t *testing.T) {
+	ctx := context.Background()
+	p := inmemory.NewInMemoryProvider()
+	p.CreateZone(testZone)
+	p.ApplyChanges(ctx, &plan.Changes{
+		Create: []*endpoint.Endpoint{
+			// records on cluster using A record for ingress address
+			newEndpointWithOwner("bar.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=cat,external-dns/resource=ingress/default/foo\"", endpoint.RecordTypeTXT, ""),
+			newEndpointWithOwner("bar.test-zone.example.org", "1.2.3.4", endpoint.RecordTypeA, ""),
+		},
+	})
+
+	r, _ := NewTXTRegistry(p, "_owner.", "", "bar", time.Hour, "", []string{}, false, nil)
+	records, _ := r.Records(ctx)
+
+	// new cluster has same ingress host as other cluster and uses CNAME ingress address
+	cname := &endpoint.Endpoint{
+		DNSName:    "bar.test-zone.example.org",
+		Targets:    endpoint.Targets{"cluster-b"},
+		RecordType: "CNAME",
+		Labels: map[string]string{
+			endpoint.ResourceLabelKey: "ingress/default/foo-127",
+		},
+	}
+	desired := []*endpoint.Endpoint{cname}
+
+	pl := &plan.Plan{
+		Policies:       []plan.Policy{&plan.SyncPolicy{}},
+		Current:        records,
+		Desired:        desired,
+		ManagedRecords: []string{endpoint.RecordTypeA, endpoint.RecordTypeAAAA, endpoint.RecordTypeCNAME},
+	}
+
+	changes := pl.Calculate()
+	p.OnApplyChanges = func(ctx context.Context, changes *plan.Changes) {
+		got := map[string][]*endpoint.Endpoint{
+			"Create":    changes.Create,
+			"UpdateNew": changes.UpdateNew,
+			"UpdateOld": changes.UpdateOld,
+			"Delete":    changes.Delete,
+		}
+		expected := map[string][]*endpoint.Endpoint{
+			"Create":    {},
+			"UpdateNew": {},
+			"UpdateOld": {},
+			"Delete":    {},
+		}
+		testutils.SamePlanChanges(got, expected)
+	}
+
+	err := r.ApplyChanges(ctx, changes.Changes)
+	if err != nil {
+		t.Error(err)
+	}
+}
+
 /**
 
 helper methods