diff --git a/controller/controller.go b/controller/controller.go
index af3e3219a2..65e73d336d 100644
--- a/controller/controller.go
+++ b/controller/controller.go
@@ -223,6 +223,7 @@ func (c *Controller) RunOnce(ctx context.Context) error {
 		DomainFilter:       endpoint.MatchAllDomainFilters{c.DomainFilter, c.Registry.GetDomainFilter()},
 		PropertyComparator: c.Registry.PropertyValuesEqual,
 		ManagedRecords:     c.ManagedRecordTypes,
+		OwnedRecordFilter:  c.Registry.GetOwnedRecordFilter(),
 	}
 
 	plan = plan.Calculate()
diff --git a/endpoint/endpoint.go b/endpoint/endpoint.go
index 7df3f882f9..ae7a93393d 100644
--- a/endpoint/endpoint.go
+++ b/endpoint/endpoint.go
@@ -281,6 +281,44 @@ 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)
 }
 
+// EndpointFilterInterface matches endpoints
+type EndpointFilterInterface interface {
+	// Match returns true if the endpoint matches the filter, false otherwise
+	Match(ep *Endpoint) bool
+}
+
+// Apply filter to slice of endpoints and return new filtered slice that includes
+// only endpoints that match.
+func ApplyEndpointFilter(filter EndpointFilterInterface, eps []*Endpoint) []*Endpoint {
+	filtered := []*Endpoint{}
+	for _, ep := range eps {
+		if filter.Match(ep) {
+			filtered = append(filtered, ep)
+		}
+	}
+
+	return filtered
+}
+
+// NewOwnedRecordFilter returns endpoint filter that matches records with a owner
+// label that matches the given owner id.
+func NewOwnedRecordFilter(ownerID string) EndpointFilterInterface {
+	return ownedRecordFilter{ownerID: ownerID}
+}
+
+type ownedRecordFilter struct {
+	ownerID string
+}
+
+func (f ownedRecordFilter) Match(ep *Endpoint) bool {
+	if endpointOwner, ok := ep.Labels[OwnerLabelKey]; !ok || endpointOwner != f.ownerID {
+		log.Debugf(`Skipping endpoint %v because owner id does not match, found: "%s", required: "%s"`, ep, endpointOwner, f.ownerID)
+		return false
+	} else {
+		return true
+	}
+}
+
 // 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 81e7f4c331..8b2dd2b2ff 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,128 @@ func TestIsLess(t *testing.T) {
 		}
 	}
 }
