@@ -61,6 +61,7 @@ public enum BusinessMetricFeatureId {
     HTTP_CLIENT_EXPLICIT_INSTANCE("AK"),
     HTTP_CLIENT_EXPLICIT_FACTORY("AL"),
     DDB_MAPPER("d"),
+    DYNAMODB_MAPPER_V2("Ac"),
     BEARER_SERVICE_ENV_VARS("3"),
     CREDENTIALS_CODE("e"),
     CREDENTIALS_JVM_SYSTEM_PROPERTIES("f"),
 
@@ -21,7 +21,7 @@
     <parent>
         <groupId>software.amazon.awssdk</groupId>
         <artifactId>services-custom</artifactId>
-        <version>2.55.5-SNAPSHOT</version>
+        <version>2.55.14-SNAPSHOT</version>
     </parent>
     <artifactId>dynamodb-mapper</artifactId>
     <name>AWS Java SDK :: DynamoDB :: Mapper</name>
 
@@ -95,6 +95,7 @@
 import java.util.Set;
 import java.util.function.Consumer;
 
+import static software.amazon.awssdk.mapper.dynamodb.internal.MapperUserAgent.applyUserAgent;
 import static software.amazon.awssdk.services.dynamodb.model.KeyType.HASH;
 import static software.amazon.awssdk.services.dynamodb.model.KeyType.RANGE;
 import static software.amazon.awssdk.mapper.dynamodb.TransactionWriteRequest.TransactionWriteOperation;
@@ -446,7 +447,7 @@ public <T extends Object> T load(T keyObject, DynamoDBMapperConfig config) {
 
         Map<String, AttributeValue> key = model.convertKey(keyObject);
 
-        GetItemRequest rq = GetItemRequest.builder()
+        GetItemRequest rq = applyUserAgent(GetItemRequest.builder())
             .key(key)
             .tableName(tableName)
             .consistentRead(config.getConsistentReads() == ConsistentReads.CONSISTENT)
@@ -876,7 +877,7 @@ protected List<ValueUpdate> getInMemoryUpdates() {
          * the returned attributes to detect silent failure on the server-side.
          */
         protected UpdateItemResponse doUpdateItem() {
-            UpdateItemRequest req = UpdateItemRequest.builder()
+            UpdateItemRequest req = applyUserAgent(UpdateItemRequest.builder())
                     .tableName(getTableName())
                     .key(getPrimaryKeyAttributeValues())
                     .attributeUpdates(
@@ -913,7 +914,7 @@ protected PutItemResponse doPutItem() {
                                  this.clazz,
                                  getTableName(),
                                  saveConfig));
-            PutItemRequest req = PutItemRequest.builder()
+            PutItemRequest req = applyUserAgent(PutItemRequest.builder())
                     .tableName(getTableName())
                     .item(attributeValues)
                     .expected(mergeExpectedAttributeValueConditions())
@@ -1084,7 +1085,7 @@ public <T> void delete(T object, DynamoDBDeleteExpression deleteExpression, Dyna
             }
         }
 
-        DeleteItemRequest.Builder reqBuilder = DeleteItemRequest.builder()
+        DeleteItemRequest.Builder reqBuilder = applyUserAgent(DeleteItemRequest.builder())
                 .key(key)
                 .tableName(tableName);
 
@@ -1141,7 +1142,7 @@ public void transactionWrite(TransactionWriteRequest transactionWriteRequest, Dy
 
         }
 
-        TransactWriteItemsRequest transactWriteItemsRequest = TransactWriteItemsRequest.builder()
+        TransactWriteItemsRequest transactWriteItemsRequest = applyUserAgent(TransactWriteItemsRequest.builder())
                 .clientRequestToken(transactionWriteRequest.getIdempotencyToken())
                 .transactItems(transactWriteItems)
                 .build();
@@ -1189,7 +1190,7 @@ public List<Object> transactionLoad(TransactionLoadRequest transactionLoadReques
             transactGetItems.add(TransactGetItem.builder().get(getItem.build()).build());
         }
 
-        TransactGetItemsRequest transactGetItemsRequest = TransactGetItemsRequest.builder()
+        TransactGetItemsRequest transactGetItemsRequest = applyUserAgent(TransactGetItemsRequest.builder())
                 .transactItems(transactGetItems)
                 .build();
         TransactGetItemsResponse transactGetItemsResult = db.transactGetItems(transactGetItemsRequest);
@@ -1359,7 +1360,7 @@ private FailedBatch doBatchWriteItemWithRetry(
 
         while (true) {
             try {
-                result = db.batchWriteItem(BatchWriteItemRequest.builder().requestItems(pendingItems).build());
+                result = db.batchWriteItem(applyUserAgent(BatchWriteItemRequest.builder()).requestItems(pendingItems).build());
             } catch (Exception e) {
                 failedBatch = new FailedBatch();
                 failedBatch.setUnprocessedItems(pendingItems);
@@ -1472,7 +1473,7 @@ private void processBatchGetRequest(
             final DynamoDBMapperConfig config) {
 
         BatchGetItemResponse batchGetItemResult = null;
-        BatchGetItemRequest batchGetItemRequest = BatchGetItemRequest.builder()
+        BatchGetItemRequest batchGetItemRequest = applyUserAgent(BatchGetItemRequest.builder())
             .requestItems(requestItems)
             .build();
 
@@ -1488,6 +1489,7 @@ private void processBatchGetRequest(
                 batchLoadContext.setRetriesAttempted(retries);
                 if (!isNullOrEmpty(batchGetItemResult.unprocessedKeys())) {
                     pause(batchLoadStrategy.getDelayBeforeNextRetry(batchLoadContext));
+                    // toBuilder() keeps the user agent tag on the retry of the unprocessed keys.
                     batchGetItemRequest = batchGetItemRequest.toBuilder()
                         .requestItems(batchGetItemResult.unprocessedKeys())
                         .build();
@@ -1731,7 +1733,9 @@ public <T> int count(Class<T> clazz, DynamoDBQueryExpression<T> queryExpression,
      * @param config never null
      */
     private ScanRequest createScanRequestFromExpression(Class<?> clazz, DynamoDBScanExpression scanExpression, DynamoDBMapperConfig config) {
-        return ScanRequest.builder()
+        // Every scan path (scan, scanPage, count, parallelScan and later pages) derives its requests from this one
+        // with toBuilder(), which keeps the user agent tag.
+        return applyUserAgent(ScanRequest.builder())
                 .tableName(getTableName(clazz, config))
                 .indexName(scanExpression.getIndexName())
                 .scanFilter(scanExpression.getScanFilter())
@@ -1781,7 +1785,9 @@ protected <T> QueryRequest createQueryRequestFromExpression(Class<T> clazz,
 
         DynamoDBMapperTableModel<T> model = getTableModel(clazz, config);
 
-        QueryRequest.Builder req = QueryRequest.builder();
+        // Every query path (query, queryPage, count and later pages) derives its requests from this one with
+        // toBuilder(), which keeps the user agent tag.
+        QueryRequest.Builder req = applyUserAgent(QueryRequest.builder());
         req.consistentRead(xpress.isConsistentRead());
         req.tableName(getTableName(clazz, xpress.getHashKeyValues(), config));
         req.indexName(xpress.getIndexName());
 
@@ -15,6 +15,8 @@
 
 package software.amazon.awssdk.mapper.dynamodb;
 
+import static software.amazon.awssdk.mapper.dynamodb.internal.MapperUserAgent.applyUserAgent;
+
 import software.amazon.awssdk.annotations.SdkPublicApi;
 import java.util.List;
 
@@ -85,7 +87,9 @@ protected boolean atEndOfResults() {
 
     @Override
     protected synchronized List<T> fetchNextPage() {
-        queryRequest = queryRequest.toBuilder().exclusiveStartKey(queryResult.lastEvaluatedKey()).build();
+        // Tag here too: the public constructor accepts a request the mapper did not build. applyUserAgent keeps the
+        // caller's override configuration and adds the tag only once.
+        queryRequest = applyUserAgent(queryRequest.toBuilder()).exclusiveStartKey(queryResult.lastEvaluatedKey()).build();
         queryResult = dynamo.query(queryRequest);
         return mapper.marshallIntoObjects(mapper.toParameters(
                 queryResult.items(),
 
@@ -15,6 +15,8 @@
 
 package software.amazon.awssdk.mapper.dynamodb;
 
+import static software.amazon.awssdk.mapper.dynamodb.internal.MapperUserAgent.applyUserAgent;
+
 import software.amazon.awssdk.annotations.SdkPublicApi;
 import java.util.List;
 
@@ -84,7 +86,9 @@ protected boolean atEndOfResults() {
 
     @Override
     protected synchronized List<T> fetchNextPage() {
-        scanRequest = scanRequest.toBuilder().exclusiveStartKey(scanResult.lastEvaluatedKey()).build();
+        // Tag here too: the public constructor accepts a request the mapper did not build. applyUserAgent keeps the
+        // caller's override configuration and adds the tag only once.
+        scanRequest = applyUserAgent(scanRequest.toBuilder()).exclusiveStartKey(scanResult.lastEvaluatedKey()).build();
         scanResult = dynamo.scan(scanRequest);
         return mapper.marshallIntoObjects(mapper.toParameters(
             scanResult.items(),
 
@@ -15,6 +15,8 @@
 
 package software.amazon.awssdk.mapper.dynamodb;
 
+import static software.amazon.awssdk.mapper.dynamodb.internal.MapperUserAgent.applyUserAgent;
+
 import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.core.exception.SdkClientException;
 import software.amazon.awssdk.core.exception.SdkException;
@@ -226,11 +228,14 @@ else if (currentSegmentState == SegmentScanState.Waiting
 
     private ScanResponse scanNextPageOfSegment(int currentSegment, boolean checkLastEvaluatedKey) {
         ScanRequest segmentScanRequest = parallelScanRequests.get(currentSegment);
+        // The mapper already tags the segment requests. Like v1, this also tags each segment page here, which costs
+        // nothing because the builder already exists.
         if (checkLastEvaluatedKey) {
             ScanResponse lastScanResult = segmentScanResults.get(currentSegment);
-            segmentScanRequest = segmentScanRequest.toBuilder().exclusiveStartKey(lastScanResult.lastEvaluatedKey()).build();
+            segmentScanRequest = applyUserAgent(segmentScanRequest.toBuilder())
+                    .exclusiveStartKey(lastScanResult.lastEvaluatedKey()).build();
         } else {
-            segmentScanRequest = segmentScanRequest.toBuilder().exclusiveStartKey(null).build();
+            segmentScanRequest = applyUserAgent(segmentScanRequest.toBuilder()).exclusiveStartKey(null).build();
         }
         ScanResponse scanResult = dynamo.scan(segmentScanRequest);
 
 
@@ -0,0 +1,80 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file 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 software.amazon.awssdk.mapper.dynamodb.internal;
+
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.awscore.AwsRequestOverrideConfiguration;
+import software.amazon.awssdk.core.ApiName;
+import software.amazon.awssdk.core.useragent.BusinessMetricFeatureId;
+import software.amazon.awssdk.services.dynamodb.model.DynamoDbRequest;
+
+/**
+ * Tags the DynamoDB requests that the mapper builds with the {@link BusinessMetricFeatureId#DYNAMODB_MAPPER_V2}
+ * business metric, so the metric is only reported for requests the mapper itself sends. This is the v2 port of v1's
+ * {@code DynamoDBMapper.applyUserAgent}.
+ * <p>
+ * v1 used separate user agent strings for batch and transaction operations. Only one feature ID is reserved for the
+ * v2 mapper, so every mapper request, including batch and transaction requests, carries the same metric.
+ */
+@SdkInternalApi
+public final class MapperUserAgent {
+
+    private static final ApiName API_NAME = ApiName.builder()
+                                                   .name("sdk-metrics")
+                                                   .version(BusinessMetricFeatureId.DYNAMODB_MAPPER_V2.value())
+                                                   .build();
+
+    /**
+     * Shared, immutable override configuration carrying only the mapper's {@code sdk-metrics} API name. It is used
+     * as-is for builders that have no override configuration, which covers every request the mapper builds itself.
+     */
+    private static final AwsRequestOverrideConfiguration USER_AGENT_OVERRIDE =
+        AwsRequestOverrideConfiguration.builder()
+                                       .addApiName(API_NAME)
+                                       .build();
+
+    private MapperUserAgent() {
+    }
+
+    /**
+     * Adds the mapper's {@code sdk-metrics} API name to a request builder's override configuration, before it is
+     * built.
+     * <p>
+     * Any override configuration already on the builder is kept, so this is safe on requests the mapper did not build
+     * (for example, a caller-built request passed to a public paginated list). The cases are:
+     * <ul>
+     *     <li>No override configuration: the shared {@code USER_AGENT_OVERRIDE} is set, so the common path copies
+     *     nothing.</li>
+     *     <li>The override configuration already has the API name: the builder is left unchanged. This keeps requests
+     *     derived page by page from accumulating one API name per page.</li>
+     *     <li>Otherwise: the API name is added to a copy of the existing override configuration.</li>
+     * </ul>
+     * Requests later derived from the built request with {@code toBuilder()} keep the tag.
+     *
+     * @param builder the request builder to tag
+     * @param <T> the request builder type
+     * @return the same builder, for method chaining
+     */
+    public static <T extends DynamoDbRequest.Builder> T applyUserAgent(T builder) {
+        AwsRequestOverrideConfiguration existing = builder.overrideConfiguration();
+        if (existing == null) {
+            builder.overrideConfiguration(USER_AGENT_OVERRIDE);
+        } else if (!existing.apiNames().contains(API_NAME)) {
+            builder.overrideConfiguration(existing.toBuilder().addApiName(API_NAME).build());
+        }
+        return builder;
+    }
+}