output/aws_dynamodb: chunk batch writes to 25 item limit

DynamoDB BatchWriteItem API limits requests to 25 items. Previously,
exceeding this limit caused a fallback to individual writes. Now batches
are automatically chunked into groups of 25, with backoff reset between
chunks.

Two data-integrity paths are handled explicitly so a chunk failure
cannot silently drop later items:

* When a chunk's BatchWriteItem fails and the individual PutItem
  fallback recovers it, the loop now continues to the next chunk
  instead of returning nil early.
* When a chunk's unprocessed-items retry budget is exhausted, the
  writer returns a service.BatchError that maps each still-unprocessed
  item back to its original batch index via reflect.DeepEqual, with a
  pessimistic whole-chunk-failed fallback if the SDK returns a shape
  we do not recognize. Items in unattempted later chunks are also
  marked failed so upstream retry targets only the unwritten items.

Unit tests cover single-chunk happy path, multi-chunk happy path,
individual-fallback continuation to the next chunk, global-index
arithmetic for partial individual fallback in a non-zero chunk, the
unprocessed-items BatchError path, and the pessimistic fallback when
DeepEqual cannot match. An integration test against dynamodb-local
exercises multi-chunk writes end to end.

Closes #992

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: twmb

internal/impl/aws/dynamodb/output.go