+
+func TestOwnedRecordFilterMatch(t *testing.T) {
+	type fields struct {
+		ownerID string
+	}
+	type args struct {
+		ep *Endpoint
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   bool
+	}{
+		{
+			name:   "no labels",
+			fields: fields{ownerID: "foo"},
+			args: args{ep: &Endpoint{
+				DNSName:    "foo.com",
+				RecordType: RecordTypeA,
+			}},
+			want: false,
+		},
+		{
+			name:   "no owner label",
+			fields: fields{ownerID: "foo"},
+			args: args{ep: &Endpoint{
+				DNSName:    "foo.com",
+				RecordType: RecordTypeA,
+				Labels:     NewLabels(),
+			}},
+			want: false,
+		},
+		{
+			name:   "owner not matched",
+			fields: fields{ownerID: "foo"},
+			args: args{ep: &Endpoint{
+				DNSName:    "foo.com",
+				RecordType: RecordTypeA,
+				Labels: Labels{
+					OwnerLabelKey: "bar",
+				},
+			}},
+			want: false,
+		},
+		{
+			name:   "owner matched",
+			fields: fields{ownerID: "foo"},
+			args: args{ep: &Endpoint{
+				DNSName:    "foo.com",
+				RecordType: RecordTypeA,
+				Labels: Labels{
+					OwnerLabelKey: "foo",
+				},
+			}},
+			want: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			f := ownedRecordFilter{
+				ownerID: tt.fields.ownerID,
+			}
+			if got := f.Match(tt.args.ep); got != tt.want {
+				t.Errorf("ownedRecordFilter.Match() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestApplyEndpointFilter(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 {
+		filter EndpointFilterInterface
+		eps    []*Endpoint
+	}
+	tests := []struct {
+		name string
+		args args
+		want []*Endpoint
+	}{
+		{
+			name: "filter values",
+			args: args{
+				filter: NewOwnedRecordFilter("foo"),
+				eps: []*Endpoint{
+					foo1,
+					foo2,
+					bar,
+				},
+			},
+			want: []*Endpoint{
+				foo1,
+				foo2,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := ApplyEndpointFilter(tt.args.filter, tt.args.eps); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("ApplyEndpointFilter() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
diff --git a/plan/plan.go b/plan/plan.go
index 312bb261b1..9acfe3bc2b 100644
--- a/plan/plan.go
+++ b/plan/plan.go
@@ -48,6 +48,8 @@ type Plan struct {
 	PropertyComparator PropertyComparator
 	// DNS record types that will be considered for management
 	ManagedRecords []string
+	// Optional record filter that matches records owned by the registry
+	OwnedRecordFilter endpoint.EndpointFilterInterface
 }
 
 // Changes holds lists of actions to be executed by dns providers
@@ -66,22 +68,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
@@ -91,37 +97,88 @@ 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 1 record may
+	// be represented here, for example A and AAAA. If current domain record is 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
 }
 
+// hasCandidateRecordTypeConflict returns true if the candidates set contains conflicting or invalid record types.
+// 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.
+//
+// [RFC 1034 3.6.2]: https://datatracker.ietf.org/doc/html/rfc1034#autoid-15
+func (t planTableRow) hasCandidateRecordTypeConflict() bool {
+	if len(t.candidates) <= 1 {
+		return false
+	}
+
+	cname := false
+	other := false
+	for _, c := range t.candidates {
+		if c.RecordType == endpoint.RecordTypeCNAME {
+			cname = true
+		} else {
+			other = true
+		}
+
+		if cname && other {
+			return true
+		}
+	}
+
+	return false
+}
+
 func (t planTableRow) String() string {
 	return fmt.Sprintf("planTableRow{current=%v, candidates=%v}", t.current, t.candidates)
 }
 
 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
 }
 
@@ -151,30 +208,90 @@ 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))
+	for key, row := range t.rows {
+		// dns name not taken
+		if len(row.current) == 0 {
+			// TODO how to resolve conflicting source candidate record types
+			if row.hasCandidateRecordTypeConflict() {
+				log.Warnf("Domain %s contains conflicting record type candidates, no updates planned", key.dnsName)
+				continue
+			}
+
+			for _, records := range row.records {
+				changes.Create = append(changes.Create, t.resolver.ResolveCreate(records.candidates))
+			}
 		}
-		if row.current != nil && len(row.candidates) == 0 {
-			changes.Delete = append(changes.Delete, 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...)
 		}
 
-		// 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 is taken
+		if len(row.current) > 0 && len(row.candidates) > 0 {
+			// TODO how to resolve conflicting source candidate record types
+			if row.hasCandidateRecordTypeConflict() {
+				log.Warnf("Domain %s contains conflicting record type candidates, no updates planned", key.dnsName)
+				continue
+			}
+
+			creates := []*endpoint.Endpoint{}
+
+			// apply changes for each record type
+			for _, records := range row.records {
+				// 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.OwnedRecordFilter != nil && !p.OwnedRecordFilter.Match(current) {
+						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.OwnedRecordFilter != nil {
+		changes.Delete = endpoint.ApplyEndpointFilter(p.OwnedRecordFilter, changes.Delete)
+		changes.UpdateOld = endpoint.ApplyEndpointFilter(p.OwnedRecordFilter, changes.UpdateOld)
+		changes.UpdateNew = endpoint.ApplyEndpointFilter(p.OwnedRecordFilter, changes.UpdateNew)
+	}
+
 	plan := &Plan{
 		Current:        p.Current,
 		Desired:        p.Desired,
diff --git a/plan/plan_test.go b/plan/plan_test.go
index cc9e56bd5a..224329d595 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",
@@ -444,19 +444,205 @@ 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},
+		OwnedRecordFilter: endpoint.NewOwnedRecordFilter(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},
+		OwnedRecordFilter: endpoint.NewOwnedRecordFilter(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},
+		OwnedRecordFilter: endpoint.NewOwnedRecordFilter(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.RecordTypeAAAA, endpoint.RecordTypeCNAME},
+		OwnedRecordFilter: endpoint.NewOwnedRecordFilter("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},
+		OwnedRecordFilter: endpoint.NewOwnedRecordFilter(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},
+		OwnedRecordFilter: endpoint.NewOwnedRecordFilter(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 there are
+// no changes planned since there is no conflict resolver for this situation.
+func (suite *PlanTestSuite) TestCurrentWithConflictingDesired() {
+	suite.fooA5.Labels[endpoint.OwnerLabelKey] = "nerf"
+	suite.fooAAAA.Labels[endpoint.OwnerLabelKey] = "nerf"
+	current := []*endpoint.Endpoint{suite.fooA5, suite.fooAAAA}
+	desired := []*endpoint.Endpoint{suite.fooV1Cname, suite.fooA5, suite.fooAAAA}
+	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.RecordTypeAAAA, endpoint.RecordTypeCNAME},
+		OwnedRecordFilter: endpoint.NewOwnedRecordFilter(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)
+}
+
+// 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 are
+// no changes planned since there is no conflict resolver for this situation.
+func (suite *PlanTestSuite) TestNoCurrentWithConflictingDesired() {
+	current := []*endpoint.Endpoint{}
+	desired := []*endpoint.Endpoint{suite.fooV1Cname, suite.fooA5, suite.fooAAAA}
+	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},
 	}
 
 	changes := p.Calculate().Changes
@@ -644,10 +830,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{}},
@@ -658,12 +844,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{}},
@@ -674,6 +864,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) {
@@ -687,6 +920,14 @@ func validateEntries(t *testing.T, entries, expected []*endpoint.Endpoint) {
 	}
 }
 
+func validateOwner(t *testing.T, entries []*endpoint.Endpoint, owner string) {
+	for _, entry := range entries {
+		if entry.Labels[endpoint.OwnerLabelKey] != owner {
+			t.Fatalf("expected owner label %q to match %q", entry.Labels[endpoint.OwnerLabelKey], owner)
+		}
+	}
+}
+
 func TestNormalizeDNSName(t *testing.T) {
 	records := []struct {
 		dnsName string
diff --git a/registry/aws_sd_registry.go b/registry/aws_sd_registry.go
index 1ecd885463..93403629e8 100644
--- a/registry/aws_sd_registry.go
+++ b/registry/aws_sd_registry.go
@@ -46,6 +46,10 @@ func (sdr *AWSSDRegistry) GetDomainFilter() endpoint.DomainFilterInterface {
 	return sdr.provider.GetDomainFilter()
 }
 
+func (im *AWSSDRegistry) GetOwnedRecordFilter() endpoint.EndpointFilterInterface {
+	return endpoint.NewOwnedRecordFilter(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) {
@@ -70,11 +74,12 @@ func (sdr *AWSSDRegistry) Records(ctx context.Context) ([]*endpoint.Endpoint, er
 // ApplyChanges filters out records not owned the External-DNS, additionally it adds the required label
 // inserted in the AWS SD instance as a CreateID field
 func (sdr *AWSSDRegistry) ApplyChanges(ctx context.Context, changes *plan.Changes) error {
+	ownedRecordFilter := sdr.GetOwnedRecordFilter()
 	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.ApplyEndpointFilter(ownedRecordFilter, changes.UpdateNew),
+		UpdateOld: endpoint.ApplyEndpointFilter(ownedRecordFilter, changes.UpdateOld),
+		Delete:    endpoint.ApplyEndpointFilter(ownedRecordFilter, changes.Delete),
 	}
 
 	sdr.updateLabels(filteredChanges.Create)
diff --git a/registry/dynamodb.go b/registry/dynamodb.go
index 6507af6507..d2dc754145 100644
--- a/registry/dynamodb.go
+++ b/registry/dynamodb.go
@@ -81,6 +81,10 @@ func (im *DynamoDBRegistry) GetDomainFilter() endpoint.DomainFilterInterface {
 	return im.provider.GetDomainFilter()
 }
 
+func (im *DynamoDBRegistry) GetOwnedRecordFilter() endpoint.EndpointFilterInterface {
+	return endpoint.NewOwnedRecordFilter(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
@@ -128,11 +132,12 @@ func (im *DynamoDBRegistry) Records(ctx context.Context) ([]*endpoint.Endpoint,
 
 // ApplyChanges updates the DNS provider and DynamoDB table with the changes.
 func (im *DynamoDBRegistry) ApplyChanges(ctx context.Context, changes *plan.Changes) error {
+	ownedRecordFilter := im.GetOwnedRecordFilter()
 	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.ApplyEndpointFilter(ownedRecordFilter, changes.UpdateNew),
+		UpdateOld: endpoint.ApplyEndpointFilter(ownedRecordFilter, changes.UpdateOld),
+		Delete:    endpoint.ApplyEndpointFilter(ownedRecordFilter, changes.Delete),
 	}
 
 	statements := make([]*dynamodb.BatchStatementRequest, 0, len(filteredChanges.Create)+len(filteredChanges.UpdateNew))
diff --git a/registry/noop.go b/registry/noop.go
index 73257730cd..c2dc8bb510 100644
--- a/registry/noop.go
+++ b/registry/noop.go
@@ -40,6 +40,10 @@ func (im *NoopRegistry) GetDomainFilter() endpoint.DomainFilterInterface {
 	return im.provider.GetDomainFilter()
 }
 
+func (im *NoopRegistry) GetOwnedRecordFilter() endpoint.EndpointFilterInterface {
+	return nil
+}
+
 // 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 fa39fb8ec0..d6931fbf3a 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"
 )
@@ -35,17 +33,6 @@ type Registry interface {
 	PropertyValuesEqual(attribute string, previous string, current string) bool
 	AdjustEndpoints(endpoints []*endpoint.Endpoint) []*endpoint.Endpoint
 	GetDomainFilter() endpoint.DomainFilterInterface
-}
-
-// 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
+	// Evaluate if endpoint id owned by the registry and return true.
+	GetOwnedRecordFilter() endpoint.EndpointFilterInterface
 }
diff --git a/registry/txt.go b/registry/txt.go
index 3300b5ef1b..ec45f07aa9 100644
--- a/registry/txt.go
+++ b/registry/txt.go
@@ -100,6 +100,10 @@ func (im *TXTRegistry) GetDomainFilter() endpoint.DomainFilterInterface {
 	return im.provider.GetDomainFilter()
 }
 
+func (im *TXTRegistry) GetOwnedRecordFilter() endpoint.EndpointFilterInterface {
+	return endpoint.NewOwnedRecordFilter(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
@@ -227,11 +231,12 @@ func (im *TXTRegistry) generateTXTRecord(r *endpoint.Endpoint) []*endpoint.Endpo
 // ApplyChanges updates dns provider with the changes
 // for each created/deleted record it will also take into account TXT records for creation/deletion
 func (im *TXTRegistry) ApplyChanges(ctx context.Context, changes *plan.Changes) error {
+	ownedRecordFilter := im.GetOwnedRecordFilter()
 	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.ApplyEndpointFilter(ownedRecordFilter, changes.UpdateNew),
+		UpdateOld: endpoint.ApplyEndpointFilter(ownedRecordFilter, changes.UpdateOld),
+		Delete:    endpoint.ApplyEndpointFilter(ownedRecordFilter, changes.Delete),
 	}
 	for _, r := range filteredChanges.Create {
 		if r.Labels == nil {
diff --git a/registry/txt_test.go b/registry/txt_test.go
index 71e56b5b75..6a86081f16 100644
--- a/registry/txt_test.go
+++ b/registry/txt_test.go
@@ -1306,6 +1306,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