Repository: kubernetes-sigs/external-dns (language: go)

Issue:
Error after deleting many records in sync policy
<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!

-->

**What happened**:
I updated external-dns to 0.13.6 using bitnami helm chart to use dynamodb registry and set sync policy(upsert-only before), and added TXT to managed records types for migration.

After external-dns initialized and removed 104 legacy TXT records, external-dns crashed with the error below.
After restart, external-dns worked fine.
Maybe there's a bug logic after deleting many existing records.

```plain
time="2023-09-08T06:35:48Z" level=info msg="104 record(s) in zone DOMAIN.com. [Id: /hostedzone/Z02AAAAAAAAAAAA] were successfully updated"
panic: runtime error: slice bounds out of range [:25] with capacity 0

goroutine 1 [running]:
sigs.k8s.io/external-dns/registry.(*DynamoDBRegistry).executeStatements(0xc0015c0000, {0x3ed5450, 0xc0015b1a70}, {0xc0006b5c00?, 0x0?, 0x0?}, 0xc0014a0748)
        /bitnami/blacksmith-sandox/external-dns-0.13.6/src/github.com/kubernetes-incubator/external-dns/registry/dynamodb.go:475 +0x328
sigs.k8s.io/external-dns/registry.(*DynamoDBRegistry).ApplyChanges(0xc0015c0000, {0x3ed5450, 0xc0015b1a70}, 0xc0006ac8a0)
        /bitnami/blacksmith-sandox/external-dns-0.13.6/src/github.com/kubernetes-incubator/external-dns/registry/dynamodb.go:329 +0x1296
sigs.k8s.io/external-dns/controller.(*Controller).RunOnce(0xc001491bc0, {0x3ed53a8, 0xc00010ee60})
        /bitnami/blacksmith-sandox/external-dns-0.13.6/src/github.com/kubernetes-incubator/external-dns/controller/controller.go:231 +0x637
sigs.k8s.io/external-dns/controller.(*Controller).Run(0xc001491bc0?, {0x3ed53a8, 0xc00010ee60})
        /bitnami/blacksmith-sandox/external-dns-0.13.6/src/github.com/kubernetes-incubator/external-dns/controller/controller.go:313 +0xc5
main.main()
        /bitnami/blacksmith-sandox/external-dns-0.13.6/src/github.com/kubernetes-incubator/external-dns/main.go:466 +0x4705 
```

**What you expected to happen**:
external-dns works fine after deleting many existing records.

**How to reproduce it (as minimally and precisely as possible)**:
1. There's more than 100 records remained with `upsert-only` policy.
2. Update external-dns's config to use dynamodb registry, with `sync` policy, adding `TXT` to managed-record-types arg.
3. Newly started external-dns pod removed old txt records and then crash

**Anything else we need to know?**:

**Environment**: eks 1.26
- External-DNS version (use `external-dns --version`):`v 0.13.6` (bitnami image `docker.io/bitnami/external-dns:0.13.6-debian-11-r0`)
- DNS provider: aws route53
- Others: dynamodb registry enabled. sync policy, managed record types (A, AAAA, CNAME, TXT)



Discussion / hints:
I encountered the same problem when trying to create about 50 entries in DynamoDB. 
Had to create them in parts, like 10-15 entries, tuning regex-domain-filter.
The same issue here. I am trying to switch to from the txt registry to dynamo having about 200 ingresses in the dev cluster and about 2k ingresses in the live cluster
We now have about 7.5k records in Route53 because of making 3 records for 1 ingress, which is a huge problem. Dynamo seemed like a problem solver, but this bug stopped us from migrating...
@johngmyers could you please take a look at this? Thanks!