@@ -20,6 +20,7 @@ import (
 	"errors"
 	"fmt"
 	"maps"
+	"reflect"
 	"strconv"
 	"sync"
 	"time"
@@ -38,6 +39,10 @@ import (
 )
 
 const (
+	// dynamoDBMaxBatchItems is the maximum number of items that can be sent in
+	// a single BatchWriteItem request per the AWS API.
+	dynamoDBMaxBatchItems = 25
+
 	// DynamoDB Output Fields
 	ddboField               = "namespace"
 	ddboFieldTable          = "table"
@@ -386,82 +391,137 @@ func (d *dynamoDBWriter) WriteBatch(ctx context.Context, b service.MessageBatch)
 		return err
 	}
 
-	batchResult, err := d.client.BatchWriteItem(ctx, &dynamodb.BatchWriteItemInput{
-		RequestItems: map[string][]types.WriteRequest{
-			*d.table: writeReqs,
-		},
-	})
-	if err != nil {
-		headlineErr := err
-
-		// None of the messages were successful, attempt to send individually
-	individualRequestsLoop:
-		for err != nil {
-			batchErr := service.NewBatchError(b, headlineErr)
-			for i, req := range writeReqs {
-				if req.PutRequest == nil {
-					continue
-				}
-				if _, iErr := d.client.PutItem(ctx, &dynamodb.PutItemInput{
-					TableName: d.table,
-					Item:      req.PutRequest.Item,
-				}); iErr != nil {
-					d.log.Errorf("Put error: %v\n", iErr)
-					wait := boff.NextBackOff()
-					if wait == backoff.Stop {
-						break individualRequestsLoop
+	// Chunk writeReqs into groups of dynamoDBMaxBatchItems (25) per the
+	// DynamoDB BatchWriteItem API limit.
+	for chunkStart := 0; chunkStart < len(writeReqs); chunkStart += dynamoDBMaxBatchItems {
+		boff.Reset()
+		chunkEnd := min(chunkStart+dynamoDBMaxBatchItems, len(writeReqs))
+		chunk := writeReqs[chunkStart:chunkEnd]
+
+		batchResult, err := d.client.BatchWriteItem(ctx, &dynamodb.BatchWriteItemInput{
+			RequestItems: map[string][]types.WriteRequest{
+				*d.table: chunk,
+			},
+		})
+		if err != nil {
+			headlineErr := err
+
+			// None of the messages were successful, attempt to send individually
+		individualRequestsLoop:
+			for err != nil {
+				batchErr := service.NewBatchError(b, headlineErr)
+				for j, req := range chunk {
+					if req.PutRequest == nil {
+						continue
 					}
-					select {
-					case <-time.After(wait):
-					case <-ctx.Done():
-						break individualRequestsLoop
+					if _, iErr := d.client.PutItem(ctx, &dynamodb.PutItemInput{
+						TableName: d.table,
+						Item:      req.PutRequest.Item,
+					}); iErr != nil {
+						d.log.Errorf("Put error: %v\n", iErr)
+						wait := boff.NextBackOff()
+						if wait == backoff.Stop {
+							break individualRequestsLoop
+						}
+						select {
+						case <-time.After(wait):
+						case <-ctx.Done():
+							break individualRequestsLoop
+						}
+						batchErr.Failed(chunkStart+j, iErr)
+					} else {
+						chunk[j].PutRequest = nil
 					}
-					batchErr.Failed(i, iErr)
+				}
+				if batchErr.IndexedErrors() == 0 {
+					err = nil
 				} else {
-					writeReqs[i].PutRequest = nil
+					// Mark all items in subsequent unattempted chunks as
+					// failed to prevent silent data loss.
+					for k := chunkEnd; k < len(writeReqs); k++ {
+						batchErr.Failed(k, headlineErr)
+					}
+					err = batchErr
 				}
 			}
-			if batchErr.IndexedErrors() == 0 {
-				err = nil
-			} else {
-				err = batchErr
+			if err != nil {
+				return err
 			}
+			// Individual fallback wrote every item in this chunk; move on.
+			continue
 		}
-		return err
-	}
 
-	unproc := batchResult.UnprocessedItems[*d.table]
-unprocessedLoop:
-	for len(unproc) > 0 {
-		wait := boff.NextBackOff()
-		if wait == backoff.Stop {
-			break unprocessedLoop
-		}
+		unproc := batchResult.UnprocessedItems[*d.table]
+	unprocessedLoop:
+		for len(unproc) > 0 {
+			wait := boff.NextBackOff()
+			if wait == backoff.Stop {
+				break unprocessedLoop
+			}
 
-		select {
-		case <-time.After(wait):
-		case <-ctx.Done():
-			break unprocessedLoop
-		}
-		if batchResult, err = d.client.BatchWriteItem(ctx, &dynamodb.BatchWriteItemInput{
-			RequestItems: map[string][]types.WriteRequest{
-				*d.table: unproc,
-			},
-		}); err != nil {
-			d.log.Errorf("Write multi error: %v\n", err)
-		} else if unproc = batchResult.UnprocessedItems[*d.table]; len(unproc) > 0 {
-			err = fmt.Errorf("setting %v items", len(unproc))
-		} else {
-			unproc = nil
+			select {
+			case <-time.After(wait):
+			case <-ctx.Done():
+				break unprocessedLoop
+			}
+			if batchResult, err = d.client.BatchWriteItem(ctx, &dynamodb.BatchWriteItemInput{
+				RequestItems: map[string][]types.WriteRequest{
+					*d.table: unproc,
+				},
+			}); err != nil {
+				d.log.Errorf("Write multi error: %v\n", err)
+			} else if unproc = batchResult.UnprocessedItems[*d.table]; len(unproc) > 0 {
+				err = fmt.Errorf("setting %v items", len(unproc))
+			} else {
+				unproc = nil
+			}
 		}
-	}
 
-	if len(unproc) > 0 {
-		if err == nil {
-			err = errors.New("ran out of request retries")
+		if len(unproc) > 0 {
+			if err == nil {
+				err = errors.New("ran out of request retries")
+			}
+			batchErr := service.NewBatchError(b, err)
+			// Map each unprocessed item back to its original index in
+			// this chunk so only the genuinely-unwritten items (and
+			// items in unattempted later chunks) are marked as failed.
+			matched := make([]bool, len(chunk))
+			matches := 0
+			for _, u := range unproc {
+				for j := range chunk {
+					if matched[j] {
+						continue
+					}
+					if reflect.DeepEqual(u, chunk[j]) {
+						matched[j] = true
+						matches++
+						break
+					}
+				}
+			}
+			if matches == len(unproc) {
+				for j, m := range matched {
+					if m {
+						batchErr.Failed(chunkStart+j, err)
+					}
+				}
+			} else {
+				// Some unprocessed items did not match anything in the
+				// chunk — the SDK returned a shape we did not expect. Fall
+				// back to marking the whole chunk failed. Writes are
+				// upserts, so retried items that already landed are
+				// harmless; the alternative is silent data loss.
+				for j := range chunk {
+					batchErr.Failed(chunkStart+j, err)
+				}
+			}
+			for k := chunkEnd; k < len(writeReqs); k++ {
+				batchErr.Failed(k, err)
+			}
+			return batchErr
 		}
 	}
-	return err
+	return nil
 }
 
 func (*dynamoDBWriter) Close(context.Context) error {

internal/impl/aws/dynamodb/output_integration_test.go

@@ -0,0 +1,125 @@
+// Copyright 2024 Redpanda Data, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package dynamodb
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
+	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/testcontainers/testcontainers-go"
+	"github.com/testcontainers/testcontainers-go/wait"
+
+	"github.com/redpanda-data/benthos/v4/public/service"
+	"github.com/redpanda-data/benthos/v4/public/service/integration"
+)
+
+func TestIntegrationDynamoDBOutput(t *testing.T) {
+	integration.CheckSkip(t)
+
+	ctr, err := testcontainers.Run(t.Context(), "amazon/dynamodb-local:latest",
+		testcontainers.WithExposedPorts("8000/tcp"),
+		testcontainers.WithWaitStrategy(
+			wait.ForListeningPort("8000/tcp").WithStartupTimeout(30*time.Second),
+		),
+	)
+	testcontainers.CleanupContainer(t, ctr)
+	require.NoError(t, err)
+
+	mp, err := ctr.MappedPort(t.Context(), "8000/tcp")
+	require.NoError(t, err)
+	dynamoPort := mp.Port()
+
+	const tableName = "output_batch_table"
+	require.Eventually(t, func() bool {
+		return createTable(t.Context(), t, dynamoPort, tableName) == nil
+	}, 30*time.Second, time.Second)
+
+	endpoint := fmt.Sprintf("http://localhost:%s", dynamoPort)
+
+	pConf, err := ddboOutputSpec().ParseYAML(fmt.Sprintf(`
+table: %s
+string_columns:
+  id: ${! json("id") }
+  content: ${! json("content") }
+region: us-east-1
+endpoint: %s
+credentials:
+  id: xxxxx
+  secret: xxxxx
+  token: xxxxx
+`, tableName, endpoint), nil)
+	require.NoError(t, err)
+
+	dConf, err := ddboConfigFromParsed(pConf)
+	require.NoError(t, err)
+
+	writer, err := newDynamoDBWriter(dConf, service.MockResources())
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, writer.Close(context.Background()))
+	})
+
+	require.NoError(t, writer.Connect(t.Context()))
+
+	const numMessages = 60
+	batch := make(service.MessageBatch, numMessages)
+	for i := range numMessages {
+		batch[i] = service.NewMessage([]byte(fmt.Sprintf(`{"id":"id-%d","content":"content-%d"}`, i, i)))
+	}
+
+	require.NoError(t, writer.WriteBatch(t.Context(), batch))
+
+	awsCfg, err := config.LoadDefaultConfig(t.Context(),
+		config.WithRegion("us-east-1"),
+		config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("xxxxx", "xxxxx", "xxxxx")),
+	)
+	require.NoError(t, err)
+	awsCfg.BaseEndpoint = &endpoint
+	client := dynamodb.NewFromConfig(awsCfg)
+
+	seenIDs := make(map[string]bool, numMessages)
+	var lastKey map[string]types.AttributeValue
+	for {
+		scanOut, err := client.Scan(t.Context(), &dynamodb.ScanInput{
+			TableName:         aws.String(tableName),
+			ExclusiveStartKey: lastKey,
+		})
+		require.NoError(t, err)
+		for _, item := range scanOut.Items {
+			idAttr, ok := item["id"].(*types.AttributeValueMemberS)
+			require.True(t, ok, "item missing string 'id' attribute")
+			seenIDs[idAttr.Value] = true
+		}
+		if len(scanOut.LastEvaluatedKey) == 0 {
+			break
+		}
+		lastKey = scanOut.LastEvaluatedKey
+	}
+
+	assert.Len(t, seenIDs, numMessages, "expected all %d items to be written", numMessages)
+	for i := range numMessages {
+		expectedID := fmt.Sprintf("id-%d", i)
+		assert.True(t, seenIDs[expectedID], "expected item %q to be present in table", expectedID)
+	}
+}