apache · darenwkt · Apr 29, 2025
diff --git a/flink-connector-aws-e2e-tests/flink-connector-dynamodb-e2e-tests/pom.xml b/flink-connector-aws-e2e-tests/flink-connector-dynamodb-e2e-tests/pom.xml
@@ -0,0 +1,123 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one
+  ~ or more contributor license agreements.  See the NOTICE file
+  ~ distributed with this work for additional information
+  ~ regarding copyright ownership.  The ASF licenses this file
+  ~ to you 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.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+    <parent>
+        <artifactId>flink-connector-aws-e2e-tests-parent</artifactId>
+        <groupId>org.apache.flink</groupId>
+        <version>5.1-SNAPSHOT</version>
+    </parent>
+
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>flink-connector-dynamodb-e2e-tests</artifactId>
+    <name>Flink : Connectors : AWS : E2E Tests : Amazon DynamoDB</name>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-api-java</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-runtime</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-planner-loader</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-connector-dynamodb</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-connector-aws-base</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-sql-connector-dynamodb</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <!-- Other third-party dependencies -->
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.datatype</groupId>
+            <artifactId>jackson-datatype-jsr310</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>s3</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>dynamodb</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>dynamodb-enhanced</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>
diff --git a/...tor/dynamodb/sink/DynamoDbSinkITCase.java → ...ynamodb/sink/test/DynamoDbSinkITCase.java b/...tor/dynamodb/sink/DynamoDbSinkITCase.java → ...ynamodb/sink/test/DynamoDbSinkITCase.java
@@ -16,14 +16,19 @@
  * limitations under the License.
  */
 
-package org.apache.flink.connector.dynamodb.sink;
+package org.apache.flink.connector.dynamodb.sink.test;
 
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.dynamodb.sink.DynamoDbSink;
+import org.apache.flink.connector.dynamodb.sink.DynamoDbWriteRequest;
+import org.apache.flink.connector.dynamodb.sink.DynamoDbWriteRequestType;
+import org.apache.flink.connector.dynamodb.testutils.DockerImageVersions;
 import org.apache.flink.connector.dynamodb.testutils.DynamoDBHelpers;
 import org.apache.flink.connector.dynamodb.testutils.DynamoDbContainer;
 import org.apache.flink.connector.dynamodb.testutils.Item;
 import org.apache.flink.connector.dynamodb.testutils.Items;
-import org.apache.flink.connector.dynamodb.util.DockerImageVersions;
 import org.apache.flink.runtime.client.JobExecutionException;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -347,4 +352,17 @@ public Scenario withClientProperties(Properties properties) {
             return this;
         }
     }
+
+    private static class TestDynamoDbElementConverter
+            implements ElementConverter<Map<String, AttributeValue>, DynamoDbWriteRequest> {
+
+        @Override
+        public DynamoDbWriteRequest apply(
+                Map<String, AttributeValue> elements, SinkWriter.Context context) {
+            return DynamoDbWriteRequest.builder()
+                    .setType(DynamoDbWriteRequestType.PUT)
+                    .setItem(elements)
+                    .build();
+        }
+    }
 }
diff --git a/...ctor/dynamodb/sink/TestRequestMapper.java → ...dynamodb/sink/test/TestRequestMapper.java b/...ctor/dynamodb/sink/TestRequestMapper.java → ...dynamodb/sink/test/TestRequestMapper.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.flink.connector.dynamodb.sink;
+package org.apache.flink.connector.dynamodb.sink.test;
 
 import org.apache.flink.api.common.functions.RichMapFunction;
 

diff --git a/...modb/table/DynamoDbDynamicSinkITCase.java → ...table/test/DynamoDbDynamicSinkITCase.java b/...modb/table/DynamoDbDynamicSinkITCase.java → ...table/test/DynamoDbDynamicSinkITCase.java
@@ -16,12 +16,13 @@
  * limitations under the License.
  */
 
-package org.apache.flink.connector.dynamodb.table;
+package org.apache.flink.connector.dynamodb.table.test;
 
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.connector.dynamodb.table.DynamoDbDynamicSink;
+import org.apache.flink.connector.dynamodb.testutils.DockerImageVersions;
 import org.apache.flink.connector.dynamodb.testutils.DynamoDBHelpers;
 import org.apache.flink.connector.dynamodb.testutils.DynamoDbContainer;
-import org.apache.flink.connector.dynamodb.util.DockerImageVersions;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.table.api.EnvironmentSettings;
 import org.apache.flink.table.api.Table;
