Add StreamSpecification to CreateTableEnhancedRequest (#4011)

* Add `StreamSpecification` to `CreateTableEnhancedRequest`

Allows you to specify a StreamSpecification when creating a DynamoDbTable using the CreateTableOperation with the EnhancedDynamoDbClient

Closes #4010

* Updated the change log and java doc

---------

Co-authored-by: John Viegas <joviegas@amazon.com>
Co-authored-by: John Viegas <70235430+joviegas@users.noreply.github.com>

.changes/next-release/feature-AmazonDynamoDB-c0398be.json

@@ -0,0 +1,6 @@
+{
+    "type": "feature",
+    "category": "DynamoDB Enhanced Client",
+    "contributor": "acouvreur",
+    "description": "Added support for StreamSpecification in the CreateTableEnhancedRequest, allowing configuration of table stream specification using DynamoDB Enhanced Client."
+}

services-custom/dynamodb-enhanced/src/main/java/software/amazon/awssdk/enhanced/dynamodb/internal/operations/CreateTableOperation.java

@@ -140,6 +140,7 @@ public CreateTableRequest generateRequest(TableSchema<T> tableSchema,
                                  .attributeDefinitions(attributeDefinitions)
                                  .billingMode(billingMode)
                                  .provisionedThroughput(this.request.provisionedThroughput())
+                                 .streamSpecification(this.request.streamSpecification())
                                  .build();
     }
 

services-custom/dynamodb-enhanced/src/main/java/software/amazon/awssdk/enhanced/dynamodb/model/CreateTableEnhancedRequest.java

@@ -26,6 +26,7 @@
 import software.amazon.awssdk.enhanced.dynamodb.DynamoDbAsyncTable;
 import software.amazon.awssdk.enhanced.dynamodb.DynamoDbTable;
 import software.amazon.awssdk.services.dynamodb.model.ProvisionedThroughput;
+import software.amazon.awssdk.services.dynamodb.model.StreamSpecification;
 
 /**
  * Defines parameters used to create a DynamoDb table using the createTable() operation (such as
@@ -38,11 +39,13 @@
 @ThreadSafe
 public final class CreateTableEnhancedRequest {
     private final ProvisionedThroughput provisionedThroughput;
+    private final StreamSpecification streamSpecification;
     private final Collection<EnhancedLocalSecondaryIndex> localSecondaryIndices;
     private final Collection<EnhancedGlobalSecondaryIndex> globalSecondaryIndices;
 
     private CreateTableEnhancedRequest(Builder builder) {
         this.provisionedThroughput = builder.provisionedThroughput;
+        this.streamSpecification = builder.streamSpecification;
         this.localSecondaryIndices = builder.localSecondaryIndices;
         this.globalSecondaryIndices = builder.globalSecondaryIndices;
     }
@@ -59,6 +62,7 @@ public static Builder builder() {
      */
     public Builder toBuilder() {
         return builder().provisionedThroughput(provisionedThroughput)
+                        .streamSpecification(streamSpecification)
                         .localSecondaryIndices(localSecondaryIndices)
                         .globalSecondaryIndices(globalSecondaryIndices);
     }
@@ -70,6 +74,13 @@ public ProvisionedThroughput provisionedThroughput() {
         return provisionedThroughput;
     }
 
+    /**
+     * Returns the stream specification value set on this request object, or null if it has not been set.
+     */
+    public StreamSpecification streamSpecification() {
+        return streamSpecification;
+    }
+
     /**
      * Returns the local secondary index set on this request object, or null if it has not been set.
      */
@@ -99,6 +110,10 @@ public boolean equals(Object o) {
             that.provisionedThroughput != null) {
             return false;
         }
+        if (streamSpecification != null ? ! streamSpecification.equals(that.streamSpecification) :
+            that.streamSpecification != null) {
+            return false;
+        }
         if (localSecondaryIndices != null ? ! localSecondaryIndices.equals(that.localSecondaryIndices) :
             that.localSecondaryIndices != null) {
             return false;
@@ -110,6 +125,7 @@ public boolean equals(Object o) {
     @Override
     public int hashCode() {
         int result = provisionedThroughput != null ? provisionedThroughput.hashCode() : 0;
+        result = 31 * result + (streamSpecification != null ? streamSpecification.hashCode() : 0);
         result = 31 * result + (localSecondaryIndices != null ? localSecondaryIndices.hashCode() : 0);
         result = 31 * result + (globalSecondaryIndices != null ? globalSecondaryIndices.hashCode() : 0);
         return result;
@@ -121,6 +137,7 @@ public int hashCode() {
     @NotThreadSafe
     public static final class Builder {
         private ProvisionedThroughput provisionedThroughput;
+        private StreamSpecification streamSpecification;
         private Collection<EnhancedLocalSecondaryIndex> localSecondaryIndices;
         private Collection<EnhancedGlobalSecondaryIndex> globalSecondaryIndices;
 
@@ -149,6 +166,27 @@ public Builder provisionedThroughput(Consumer<ProvisionedThroughput.Builder> pro
             return provisionedThroughput(builder.build());
         }
 
+        /**
+         * Sets the {@link StreamSpecification} for this table.
+         * <p>
+         * See the DynamoDb documentation for more information on stream specification values.
+         */
+        public Builder streamSpecification(StreamSpecification streamSpecification) {
+            this.streamSpecification = streamSpecification;
+            return this;
+        }
+
+        /**
+         * This is a convenience method for {@link #streamSpecification(StreamSpecification)} that creates an instance of the
+         * {@link StreamSpecification.Builder} for you, avoiding the need to create one manually via
+         * {@link StreamSpecification#builder()}.
+         */
+        public Builder streamSpecification(Consumer<StreamSpecification.Builder> streamSpecification) {
+            StreamSpecification.Builder builder = StreamSpecification.builder();
+            streamSpecification.accept(builder);
+            return streamSpecification(builder.build());
+        }
+
         /**
          * Defines a local secondary index for this table.
          * <p>

services-custom/dynamodb-enhanced/src/test/java/software/amazon/awssdk/enhanced/dynamodb/internal/operations/CreateTableOperationTest.java

@@ -19,6 +19,7 @@
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
 import static org.hamcrest.Matchers.sameInstance;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.same;
@@ -57,6 +58,9 @@
 import software.amazon.awssdk.services.dynamodb.model.ProjectionType;
 import software.amazon.awssdk.services.dynamodb.model.ProvisionedThroughput;
 import software.amazon.awssdk.services.dynamodb.model.ScalarAttributeType;
+import software.amazon.awssdk.services.dynamodb.model.StreamSpecification;
+import software.amazon.awssdk.services.dynamodb.model.StreamViewType;
+
 
 @RunWith(MockitoJUnitRunner.class)
 public class CreateTableOperationTest {
@@ -329,6 +333,34 @@ public void generateRequest_withNoProvisionedThroughput() {
         assertThat(request.billingMode(), is(BillingMode.PAY_PER_REQUEST));
     }
 
+    @Test
+    public void generateRequest_withStreamSpecification() {
+        StreamSpecification streamSpecification = StreamSpecification.builder()
+                                                                     .streamEnabled(true)
+                                                                     .streamViewType(StreamViewType.NEW_IMAGE)
+                                                                     .build();
+
+        CreateTableOperation<FakeItem> operation = CreateTableOperation.create(
+            CreateTableEnhancedRequest.builder().streamSpecification(streamSpecification).build());
+
+        CreateTableRequest request = operation.generateRequest(FakeItem.getTableSchema(),
+                                                               PRIMARY_CONTEXT,
+                                                               null);
+
+        assertThat(request.streamSpecification(), is(streamSpecification));
+    }
+
+    @Test
+    public void generateRequest_withNoStreamSpecification() {
+        CreateTableOperation<FakeItem> operation = CreateTableOperation.create(CreateTableEnhancedRequest.builder().build());
+
+        CreateTableRequest request = operation.generateRequest(FakeItem.getTableSchema(),
+                                                               PRIMARY_CONTEXT,
+                                                               null);
+
+        assertThat(request.streamSpecification(), is(nullValue()));
+    }
+
 
     @Test
     public void generateRequest_withNumericKey() {