Fix patch:
diff --git a/registry/dynamodb.go b/registry/dynamodb.go
index a9732d4d46..aeb38d9a9a 100644
--- a/registry/dynamodb.go
+++ b/registry/dynamodb.go
@@ -68,6 +68,9 @@ type DynamoDBRegistry struct {
 
 const dynamodbAttributeMigrate = "dynamodb/needs-migration"
 
+// DynamoDB allows a maximum batch size of 25 items.
+var dynamodbMaxBatchSize uint8 = 25
+
 // NewDynamoDBRegistry returns a new DynamoDBRegistry object.
 func NewDynamoDBRegistry(provider provider.Provider, ownerID string, dynamodbAPI DynamoDBAPI, table string, txtPrefix, txtSuffix, txtWildcardReplacement string, managedRecordTypes, excludeRecordTypes []string, txtEncryptAESKey []byte, cacheInterval time.Duration) (*DynamoDBRegistry, error) {
 	if ownerID == "" {
@@ -477,9 +480,9 @@ func (im *DynamoDBRegistry) appendDelete(statements []*dynamodb.BatchStatementRe
 func (im *DynamoDBRegistry) executeStatements(ctx context.Context, statements []*dynamodb.BatchStatementRequest, handleErr func(request *dynamodb.BatchStatementRequest, response *dynamodb.BatchStatementResponse) error) error {
 	for len(statements) > 0 {
 		var chunk []*dynamodb.BatchStatementRequest
-		if len(statements) > 25 {
-			chunk = chunk[:25]
-			statements = statements[25:]
+		if len(statements) > int(dynamodbMaxBatchSize) {
+			chunk = statements[:dynamodbMaxBatchSize]
+			statements = statements[dynamodbMaxBatchSize:]
 		} else {
 			chunk = statements
 			statements = nil


Test patch:
diff --git a/registry/dynamodb_test.go b/registry/dynamodb_test.go
index 7c3531b355..1cadd7dcd8 100644
--- a/registry/dynamodb_test.go
+++ b/registry/dynamodb_test.go
@@ -217,6 +217,7 @@ func TestDynamoDBRegistryRecords(t *testing.T) {
 func TestDynamoDBRegistryApplyChanges(t *testing.T) {
 	for _, tc := range []struct {
 		name            string
+		maxBatchSize    uint8
 		stubConfig      DynamoDBStubConfig
 		addRecords      []*endpoint.Endpoint
 		changes         plan.Changes
@@ -294,6 +295,118 @@ func TestDynamoDBRegistryApplyChanges(t *testing.T) {
 				},
 			},
 		},
+		{
+			name:         "create more entries than DynamoDB batch size limit",
+			maxBatchSize: 2,
+			changes: plan.Changes{
+				Create: []*endpoint.Endpoint{
+					{
+						DNSName:       "new1.test-zone.example.org",
+						Targets:       endpoint.Targets{"new1.loadbalancer.com"},
+						RecordType:    endpoint.RecordTypeCNAME,
+						SetIdentifier: "set-new",
+						Labels: map[string]string{
+							endpoint.ResourceLabelKey: "ingress/default/new1-ingress",
+						},
+					},
+					{
+						DNSName:       "new2.test-zone.example.org",
+						Targets:       endpoint.Targets{"new2.loadbalancer.com"},
+						RecordType:    endpoint.RecordTypeCNAME,
+						SetIdentifier: "set-new",
+						Labels: map[string]string{
+							endpoint.ResourceLabelKey: "ingress/default/new2-ingress",
+						},
+					},
+					{
+						DNSName:       "new3.test-zone.example.org",
+						Targets:       endpoint.Targets{"new3.loadbalancer.com"},
+						RecordType:    endpoint.RecordTypeCNAME,
+						SetIdentifier: "set-new",
+						Labels: map[string]string{
+							endpoint.ResourceLabelKey: "ingress/default/new3-ingress",
+						},
+					},
+				},
+			},
+			stubConfig: DynamoDBStubConfig{
+				ExpectInsert: map[string]map[string]string{
+					"new1.test-zone.example.org#CNAME#set-new": {endpoint.ResourceLabelKey: "ingress/default/new1-ingress"},
+					"new2.test-zone.example.org#CNAME#set-new": {endpoint.ResourceLabelKey: "ingress/default/new2-ingress"},
+					"new3.test-zone.example.org#CNAME#set-new": {endpoint.ResourceLabelKey: "ingress/default/new3-ingress"},
+				},
+				ExpectDelete: sets.New("quux.test-zone.example.org#A#set-2"),
+			},
+			expectedRecords: []*endpoint.Endpoint{
+				{
+					DNSName:    "foo.test-zone.example.org",
+					Targets:    endpoint.Targets{"foo.loadbalancer.com"},
+					RecordType: endpoint.RecordTypeCNAME,
+					Labels: map[string]string{
+						endpoint.OwnerLabelKey: "",
+					},
+				},
+				{
+					DNSName:    "bar.test-zone.example.org",
+					Targets:    endpoint.Targets{"my-domain.com"},
+					RecordType: endpoint.RecordTypeCNAME,
+					Labels: map[string]string{
+						endpoint.OwnerLabelKey:    "test-owner",
+						endpoint.ResourceLabelKey: "ingress/default/my-ingress",
+					},
+				},
+				{
+					DNSName:       "baz.test-zone.example.org",
+					Targets:       endpoint.Targets{"1.1.1.1"},
+					RecordType:    endpoint.RecordTypeA,
+					SetIdentifier: "set-1",
+					Labels: map[string]string{
+						endpoint.OwnerLabelKey:    "test-owner",
+						endpoint.ResourceLabelKey: "ingress/default/my-ingress",
+					},
+				},
+				{
+					DNSName:       "baz.test-zone.example.org",
+					Targets:       endpoint.Targets{"2.2.2.2"},
+					RecordType:    endpoint.RecordTypeA,
+					SetIdentifier: "set-2",
+					Labels: map[string]string{
+						endpoint.OwnerLabelKey:    "test-owner",
+						endpoint.ResourceLabelKey: "ingress/default/other-ingress",
+					},
+				},
+				{
+					DNSName:       "new1.test-zone.example.org",
+					Targets:       endpoint.Targets{"new1.loadbalancer.com"},
+					RecordType:    endpoint.RecordTypeCNAME,
+					SetIdentifier: "set-new",
+					Labels: map[string]string{
+						endpoint.OwnerLabelKey:    "test-owner",
+						endpoint.ResourceLabelKey: "ingress/default/new1-ingress",
+					},
+				},
+				{
+					DNSName:       "new2.test-zone.example.org",
+					Targets:       endpoint.Targets{"new2.loadbalancer.com"},
+					RecordType:    endpoint.RecordTypeCNAME,
+					SetIdentifier: "set-new",
+					Labels: map[string]string{
+						endpoint.OwnerLabelKey:    "test-owner",
+						endpoint.ResourceLabelKey: "ingress/default/new2-ingress",
+					},
+				},
+				{
+					DNSName:       "new3.test-zone.example.org",
+					Targets:       endpoint.Targets{"new3.loadbalancer.com"},
+					RecordType:    endpoint.RecordTypeCNAME,
+					SetIdentifier: "set-new",
+					Labels: map[string]string{
+						endpoint.OwnerLabelKey:    "test-owner",
+						endpoint.ResourceLabelKey: "ingress/default/new3-ingress",
+					},
+				},
+			},
+		},
 		{
 			name: "create orphaned",
 			changes: plan.Changes{
@@ -911,6 +1024,11 @@ func TestDynamoDBRegistryApplyChanges(t *testing.T) {
 		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
+			originalMaxBatchSize := dynamodbMaxBatchSize
+			if tc.maxBatchSize > 0 {
+				dynamodbMaxBatchSize = tc.maxBatchSize
+			}
+
 			api, p := newDynamoDBAPIStub(t, &tc.stubConfig)
 			if len(tc.addRecords) > 0 {
 				_ = p.(*wrappedProvider).Provider.ApplyChanges(context.Background(), &plan.Changes{
@@ -945,6 +1063,8 @@ func TestDynamoDBRegistryApplyChanges(t *testing.T) {
 			if tc.expectedError == "" {
 				assert.Empty(t, r.orphanedLabels)
 			}
+
+			dynamodbMaxBatchSize = originalMaxBatchSize
 		})
 	}
 }