@@ -49,7 +50,7 @@
 import java.util.UUID;
 import java.util.concurrent.ExecutionException;
 
-/** Integration test for {@link org.apache.flink.connector.dynamodb.table.DynamoDbDynamicSink}. */
+/** Integration test for {@link DynamoDbDynamicSink}. */
 @Testcontainers
 @ExtendWith(MiniClusterExtension.class)
 public class DynamoDbDynamicSinkITCase {

diff --git a/...or/dynamodb/util/DockerImageVersions.java → ...namodb/testutils/DockerImageVersions.java b/...or/dynamodb/util/DockerImageVersions.java → ...namodb/testutils/DockerImageVersions.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.flink.connector.dynamodb.util;
+package org.apache.flink.connector.dynamodb.testutils;
 
 /**
  * Utility class for defining the image names and versions of Docker containers used during the Java

diff --git a/...r/dynamodb/testutils/DynamoDBHelpers.java → ...r/dynamodb/testutils/DynamoDBHelpers.java b/...r/dynamodb/testutils/DynamoDBHelpers.java → ...r/dynamodb/testutils/DynamoDBHelpers.java
diff --git a/...dynamodb/testutils/DynamoDbContainer.java → ...dynamodb/testutils/DynamoDbContainer.java b/...dynamodb/testutils/DynamoDbContainer.java → ...dynamodb/testutils/DynamoDbContainer.java
diff --git a/...nk/connector/dynamodb/testutils/Item.java → ...nk/connector/dynamodb/testutils/Item.java b/...nk/connector/dynamodb/testutils/Item.java → ...nk/connector/dynamodb/testutils/Item.java
@@ -26,15 +26,15 @@
 /** DynamoDB item container. */
 public class Item {
 
-    public static Item.ItemBuilder builder() {
-        return new Item.ItemBuilder();
+    public static ItemBuilder builder() {
+        return new ItemBuilder();
     }
 
     /** Builder to constrict DynamoDB item. */
     public static final class ItemBuilder {
         Map<String, AttributeValue> item = new HashMap<>();
 
-        public Item.ItemBuilder attr(String name, String value) {
+        public ItemBuilder attr(String name, String value) {
             item.put(name, AttributeValue.builder().s(value).build());
             return this;
         }

diff --git a/...k/connector/dynamodb/testutils/Items.java → ...k/connector/dynamodb/testutils/Items.java b/...k/connector/dynamodb/testutils/Items.java → ...k/connector/dynamodb/testutils/Items.java
@@ -28,7 +28,7 @@
 public class Items {
 
     public static ItemsBuilder builder() {
-        return new Items.ItemsBuilder();
+        return new ItemsBuilder();
     }
 
     /** Builder to constrict DynamoDB items. */

diff --git a/...amodb/src/test/resources/create-table.sql → ...tests/src/test/resources/create-table.sql b/...amodb/src/test/resources/create-table.sql → ...tests/src/test/resources/create-table.sql
diff --git a/...r-dynamodb/src/test/resources/datagen.sql → ...-e2e-tests/src/test/resources/datagen.sql b/...r-dynamodb/src/test/resources/datagen.sql → ...-e2e-tests/src/test/resources/datagen.sql
diff --git a/...ws-e2e-tests/flink-connector-dynamodb-e2e-tests/src/test/resources/log4j2-test.properties b/...ws-e2e-tests/flink-connector-dynamodb-e2e-tests/src/test/resources/log4j2-test.properties
@@ -0,0 +1,28 @@
+################################################################################
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you 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.
+################################################################################
+
+# Set root logger level to OFF to not flood build logs
+# set manually to INFO for debugging purposes
+rootLogger.level = OFF
+rootLogger.appenderRef.test.ref = TestLogger
+
+appender.testlogger.name = TestLogger
+appender.testlogger.type = CONSOLE
+appender.testlogger.target = SYSTEM_ERR
+appender.testlogger.layout.type = PatternLayout
+appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
diff --git a/flink-connector-aws-e2e-tests/pom.xml b/flink-connector-aws-e2e-tests/pom.xml
@@ -42,6 +42,7 @@ under the License.
         <module>flink-connector-aws-kinesis-streams-e2e-tests</module>
         <module>flink-connector-kinesis-e2e-tests</module>
         <module>flink-connector-aws-sqs-e2e-tests</module>
+        <module>flink-connector-dynamodb-e2e-tests</module>
         <module>flink-formats-avro-glue-schema-registry-e2e-tests</module>
         <module>flink-formats-json-glue-schema-registry-e2e-tests</module>
     </modules>