[Improve][DynamoDb] Dynamodb client supports endpointOverride and create credentials from container or system env.

docs/en/connector-v2/sink/AmazonDynamoDB.md

@@ -14,34 +14,38 @@ Write data to Amazon DynamoDB
 
 |       Name        |  Type  | Required | Default value |
 |-------------------|--------|----------|---------------|
-| url               | string | yes      | -             |
-| region            | string | yes      | -             |
-| access_key_id     | string | yes      | -             |
-| secret_access_key | string | yes      | -             |
+| url               | string | no       | -             |
+| region            | string | no       | -             |
+| access_key_id     | string | no       | -             |
+| secret_access_key | string | no       | -             |
 | table             | string | yes      | -             |
 | batch_size        | string | no       | 25            |
 | common-options    |        | no       | -             |
 
 ### url [string]
 
-The URL to write to Amazon DynamoDB.
+The URL to write to Amazon DynamoDB. It will override the `endpoint` of AWS DynamoDB.
 
 ### region [string]
 
-The region of Amazon DynamoDB.
+The region of Amazon DynamoDB. The DynamoDB client will use the region to determine the service endpoint.
 
-### accessKeyId [string]
+### access_key_id [string]
 
-The access id of Amazon DynamoDB.
+The access id of Amazon DynamoDB. If you don't set it, the plugin will use the container credential provider chain to get the access id.
 
-### secretAccessKey [string]
+### secret_access_key [string]
 
-The access secret of Amazon DynamoDB.
+The access secret of Amazon DynamoDB. If you don't set it, the plugin will use the container credential provider chain to get the access secret.
 
 ### table [string]
 
 The table of Amazon DynamoDB.
 
+### batch_size [string]
+
+The number of records to write to Amazon DynamoDB in a batch. The default value is 25, and the maximum value is 25 because the Amazon DynamoDB batch write item limit is 25.
+
 ### common options
 
 Sink plugin common parameters, please refer to [Sink Common Options](../sink-common-options.md) for details.
