[Improve] Code clean for AmazonDynamoDB connector (#5791)

Hisoka-X · web-flow · commit a17dd7afc180 · 2023-11-07T13:34:03.000+08:00
diff --git a/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/exception/AmazonDynamoDBConnectorException.java b/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/exception/AmazonDynamoDBConnectorException.java
@@ -25,14 +25,4 @@ public AmazonDynamoDBConnectorException(
             SeaTunnelErrorCode seaTunnelErrorCode, String errorMessage) {
         super(seaTunnelErrorCode, errorMessage);
     }
-
-    public AmazonDynamoDBConnectorException(
-            SeaTunnelErrorCode seaTunnelErrorCode, String errorMessage, Throwable cause) {
-        super(seaTunnelErrorCode, errorMessage, cause);
-    }
-
-    public AmazonDynamoDBConnectorException(
-            SeaTunnelErrorCode seaTunnelErrorCode, Throwable cause) {
-        super(seaTunnelErrorCode, cause);
-    }
 }
diff --git a/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/sink/AmazonDynamoDBWriter.java b/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/sink/AmazonDynamoDBWriter.java
@@ -35,7 +35,7 @@ public class AmazonDynamoDBWriter extends AbstractSinkWriter<SeaTunnelRow, Void>
     public AmazonDynamoDBWriter(
             AmazonDynamoDBSourceOptions amazondynamodbSourceOptions,
             SeaTunnelRowType seaTunnelRowType) {
-        dynamoDbSinkClient = new DynamoDbSinkClient(amazondynamodbSourceOptions, seaTunnelRowType);
+        dynamoDbSinkClient = new DynamoDbSinkClient(amazondynamodbSourceOptions);
         serializer =
                 new DefaultSeaTunnelRowSerializer(seaTunnelRowType, amazondynamodbSourceOptions);
     }
diff --git a/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/sink/DynamoDbSinkClient.java b/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/sink/DynamoDbSinkClient.java
@@ -17,12 +17,7 @@
 
 package org.apache.seatunnel.connectors.seatunnel.amazondynamodb.sink;
 
-import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
-import org.apache.seatunnel.common.exception.CommonErrorCode;
 import org.apache.seatunnel.connectors.seatunnel.amazondynamodb.config.AmazonDynamoDBSourceOptions;
-import org.apache.seatunnel.connectors.seatunnel.amazondynamodb.exception.AmazonDynamoDBConnectorException;
-import org.apache.seatunnel.connectors.seatunnel.amazondynamodb.serialize.DefaultSeaTunnelRowDeserializer;
-import org.apache.seatunnel.connectors.seatunnel.amazondynamodb.serialize.SeaTunnelRowDeserializer;
 
 import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
 import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
@@ -33,7 +28,6 @@
 import software.amazon.awssdk.services.dynamodb.model.PutRequest;
 import software.amazon.awssdk.services.dynamodb.model.WriteRequest;
 
