Adding overflow protection

Author: RanVaknin

services-custom/dynamodb-enhanced/src/main/java/software/amazon/awssdk/enhanced/dynamodb/extensions/VersionedRecordExtension.java

@@ -175,6 +175,17 @@ public WriteModification beforeWrite(DynamoDbExtensionContext.BeforeWrite contex
             String existingVersionValueKey = VERSIONED_RECORD_EXPRESSION_VALUE_KEY_MAPPER.apply(versionAttributeKey.get());
 
             long increment = versionIncrementByFromAnnotation;
+
+            /*
+            since the new incrementBy and StartAt functionality can now accept any positive number, though unlikely
+            to happen in a rela life scenario, we should add overflow protection.
+            */
+            if (existingVersion > Long.MAX_VALUE - increment) {
+                throw new IllegalStateException(
+                    String.format("Version overflow detected. Current version %d + increment %d would exceed Long.MAX_VALUE",
+                                  existingVersion, increment));
+            }
+
             newVersionValue = AttributeValue.builder().n(Long.toString(existingVersion + increment)).build();
 
             condition = Expression.builder()

services-custom/dynamodb-enhanced/src/test/java/software/amazon/awssdk/enhanced/dynamodb/extensions/VersionedRecordExtensionTest.java

@@ -582,6 +582,30 @@ public void customStartingValueAndIncrementWithImmutableClass_worksAsExpected()
                                 .build()));
     }
 
+    @Test(expected = IllegalStateException.class)
+    public void versionOverflowThrowsException() {
+        VersionedRecordExtension recordExtension = VersionedRecordExtension.builder()
+                                                                           .incrementBy(2L)
+                                                                           .build();
+
+        FakeVersionedThroughAnnotationItem item = new FakeVersionedThroughAnnotationItem();
+        item.setId(UUID.randomUUID().toString());
+        item.setVersion(Long.MAX_VALUE - 1);
+
+        TableSchema<FakeVersionedThroughAnnotationItem> schema =
+            TableSchema.fromBean(FakeVersionedThroughAnnotationItem.class);
+
+        Map<String, AttributeValue> inputMap =
+            new HashMap<>(schema.itemToMap(item, true));
+
+        recordExtension.beforeWrite(DefaultDynamoDbExtensionContext
+                                        .builder()
+                                        .items(inputMap)
+                                        .tableMetadata(schema.tableMetadata())
+                                        .operationContext(PRIMARY_CONTEXT)
+                                        .build());
+    }
+
     public static Stream<Arguments> customIncrementForExistingVersionValues() {
         return Stream.of(
             Arguments.of(0L, 1L, 5L, "6"),