@@ -52,8 +56,8 @@ Sink plugin common parameters, please refer to [Sink Common Options](../sink-com
 Amazondynamodb {
     url = "http://127.0.0.1:8000"
     region = "us-east-1"
-    accessKeyId = "dummy-key"
-    secretAccessKey = "dummy-secret"
+    access_key_id = "dummy-key"
+    secret_access_key = "dummy-secret"
     table = "TableName"
   }
 ```

docs/en/connector-v2/source/AmazonDynamoDB.md

@@ -19,10 +19,10 @@ Read data from Amazon DynamoDB.
 
 |         name          |  type  | required | default value |
 |-----------------------|--------|----------|---------------|
-| url                   | string | yes      | -             |
-| region                | string | yes      | -             |
-| access_key_id         | string | yes      | -             |
-| secret_access_key     | string | yes      | -             |
+| url                   | string | no       | -             |
+| region                | string | no       | -             |
+| access_key_id         | string | no       | -             |
+| secret_access_key     | string | no       | -             |
 | table                 | string | yes      | -             |
 | schema                | config | yes      | -             |
 | common-options        |        | yes      | -             |
@@ -31,19 +31,19 @@ Read data from Amazon DynamoDB.
 
 ### url [string]
 
-The URL to read to Amazon Dynamodb.
+The URL to read to Amazon Dynamodb. It will override the `endpoint` of AWS DynamoDB. 
 
 ### region [string]
 
-The region of Amazon Dynamodb.
+The region of Amazon Dynamodb. The DynamoDB client will use the region to determine the service endpoint.
 
-### accessKeyId [string]
+### access_key_id [string]
 
-The access id of Amazon DynamoDB.
+The access id of Amazon DynamoDB. If you don't set it, the plugin will use the container credential provider chain to get the access id.
 
-### secretAccessKey [string]
+### secret_access_key [string]
 
-The access secret of Amazon DynamoDB.
+The access secret of Amazon DynamoDB. If you don't set it, the plugin will use the container credential provider chain to get the access secret.
 
 ### table [string]
 
@@ -53,7 +53,7 @@ The table of Amazon DynamoDB.
 
 #### fields [config]
 
-Amazon Dynamodb is a NOSQL database service of support keys-value storage and document data structure,there is no way to get the data type.Therefore, we must configure schema.
+Amazon Dynamodb is a NOSQL database service of support keys-value storage and document data structure, there is no way to get the data type. Therefore, we must configure schema.
 
 such as:
 
@@ -85,8 +85,8 @@ number of logical segments for parallel scan
 Amazondynamodb {
   url = "http://127.0.0.1:8000"
   region = "us-east-1"
-  accessKeyId = "dummy-key"
-  secretAccessKey = "dummy-secret"
+  access_key_id = "dummy-key"
+  secret_access_key = "dummy-secret"
   table = "TableName"
   schema = {
     fields {

docs/zh/connector-v2/sink/AmazonDynamoDB.md

@@ -14,34 +14,38 @@
 
 |       名称        |  类型  | 必需 | 默认值 |
 |-------------------|--------|----|---------------|
-| url               | string | 是  | -             |
-| region            | string | 是  | -             |
-| access_key_id     | string | 是  | -             |
-| secret_access_key | string | 是  | -             |
+| url               | string | 否  | -             |
+| region            | string | 否  | -             |
+| access_key_id     | string | 否  | -             |
+| secret_access_key | string | 否  | -             |
 | table             | string | 是  | -             |
 | batch_size        | string | 否  | 25            |
-| common-options    |        | 否 | -             |
+| common-options    |        | 否  | -             |
 
 ### url [string]
 
-要写入Amazon DynamoDB的URL.
+要写入Amazon DynamoDB的URL. 它将覆盖AWS DynamoDB的`endpoint`.
 
 ### region [string]
 
-Amazon DynamoDB 的分区.
+Amazon DynamoDB 的分区. DynamoDB客户端将使用该区域确定服务端点. 
 
-### accessKeyId [string]
+### access_key_id [string]
 
-Amazon DynamoDB的访问id.
+Amazon DynamoDB的访问id. 如果未设置，插件将使用容器凭证提供程序链获取访问id.
 
-### secretAccessKey [string]
+### secret_access_key [string]
 
-Amazon DynamoDB的访问密钥.
+Amazon DynamoDB的访问密钥. 如果未设置，插件将使用容器凭证提供程序链获取访问密钥.
 
 ### table [string]
 
 Amazon DynamoDB 的表名.
 
+### batch_size [string]
+
+在批处理中写入Amazon DynamoDB的记录数. 默认值为25，最大值为25，因为Amazon DynamoDB批处理写入项限制为25.
+
 ### 常见选项
 
 Sink插件常用参数，请参考 [Sink Common Options](../sink-common-options.md) 了解详细信息.

docs/zh/connector-v2/source/AmazonDynamoDB.md

@@ -17,33 +17,33 @@
 
 ## 选项
 
-|         名称        |  类型  | 必需    | 默认值 |
-|-----------------------|--------|-------|---------------|
-| url                   | string | 是     | -             |
-| region                | string | 是     | -             |
-| access_key_id         | string | 是     | -             |
-| secret_access_key     | string | 是     | -             |
-| table                 | string | 是     | -             |
-| schema                | config | 是     | -             |
-| common-options        |        | 是     | -             |
-| scan_item_limit       |        | 否     | -             |
-| parallel_scan_threads |        | 否 | -             |
+|         名称        |  类型  | 必需 | 默认值 |
+|-----------------------|--------|----|---------------|
+| url                   | string | 否  | -             |
+| region                | string | 否  | -             |
+| access_key_id         | string | 否  | -             |
+| secret_access_key     | string | 否  | -             |
+| table                 | string | 是  | -             |
+| schema                | config | 是  | -             |
+| common-options        |        | 是  | -             |
+| scan_item_limit       |        | 否  | -             |
+| parallel_scan_threads |        | 否  | -             |
 
 ### url [string]
 
-读取Amazon Dynamodb的URL.
+读取Amazon Dynamodb的URL. 它将覆盖AWS DynamoDB的`endpoint`.
 
 ### region [string]
 
-Amazon DynamoDB 的分区.
+Amazon DynamoDB 的分区. DynamoDB客户端将使用该区域确定服务端点.
 
-### accessKeyId [string]
+### access_key_id [string]
 
-Amazon DynamoDB的访问id.
+Amazon DynamoDB的访问id. 如果未设置，插件将使用容器凭证提供程序链获取访问id.
 
-### secretAccessKey [string]
+### secret_access_key [string]
 
-Amazon DynamoDB的访问密钥.
+Amazon DynamoDB的访问密钥. 如果未设置，插件将使用容器凭证提供程序链获取访问密钥.
 
 ### table [string]
 

seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/client/DynamoDbClientProvider.java

@@ -0,0 +1,56 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.amazondynamodb.client;
+
+import org.apache.seatunnel.connectors.seatunnel.amazondynamodb.config.AmazonDynamoDBConfig;
+
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.ContainerCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
+import software.amazon.awssdk.services.dynamodb.DynamoDbClientBuilder;
+
+import java.net.URI;
+
+public class DynamoDbClientProvider {
+
+    public static DynamoDbClient createDynamoDBClient(AmazonDynamoDBConfig config) {
+        DynamoDbClientBuilder builder = DynamoDbClient.builder();
+
+        if (config.getUrl() != null) {
+            builder.endpointOverride(URI.create(config.getUrl()));
+        }
+
+        if (config.getRegion() != null) {
+            builder.region(Region.of(config.getRegion()));
+        }
+
+        if (config.getAccessKeyId() != null && config.getSecretAccessKey() != null) {
+            // Set up AWS credentials with accessKeyId and secretAccessKey
+            builder.credentialsProvider(
+                    StaticCredentialsProvider.create(
+                            AwsBasicCredentials.create(
+                                    config.getAccessKeyId(), config.getSecretAccessKey())));
+        } else {
+            // If accessKeyId and secretAccessKey are not provided, use container credentials.
+            builder.credentialsProvider(ContainerCredentialsProvider.builder().build());
+        }
+        return builder.build();
+    }
+}

seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/sink/AmazonDynamoDBSinkFactory.java

@@ -43,8 +43,8 @@ public String factoryIdentifier() {
     @Override
     public OptionRule optionRule() {
         return OptionRule.builder()
-                .required(URL, REGION, ACCESS_KEY_ID, SECRET_ACCESS_KEY, TABLE)
-                .optional(BATCH_SIZE)
+                .required(TABLE)
+                .optional(BATCH_SIZE, ACCESS_KEY_ID, SECRET_ACCESS_KEY, URL, REGION)
                 .build();
     }
 

seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/sink/DynamoDbSinkClient.java

@@ -17,18 +17,15 @@
 
 package org.apache.seatunnel.connectors.seatunnel.amazondynamodb.sink;
 
+import org.apache.seatunnel.connectors.seatunnel.amazondynamodb.client.DynamoDbClientProvider;
 import org.apache.seatunnel.connectors.seatunnel.amazondynamodb.config.AmazonDynamoDBConfig;
 
-import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
-import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
-import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
 import software.amazon.awssdk.services.dynamodb.model.BatchWriteItemRequest;
 import software.amazon.awssdk.services.dynamodb.model.PutItemRequest;
 import software.amazon.awssdk.services.dynamodb.model.PutRequest;
 import software.amazon.awssdk.services.dynamodb.model.WriteRequest;
 
-import java.net.URI;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -49,18 +46,7 @@ private void tryInit() {
         if (initialize) {
             return;
         }
-        dynamoDbClient =
-                DynamoDbClient.builder()
-                        .endpointOverride(URI.create(amazondynamodbConfig.getUrl()))
-                        // The region is meaningless for local DynamoDb but required for client
-                        // builder validation
-                        .region(Region.of(amazondynamodbConfig.getRegion()))
-                        .credentialsProvider(
-                                StaticCredentialsProvider.create(
-                                        AwsBasicCredentials.create(
-                                                amazondynamodbConfig.getAccessKeyId(),
-                                                amazondynamodbConfig.getSecretAccessKey())))
-                        .build();
+        dynamoDbClient = DynamoDbClientProvider.createDynamoDBClient(amazondynamodbConfig);
         initialize = true;
     }
 

seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceFactory.java

@@ -50,8 +50,14 @@ public String factoryIdentifier() {
     @Override
     public OptionRule optionRule() {
         return OptionRule.builder()
-                .required(URL, REGION, ACCESS_KEY_ID, SECRET_ACCESS_KEY, TABLE, SCHEMA)
-                .optional(SCAN_ITEM_LIMIT, PARALLEL_SCAN_THREADS)
+                .required(TABLE, SCHEMA)
+                .optional(
+                        SCAN_ITEM_LIMIT,
+                        PARALLEL_SCAN_THREADS,
+                        ACCESS_KEY_ID,
+                        SECRET_ACCESS_KEY,
+                        URL,
+                        REGION)
                 .build();
     }
 

seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceReader.java

@@ -21,22 +21,20 @@
 import org.apache.seatunnel.api.source.SourceReader;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.connectors.seatunnel.amazondynamodb.client.DynamoDbClientProvider;
 import org.apache.seatunnel.connectors.seatunnel.amazondynamodb.config.AmazonDynamoDBConfig;
 import org.apache.seatunnel.connectors.seatunnel.amazondynamodb.serialize.DefaultSeaTunnelRowDeserializer;
 import org.apache.seatunnel.connectors.seatunnel.amazondynamodb.serialize.SeaTunnelRowDeserializer;
 
 import lombok.extern.slf4j.Slf4j;
-import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
-import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
-import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
 import software.amazon.awssdk.services.dynamodb.model.ScanRequest;
 import software.amazon.awssdk.services.dynamodb.paginators.ScanIterable;
 
-import java.net.URI;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedDeque;
 
@@ -63,23 +61,12 @@ public AmazonDynamoDBSourceReader(
 
     @Override
     public void open() {
-        dynamoDbClient =
-                DynamoDbClient.builder()
-                        .endpointOverride(URI.create(amazondynamodbConfig.getUrl()))
-                        // The region is meaningless for local DynamoDb but required for client
-                        // builder validation
-                        .region(Region.of(amazondynamodbConfig.getRegion()))
-                        .credentialsProvider(
-                                StaticCredentialsProvider.create(
-                                        AwsBasicCredentials.create(
-                                                amazondynamodbConfig.getAccessKeyId(),
-                                                amazondynamodbConfig.getSecretAccessKey())))
-                        .build();
+        dynamoDbClient = DynamoDbClientProvider.createDynamoDBClient(amazondynamodbConfig);
     }
 
     @Override
     public void close() {
-        dynamoDbClient.close();
+        Optional.ofNullable(dynamoDbClient).ifPresent(DynamoDbClient::close);
     }
 
     @Override