-import java.io.IOException;
 import java.net.URI;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -43,16 +37,12 @@
 public class DynamoDbSinkClient {
     private final AmazonDynamoDBSourceOptions amazondynamodbSourceOptions;
     private volatile boolean initialize;
-    private volatile Exception flushException;
     private DynamoDbClient dynamoDbClient;
     private final List<WriteRequest> batchList;
-    protected SeaTunnelRowDeserializer seaTunnelRowDeserializer;
 
-    public DynamoDbSinkClient(
-            AmazonDynamoDBSourceOptions amazondynamodbSourceOptions, SeaTunnelRowType typeInfo) {
+    public DynamoDbSinkClient(AmazonDynamoDBSourceOptions amazondynamodbSourceOptions) {
         this.amazondynamodbSourceOptions = amazondynamodbSourceOptions;
         this.batchList = new ArrayList<>();
-        this.seaTunnelRowDeserializer = new DefaultSeaTunnelRowDeserializer(typeInfo);
     }
 
     private void tryInit() {
@@ -74,9 +64,8 @@ private void tryInit() {
         initialize = true;
     }
 
-    public synchronized void write(PutItemRequest putItemRequest) throws IOException {
+    public synchronized void write(PutItemRequest putItemRequest) {
         tryInit();
-        checkFlushException();
         batchList.add(
                 WriteRequest.builder()
                         .putRequest(PutRequest.builder().item(putItemRequest.item()).build())
@@ -87,15 +76,14 @@ public synchronized void write(PutItemRequest putItemRequest) throws IOException
         }
     }
 
-    public synchronized void close() throws IOException {
+    public synchronized void close() {
         if (dynamoDbClient != null) {
             flush();
             dynamoDbClient.close();
         }
     }
 
     synchronized void flush() {
-        checkFlushException();
         if (batchList.isEmpty()) {
             return;
         }
@@ -106,13 +94,4 @@ synchronized void flush() {
 
         batchList.clear();
     }
-
-    private void checkFlushException() {
-        if (flushException != null) {
-            throw new AmazonDynamoDBConnectorException(
-                    CommonErrorCode.FLUSH_DATA_FAILED,
-                    "Flush data to AmazonDynamoDB failed.",
-                    flushException);
-        }
-    }
 }
diff --git a/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceReader.java b/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceReader.java
@@ -30,15 +30,12 @@
 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.AttributeValue;
 import software.amazon.awssdk.services.dynamodb.model.ScanRequest;
 import software.amazon.awssdk.services.dynamodb.paginators.ScanIterable;
 
-import java.io.IOException;
 import java.net.URI;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedDeque;
 
@@ -64,7 +61,7 @@ public AmazonDynamoDBSourceReader(
     }
 
     @Override
-    public void open() throws Exception {
+    public void open() {
         dynamoDbClient =
                 DynamoDbClient.builder()
                         .endpointOverride(URI.create(amazondynamodbSourceOptions.getUrl()))
@@ -80,27 +77,26 @@ public void open() throws Exception {
     }
 
     @Override
-    public void close() throws IOException {
+    public void close() {
         dynamoDbClient.close();
     }
 
     @Override
     @SuppressWarnings("magicnumber")
-    public void pollNext(Collector<SeaTunnelRow> output) throws Exception {
+    public void pollNext(Collector<SeaTunnelRow> output) {
         while (!pendingSplits.isEmpty()) {
             synchronized (output.getCheckpointLock()) {
                 AmazonDynamoDBSourceSplit split = pendingSplits.poll();
-
                 read(split, output);
             }
         }
-        if (pendingSplits.isEmpty() && noMoreSplit) {
+        if (noMoreSplit && pendingSplits.isEmpty()) {
             context.signalNoMoreElement();
         }
     }
 
     @Override
-    public List<AmazonDynamoDBSourceSplit> snapshotState(long checkpointId) throws Exception {
+    public List<AmazonDynamoDBSourceSplit> snapshotState(long checkpointId) {
         return new ArrayList<>(pendingSplits);
     }
 
@@ -115,9 +111,7 @@ public void handleNoMoreSplits() {
         noMoreSplit = true;
     }
 
-    private void read(AmazonDynamoDBSourceSplit split, Collector<SeaTunnelRow> output)
-            throws Exception {
-        Map<String, AttributeValue> lastKeyEvaluated = null;
+    private void read(AmazonDynamoDBSourceSplit split, Collector<SeaTunnelRow> output) {
         ScanIterable scan;
         ScanRequest scanRequest =
                 ScanRequest.builder()
@@ -145,5 +139,5 @@ private void read(AmazonDynamoDBSourceSplit split, Collector<SeaTunnelRow> outpu
     }
 
     @Override
-    public void notifyCheckpointComplete(long checkpointId) throws Exception {}
+    public void notifyCheckpointComplete(long checkpointId) {}
 }

Original file line number	Diff line number	Diff line change
`@@ -35,7 +35,7 @@ public class AmazonDynamoDBWriter extends AbstractSinkWriter<SeaTunnelRow, Void>`
`35`	`35`	`public AmazonDynamoDBWriter(`
`36`	`36`	`AmazonDynamoDBSourceOptions amazondynamodbSourceOptions,`
`37`	`37`	`SeaTunnelRowType seaTunnelRowType) {`
`38`		`- dynamoDbSinkClient = new DynamoDbSinkClient(amazondynamodbSourceOptions, seaTunnelRowType);`
	`38`	`+ dynamoDbSinkClient = new DynamoDbSinkClient(amazondynamodbSourceOptions);`
`39`	`39`	`serializer =`
`40`	`40`	`new DefaultSeaTunnelRowSerializer(seaTunnelRowType, amazondynamodbSourceOptions);`
`41`	`41`	`}`